Prototype MAP and MULTISET JSON batch boundaries

divergences/32-sql-json-definite-paths.md

@@ -247,9 +247,12 @@ as a wildcard. Document validation and the shared Jackson buffer contract stay u
 No JNI or Arrow ownership change is required. Member-name unions and JSON_QUERY need
 separate result-shape contracts and remain outside this admission.
 
-The whole-Calc JVM bridge also carries verified nested ARRAY/ROW boundaries. It follows Comet's
+The whole-Calc JVM bridge also carries verified nested ARRAY/ROW/MAP/MULTISET boundaries. It follows Comet's
 import/evaluate/export lifetime: the imported argument batch owns nested views until generated
 Flink evaluation and synchronous output writing finish; output vectors own the exported result.
 No nested view escapes into the returned native batch, and JNI remains one call per batch. MAP
-and MULTISET boundaries are still outside this admission. This extends host-exact coverage rather
+keys and MULTISET elements must be declared non-null because Arrow maps require non-null keys.
+The planner rejects nullable keys recursively instead of inferring safety from observed values.
+The existing Flink map reader supplies borrowed MapData, and its writer copies nested output
+into owned vectors within the same import/evaluate/export scope. This extends host-exact coverage rather
 than replacing the measured native JSON fast paths.

docs/benchmarks/scalar-functions.md

@@ -1264,3 +1264,39 @@ SF_BENCHMARK=true mvn -pl streamfusion-runtime -am test -Pbench \
   -Dscalar.rows=1000000 -Dscalar.bytes=264 -Dscalar.nullEvery=8 \
   -Dscalar.warmup=2 -Dscalar.runs=5 -Dscalar.engine=both
 ```
+
+## MAP SQL/JSON batch boundaries (2026-09-19)
+
+The generated Calc bridge also carries MAP values whose keys are declared non-null, and
+MULTISET values whose elements are declared non-null. This extends coverage through the same
+borrowed-input/owned-output lifetime; it adds no per-row JNI calls and no native JSON parser.
+
+The existing scalar benchmark measured 2M rows on M1 Max/JDK 17, release + mimalloc,
+parallelism one, 264-byte ASCII payloads and a NULL outer map every seventh row. Each non-null
+map contains three string keys and one NULL value. Both transposes and the rowwise blackhole
+sink remain in the timed path. There were two warmups and five interleaved measured trials;
+medians below include all conversion and callback work.
+
+| Case | Flink (s) | Native pipeline (s) | Flink / native |
+|---|---:|---:|---:|
+| MAP identity control | 1.318 | 2.379 | 0.554x |
+| `JSON_STRING(m)` | 2.727 | 4.179 | 0.652x |
+| `ROW(m, JSON_QUERY(s, '$.items[*]'))` | 2.611 | 5.599 | 0.466x |
+
+These isolated projections are slower than Flink. The purpose is to keep supported MAP-bearing
+JSON Calcs inside a larger native pipeline; this measurement does not establish a speedup for
+that composition. Existing measured native JSON fast paths remain in use. Nullable keys still
+fall back because Arrow's map representation cannot preserve them safely.
+
+[All measured trials](sql-json-map-2026-09-19.csv) are retained. Reproduce with the existing
+`ScalarFunctionBenchmark#individualFunctions` under `-Pbench`, `SF_BENCHMARK=true`,
+`-Dscalar.functions=JSON_STRING_MAP,JSON_QUERY_MAP_RESULT -Dscalar.nullEvery=7`.
+
+A second benchmark put the same JSON map serialization inside a filter followed by grouped
+COUNT over 4,096 keys, retaining both transposes and the rowwise blackhole sink. It used the
+same 2M rows, payload sizes, NULL rate, warmups and interleaving. Flink took **3.365s** and
+the native pipeline **4.774s** (**0.705x**). Thus this measured composition is also slower;
+there is no demonstrated speedup for the MAP extension. The implementation is small because
+it reuses the generated host evaluator, but performance needs further work before recommending
+it as an optimization. [Composition trials](sql-json-map-composition-2026-09-19.csv) are retained;
+reproduce with `JsonMapCompositionBenchmark` under `-Pbench` and `SF_BENCHMARK=true`.

docs/benchmarks/sql-json-map-2026-09-19.csv

@@ -0,0 +1,31 @@
+function,input,output_type,payload_bytes,json_fields,unicode,null_every,rows,engine,trial,seconds
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,flink,0,1.326055
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,native,0,2.398234
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,native,1,2.370823
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,flink,1,1.324978
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,flink,2,1.303962
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,native,2,2.357196
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,native,3,2.378952
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,flink,3,1.318076
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,flink,4,1.303520
+BASELINE_tt_json_map,tt_json_map,"MAP<STRING NOT NULL, STRING>",264,0,false,7,2000000,native,4,2.407260
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,flink,0,2.748544
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,native,0,4.213936
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,native,1,4.173463
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,flink,1,2.733266
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,flink,2,2.726742
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,native,2,4.192609
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,native,3,4.179333
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,flink,3,2.713187
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,flink,4,2.715054
+JSON_STRING_MAP,tt_json_map,"STRING",264,0,false,7,2000000,native,4,4.141088
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,flink,0,2.602456
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,native,0,5.611931
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,native,1,5.587792
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,flink,1,2.641495
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,flink,2,2.601597
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,native,2,5.598744
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,native,3,5.594735
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,flink,3,2.611287
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,flink,4,2.624480
+JSON_QUERY_MAP_RESULT,tt_json_map,"ROW<items MAP<STRING NOT NULL, STRING>, json_result STRING>",264,0,false,7,2000000,native,4,5.611542

docs/benchmarks/sql-json-map-composition-2026-09-19.csv

@@ -0,0 +1,11 @@
+engine,trial,rows,seconds
+flink,0,2000000,3.365431041
+native,0,2000000,4.770736125
+native,1,2000000,4.773890959
+flink,1,2000000,3.361692250
+flink,2,2000000,3.424452125
+native,2,2000000,4.830700667
+native,3,2000000,4.728313458
+flink,3,2000000,3.380699125
+flink,4,2000000,3.358331708
+native,4,2000000,4.781714833

docs/operators/calc-filter.md

@@ -818,12 +818,14 @@ iteration happens inside that callback. Rejected rows and changelog tags share a
 before Arrow validates nonnullable result fields.
 
 Referenced inputs and projected outputs can carry character, binary, boolean, numeric/decimal,
-date/time, interval, supported timestamp types, and recursively nested ARRAY/ROW values with
-those leaves. Nested arguments use Flink internal views over the imported Arrow batch; generated
+date/time, interval, supported timestamp types, and recursively nested ARRAY/ROW/MAP/MULTISET
+values with those leaves. MAP keys and MULTISET elements must be declared non-null, matching the
+Arrow map key representation. Nullable map values, outer containers and nested array/row values
+remain supported. Nested arguments use Flink internal views over the imported Arrow batch; generated
 results are copied into owned Arrow output vectors before that batch closes. The callback still
 crosses JNI once per batch, including multi-column results and filtering.
-MAP/MULTISET boundary values, including maps nested inside an array or row, remain explicit
-fallback. Constructing containers internally is allowed when the resulting boundary types are
+Nullable MAP keys and nullable MULTISET elements remain explicit fallback, including when nested
+inside arrays or rows; observing only non-null keys in a fixture does not establish the type contract. Constructing containers internally is allowed when the resulting boundary types are
 admitted. Unsupported host code generation
 and UDF signatures retain explicit fallback. Flink 2.2.1 rejects dynamic JSON_EXISTS paths;
 that host failure is preserved. No configuration opt-in is required.

src/main/java/tech/streamfusion/operator/NativeUdf.java

@@ -89,7 +89,7 @@ public interface InternalArguments {
   public static final int TYPE_INTERVAL_MILLIS = 13;
   public static final int TYPE_BINARY = 14;
   public static final int TYPE_ROW = 15;
-  // Internal ARRAY/ROW views borrowed only while the imported argument batch remains open.
+  // Internal nested views borrowed only while the imported argument batch remains open.
   public static final int TYPE_INTERNAL = 16;
 
   // DECIMAL(p, s) argument/result values, marshalled as BigDecimal. The precision and scale ride in

src/main/java/tech/streamfusion/planner/RexExpression.java

@@ -459,6 +459,13 @@ private static int rowCalcTypeCode(RelDataType type) {
     boolean nested =
         switch (type.getSqlTypeName()) {
           case ARRAY -> rowCalcTypeCode(type.getComponentType()) >= 0;
+          case MAP ->
+              !type.getKeyType().isNullable()
+                  && rowCalcTypeCode(type.getKeyType()) >= 0
+                  && rowCalcTypeCode(type.getValueType()) >= 0;
+          case MULTISET ->
+              !type.getComponentType().isNullable()
+                  && rowCalcTypeCode(type.getComponentType()) >= 0;
           case ROW ->
               type.getFieldList().stream().allMatch(field -> rowCalcTypeCode(field.getType()) >= 0);
           default -> false;

src/test/java/tech/streamfusion/FlinkJsonJvmSqlHarnessTest.java

@@ -179,30 +179,49 @@ void mixedJsonCalcExecutesAcrossBatchesAndFiltersBeforeEvaluation() throws Excep
                       Row.of(
                           id,
                           id % 3 == 0 ? "invalid" : "{\"a\":[1,2,3],\"n\":" + id + "}",
-                          new String[] {id.toString(), null}))
+                          new String[] {id.toString(), null},
+                          java.util.Map.of("id", new String[] {id.toString(), null})))
               .returns(
                   Types.ROW_NAMED(
-                      new String[] {"id", "s", "a"},
+                      new String[] {"id", "s", "a", "m"},
                       Types.LONG,
                       Types.STRING,
-                      Types.OBJECT_ARRAY(Types.STRING))));
+                      Types.OBJECT_ARRAY(Types.STRING),
+                      Types.MAP(Types.STRING, Types.OBJECT_ARRAY(Types.STRING)))),
+          org.apache.flink.table.api.Schema.newBuilder()
+              .column("id", org.apache.flink.table.api.DataTypes.BIGINT())
+              .column("s", org.apache.flink.table.api.DataTypes.STRING())
+              .column(
+                  "a",
+                  org.apache.flink.table.api.DataTypes.ARRAY(
+                      org.apache.flink.table.api.DataTypes.STRING()))
+              .column(
+                  "m",
+                  org.apache.flink.table.api.DataTypes.MAP(
+                      org.apache.flink.table.api.DataTypes.STRING().notNull(),
+                      org.apache.flink.table.api.DataTypes.ARRAY(
+                          org.apache.flink.table.api.DataTypes.STRING())))
+              .build());
       String sql =
           "SELECT id, JSON_QUERY(s, '$.a[0:2]' ERROR ON ERROR), JSON_VALUE(s, '$.n' ERROR ON"
-              + " ERROR), JSON_STRING(a), a FROM inputs WHERE MOD(id, 3) <> 0";
+              + " ERROR), JSON_STRING(a), a, JSON_STRING(m), m FROM inputs WHERE MOD(id, 3) <> 0";
       var scan = tech.streamfusion.planner.NativePlanner.install(table);
       assertTrue(table.explainSql(sql).contains("jsonEvaluation=[JVM]"));
       var result = table.executeSql(sql);
       try (var rows = result.collect()) {
         for (long id = 0; id < 5003; id++) {
           if (id % 3 != 0)
             assertEquals(
-                Row.of(
-                    id,
-                    "[1,2]",
-                    Long.toString(id),
-                    "[\"" + id + "\",null]",
-                    new String[] {Long.toString(id), null}),
-                rows.next());
+                NativeParity.comparableValue(
+                    Row.of(
+                        id,
+                        "[1,2]",
+                        Long.toString(id),
+                        "[\"" + id + "\",null]",
+                        new String[] {Long.toString(id), null},
+                        "{\"id\":[\"" + id + "\",null]}",
+                        java.util.Map.of("id", new String[] {Long.toString(id), null}))),
+                NativeParity.comparableValue(rows.next()));
         }
         org.junit.jupiter.api.Assertions.assertFalse(rows.hasNext());
       }