From 286b75fd3380a266d7d636030e6b640c7fd850cc Mon Sep 17 00:00:00 2001 From: Jordan Epstein Date: Sat, 19 Sep 2026 05:23:40 -0400 Subject: [PATCH] Prototype non-null-key maps through the generated JSON batch boundary Reuse borrowed Flink MapData inside the existing import/evaluate/export lifetime, recursively admitting declared non-null MAP keys and MULTISET elements while preserving nullable values. Compare nested collection contents in the parity harness instead of Java array identities, without losing changelog kinds. No additional JSON parser or per-row JNI call is introduced. 70 focused SQL, bridge and comparison cases pass, including 5003 input rows across filtered batches; all 32 unchanged upstream Calc cases pass with three actual native execution contracts. Release+mimalloc 2M-row measurements are slower than Flink: 0.652x map serialization, 0.466x nested output and 0.705x JSON-filter/grouped-count composition, including both transposes. Preserve every measured trial and keep this as a draft coverage prototype pending a stronger performance case; do not claim an optimization. Advances investigation for #91. --- divergences/32-sql-json-definite-paths.md | 7 +- docs/benchmarks/scalar-functions.md | 36 ++++ docs/benchmarks/sql-json-map-2026-09-19.csv | 31 ++++ .../sql-json-map-composition-2026-09-19.csv | 11 ++ docs/operators/calc-filter.md | 10 +- .../tech/streamfusion/operator/NativeUdf.java | 2 +- .../streamfusion/planner/RexExpression.java | 7 + .../FlinkJsonJvmSqlHarnessTest.java | 41 +++-- .../FlinkMapJsonJvmSqlHarnessTest.java | 170 ++++++++++++++++++ .../JsonMapCompositionBenchmark.java | 117 ++++++++++++ .../java/tech/streamfusion/NativeParity.java | 13 ++ .../streamfusion/NativeParityValueTest.java | 32 ++++ .../streamfusion/ScalarFunctionBenchmark.java | 4 + .../streamfusion/TextTimeBenchmarkInputs.java | 14 +- 14 files changed, 476 insertions(+), 19 deletions(-) create mode 100644 docs/benchmarks/sql-json-map-2026-09-19.csv create mode 100644 docs/benchmarks/sql-json-map-composition-2026-09-19.csv create mode 100644 src/test/java/tech/streamfusion/FlinkMapJsonJvmSqlHarnessTest.java create mode 100644 src/test/java/tech/streamfusion/JsonMapCompositionBenchmark.java create mode 100644 src/test/java/tech/streamfusion/NativeParityValueTest.java diff --git a/divergences/32-sql-json-definite-paths.md b/divergences/32-sql-json-definite-paths.md index f072c3e7..6a8af21e 100644 --- a/divergences/32-sql-json-definite-paths.md +++ b/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. diff --git a/docs/benchmarks/scalar-functions.md b/docs/benchmarks/scalar-functions.md index d8580787..163418f9 100644 --- a/docs/benchmarks/scalar-functions.md +++ b/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`. diff --git a/docs/benchmarks/sql-json-map-2026-09-19.csv b/docs/benchmarks/sql-json-map-2026-09-19.csv new file mode 100644 index 00000000..f6971c87 --- /dev/null +++ b/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",264,0,false,7,2000000,flink,0,1.326055 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,native,0,2.398234 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,native,1,2.370823 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,flink,1,1.324978 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,flink,2,1.303962 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,native,2,2.357196 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,native,3,2.378952 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,flink,3,1.318076 +BASELINE_tt_json_map,tt_json_map,"MAP",264,0,false,7,2000000,flink,4,1.303520 +BASELINE_tt_json_map,tt_json_map,"MAP",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, json_result STRING>",264,0,false,7,2000000,flink,0,2.602456 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,native,0,5.611931 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,native,1,5.587792 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,flink,1,2.641495 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,flink,2,2.601597 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,native,2,5.598744 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,native,3,5.594735 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,flink,3,2.611287 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,flink,4,2.624480 +JSON_QUERY_MAP_RESULT,tt_json_map,"ROW, json_result STRING>",264,0,false,7,2000000,native,4,5.611542 diff --git a/docs/benchmarks/sql-json-map-composition-2026-09-19.csv b/docs/benchmarks/sql-json-map-composition-2026-09-19.csv new file mode 100644 index 00000000..75b41f44 --- /dev/null +++ b/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 diff --git a/docs/operators/calc-filter.md b/docs/operators/calc-filter.md index ba24c4f6..ac0049af 100644 --- a/docs/operators/calc-filter.md +++ b/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. diff --git a/src/main/java/tech/streamfusion/operator/NativeUdf.java b/src/main/java/tech/streamfusion/operator/NativeUdf.java index a01c4390..269f6edd 100644 --- a/src/main/java/tech/streamfusion/operator/NativeUdf.java +++ b/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 diff --git a/src/main/java/tech/streamfusion/planner/RexExpression.java b/src/main/java/tech/streamfusion/planner/RexExpression.java index 52855abe..52839170 100644 --- a/src/main/java/tech/streamfusion/planner/RexExpression.java +++ b/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; diff --git a/src/test/java/tech/streamfusion/FlinkJsonJvmSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkJsonJvmSqlHarnessTest.java index cb90fcd0..775e733f 100644 --- a/src/test/java/tech/streamfusion/FlinkJsonJvmSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkJsonJvmSqlHarnessTest.java @@ -179,16 +179,32 @@ 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); @@ -196,13 +212,16 @@ void mixedJsonCalcExecutesAcrossBatchesAndFiltersBeforeEvaluation() throws Excep 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()); } diff --git a/src/test/java/tech/streamfusion/FlinkMapJsonJvmSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkMapJsonJvmSqlHarnessTest.java new file mode 100644 index 00000000..af1fb6a2 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkMapJsonJvmSqlHarnessTest.java @@ -0,0 +1,170 @@ +package tech.streamfusion; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.typeutils.MultisetTypeInfo; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import tech.streamfusion.planner.NativePlanner; + +class FlinkMapJsonJvmSqlHarnessTest { + @ParameterizedTest + @ValueSource( + strings = { + "JSON_STRING(m)", + "JSON_STRING(ms)", + "JSON_STRING(nested)", + "m, ms, JSON_QUERY(s, '$.items[*]')", + "ROW(m, nested, JSON_QUERY(s, '$.items[*]'))", + "ARRAY[m, CAST(NULL AS MAP)]", + "MAP['first', JSON_QUERY(s, '$.items[*]'), 'second', JSON_STRING(m)]", + "JSON_OBJECT('map' VALUE m, 'bag' VALUE ms)", + "JSON_ARRAY(m, nested, ms)", + "JSON_STRING(m['a'])", + "CASE WHEN id = 2 THEN CAST(NULL AS MAP) ELSE m END, JSON_QUERY(s," + + " '$')" + }) + void mapsAndMultisetsUseTheGeneratedBatchBridge(String projection) throws Exception { + // The extra dynamic query keeps even a plain MAP result on the complete Calc bridge. + String sql = "SELECT id, " + projection + ", JSON_QUERY(s, '$.items[*]') FROM inputs"; + String plan = NativePlanner.explain(environment(false), sql); + assertTrue(plan.contains("NativeCalc") && plan.contains("jsonEvaluation=[JVM]"), plan); + NativeParity.assertParity(() -> environment(false), sql); + } + + @Test + void borrowedMapsPreserveChangelogKindsAndFiltering() throws Exception { + NativeParity.assertKindedParity( + () -> environment(true), + "SELECT id, m, nested, ms, JSON_STRING(m), JSON_QUERY(s, '$') FROM inputs WHERE id <> 2"); + } + + @ParameterizedTest + @ValueSource( + strings = {"JSON_STRING(ms)", "ROW(ms), JSON_QUERY(s, '$')", "ARRAY[ms], JSON_QUERY(s, '$')"}) + void nullableMultisetElementsStayOnFlink(String projection) throws Exception { + NativeParity.assertFallbackReasonContains( + () -> environment(false, true), "SELECT " + projection + " FROM inputs", "row-fused UDF"); + } + + @Test + void integerKeysAndExactDecimalValuesMatchFlink() throws Exception { + NativeParity.assertParity( + () -> { + var env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + var table = StreamTableEnvironment.create(env); + Map values = new LinkedHashMap<>(); + values.put(1, new java.math.BigDecimal("12345678901234567890.123456789")); + values.put(-2, null); + table.createTemporaryView( + "inputs", + env.fromData( + List.of(Row.of(values, "{}"), Row.of(Map.of(), "[]"), Row.of(null, null)), + Types.ROW_NAMED( + new String[] {"m", "s"}, Types.MAP(Types.INT, Types.BIG_DEC), Types.STRING)), + Schema.newBuilder() + .column("m", DataTypes.MAP(DataTypes.INT().notNull(), DataTypes.DECIMAL(38, 9))) + .column("s", DataTypes.STRING()) + .build()); + return table; + }, + "SELECT JSON_STRING(m[1]), m, JSON_QUERY(s, '$') FROM inputs"); + } + + @Test + void jsonMapFilterComposesWithGroupedAggregation() throws Exception { + String sql = + "SELECT MOD(id, 2) AS k, COUNT(*) AS n FROM inputs " + + "WHERE CHAR_LENGTH(JSON_STRING(m)) > 0 GROUP BY MOD(id, 2)"; + String plan = NativePlanner.explain(environment(false), sql); + assertTrue( + plan.contains("jsonEvaluation=[JVM]") && plan.contains("NativeColumnarGroupAggregate"), + plan); + NativeParity.assertParity(() -> environment(false), sql); + } + + private static TableEnvironment environment(boolean changelog) { + return environment(changelog, false); + } + + private static TableEnvironment environment(boolean changelog, boolean nullableMultisetElements) { + var env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + var table = StreamTableEnvironment.create(env); + Map nullableValues = new LinkedHashMap<>(); + nullableValues.put("a", "snowman ☃"); + nullableValues.put("b", null); + Map nested = new LinkedHashMap<>(); + nested.put("values", new String[] {"x", null, "y"}); + nested.put("empty", new String[] {}); + nested.put("absent", null); + Map bag = new LinkedHashMap<>(); + bag.put("first", 2); + bag.put("second", 3); + List rows = + new ArrayList<>( + List.of( + Row.of(1, nullableValues, nested, bag, "{\"items\":[1,2]}"), + Row.of(2, Map.of(), Map.of(), Map.of(), "{\"items\":[]}"), + Row.of(3, null, null, null, "broken"), + Row.of( + 4, + Map.of("last", "value"), + Map.of("last", new String[] {"z"}), + Map.of("last", 1), + null))); + if (changelog) { + Row before = Row.copy(rows.get(0)); + before.setKind(RowKind.UPDATE_BEFORE); + rows.add(before); + Row after = Row.copy(rows.get(0)); + after.setKind(RowKind.UPDATE_AFTER); + after.setField(1, Map.of("updated", "new")); + rows.add(after); + Row deleted = Row.copy(rows.get(3)); + deleted.setKind(RowKind.DELETE); + rows.add(deleted); + } + var input = + env.fromData( + rows, + Types.ROW_NAMED( + new String[] {"id", "m", "nested", "ms", "s"}, + Types.INT, + Types.MAP(Types.STRING, Types.STRING), + Types.MAP(Types.STRING, Types.OBJECT_ARRAY(Types.STRING)), + new MultisetTypeInfo<>(Types.STRING), + Types.STRING)); + var schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("m", DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.STRING())) + .column( + "nested", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.ARRAY(DataTypes.STRING()))) + .column( + "ms", + DataTypes.MULTISET( + nullableMultisetElements ? DataTypes.STRING() : DataTypes.STRING().notNull())) + .column("s", DataTypes.STRING()) + .build(); + table.createTemporaryView( + "inputs", + changelog ? table.fromChangelogStream(input, schema) : table.fromDataStream(input, schema)); + return table; + } +} diff --git a/src/test/java/tech/streamfusion/JsonMapCompositionBenchmark.java b/src/test/java/tech/streamfusion/JsonMapCompositionBenchmark.java new file mode 100644 index 00000000..3328edaa --- /dev/null +++ b/src/test/java/tech/streamfusion/JsonMapCompositionBenchmark.java @@ -0,0 +1,117 @@ +package tech.streamfusion; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; +import tech.streamfusion.planner.NativePlanner; + +@EnabledIfEnvironmentVariable(named = "SF_BENCHMARK", matches = "true") +class JsonMapCompositionBenchmark { + private static final long ROWS = Long.getLong("jsonmap.rows", 2_000_000L); + private static final int WARMUP = Integer.getInteger("jsonmap.warmup", 2); + private static final int RUNS = Integer.getInteger("jsonmap.runs", 5); + private static final String SQL = + "INSERT INTO sink SELECT k, COUNT(*) FROM inputs " + + "WHERE CHAR_LENGTH(JSON_STRING(m)) > 264 GROUP BY k"; + + @Test + void jsonFilterAndGroupedCount() throws Exception { + String plan = NativePlanner.explain(environment(), SQL); + for (String required : + new String[] { + "jsonEvaluation=[JVM]", "NativeColumnarGroupAggregate", "RowDataToArrow", "ArrowToRowData" + }) { + if (!plan.contains(required)) throw new IllegalStateException(plan); + } + double[][] times = new double[2][RUNS]; + List csv = new ArrayList<>(List.of("engine,trial,rows,seconds")); + for (int trial = 0; trial < WARMUP + RUNS; trial++) { + for (int turn = 0; turn < 2; turn++) { + int engine = (trial + turn) % 2; + var table = environment(); + var scan = engine == 1 ? NativePlanner.install(table) : null; + long start = System.nanoTime(); + table.executeSql(SQL).await(); + double seconds = (System.nanoTime() - start) / 1e9; + if (scan != null && scan.substitutions() == 0) + throw new IllegalStateException(scan.explainSummary()); + if (trial >= WARMUP) { + times[engine][trial - WARMUP] = seconds; + csv.add( + String.format( + Locale.ROOT, + "%s,%d,%d,%.9f", + engine == 1 ? "native" : "flink", + trial - WARMUP, + ROWS, + seconds)); + } + } + } + double host = median(times[0]); + double nativeTime = median(times[1]); + Files.write( + Path.of(System.getProperty("jsonmap.output", "target/json-map-composition.csv")), csv); + System.out.printf( + Locale.ROOT, + "[json-map-composition] rows=%d Flink=%.6fs Native=%.6fs ratio=%.3fx host_trials=%s" + + " native_trials=%s%n", + ROWS, + host, + nativeTime, + host / nativeTime, + Arrays.toString(times[0]), + Arrays.toString(times[1])); + } + + private static double median(double[] values) { + double[] sorted = values.clone(); + Arrays.sort(sorted); + return sorted[sorted.length / 2]; + } + + private static TableEnvironment environment() { + var env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + var table = StreamTableEnvironment.create(env); + String[] payloads = {"a".repeat(264), "b".repeat(264)}; + var source = + env.fromSequence(0, ROWS - 1) + .map( + i -> { + if (i % 7 == 0) return Row.of(i % 4096, null); + Map map = new LinkedHashMap<>(); + map.put("first", payloads[(int) (i % 2)]); + map.put("absent", null); + map.put("last", "tail"); + return Row.of(i % 4096, map); + }) + .returns( + Types.ROW_NAMED( + new String[] {"k", "m"}, Types.LONG, Types.MAP(Types.STRING, Types.STRING))); + table.createTemporaryView( + "inputs", + source, + Schema.newBuilder() + .column("k", DataTypes.BIGINT()) + .column("m", DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.STRING())) + .build()); + table.executeSql( + "CREATE TEMPORARY TABLE sink (k BIGINT, n BIGINT) WITH ('connector'='blackhole')"); + return table; + } +} diff --git a/src/test/java/tech/streamfusion/NativeParity.java b/src/test/java/tech/streamfusion/NativeParity.java index 2f41621c..0dc27569 100644 --- a/src/test/java/tech/streamfusion/NativeParity.java +++ b/src/test/java/tech/streamfusion/NativeParity.java @@ -218,6 +218,19 @@ static Object comparableValue(Object value) { if (value instanceof byte[] bytes) { return HexFormat.of().formatHex(bytes); } + if (value instanceof Map map) { + Map values = new java.util.LinkedHashMap<>(); + map.entrySet().stream() + .sorted(Comparator.comparing(entry -> String.valueOf(comparableValue(entry.getKey())))) + .forEach(entry -> values.put(comparableValue(entry.getKey()), comparableValue(entry.getValue()))); + return values; + } + if (value instanceof Row row) { + List fields = new ArrayList<>(); + fields.add(row.getKind()); + for (int i = 0; i < row.getArity(); i++) fields.add(comparableValue(row.getField(i))); + return fields; + } if (value != null && value.getClass().isArray()) { List values = new ArrayList<>(); for (int i = 0; i < java.lang.reflect.Array.getLength(value); i++) { diff --git a/src/test/java/tech/streamfusion/NativeParityValueTest.java b/src/test/java/tech/streamfusion/NativeParityValueTest.java new file mode 100644 index 00000000..613f13ab --- /dev/null +++ b/src/test/java/tech/streamfusion/NativeParityValueTest.java @@ -0,0 +1,32 @@ +package tech.streamfusion; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; +import org.junit.jupiter.api.Test; + +class NativeParityValueTest { + @Test + void nestedMapsCompareArrayContentsRegardlessOfEntryOrder() { + Map left = new LinkedHashMap<>(); + left.put("values", new String[] {"one", null}); + left.put("row", Row.of(new byte[] {1, 2}, null)); + left.put("absent", null); + Map right = new LinkedHashMap<>(); + right.put("absent", null); + right.put("row", Row.of(new byte[] {1, 2}, null)); + right.put("values", new String[] {"one", null}); + Object expected = NativeParity.comparableValue(left); + assertEquals(expected, NativeParity.comparableValue(right)); + assertEquals(expected.toString(), NativeParity.comparableValue(right).toString()); + right.put("values", new String[] {"different", null}); + assertNotEquals(expected, NativeParity.comparableValue(right)); + right.put("values", new String[] {"one", null}); + right.put("row", Row.ofKind(RowKind.DELETE, new byte[] {1, 2}, null)); + assertNotEquals(expected, NativeParity.comparableValue(right)); + } +} diff --git a/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java b/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java index eebf2728..e0025ae1 100644 --- a/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java +++ b/src/test/java/tech/streamfusion/ScalarFunctionBenchmark.java @@ -178,6 +178,10 @@ String ddl() { new Query("SHA1", "tt_text", "SHA1(s)"), new Query("JSON_STRING_TEXT", "tt_text", "JSON_STRING(s)"), new Query("JSON_STRING_ARRAY", "tt_json_array", "JSON_STRING(a)"), + new Query("JSON_STRING_MAP", "tt_json_map", "JSON_STRING(m)"), + new Query("JSON_QUERY_MAP_RESULT", "tt_json_map", + "ROW(m, JSON_QUERY(s, '$.items[*]'))", + "ROW, json_result STRING>"), new Query( "JSON_QUERY_ARRAY_RESULT", "tt_json_array", diff --git a/src/test/java/tech/streamfusion/TextTimeBenchmarkInputs.java b/src/test/java/tech/streamfusion/TextTimeBenchmarkInputs.java index 009e132d..85e5cf01 100644 --- a/src/test/java/tech/streamfusion/TextTimeBenchmarkInputs.java +++ b/src/test/java/tech/streamfusion/TextTimeBenchmarkInputs.java @@ -21,6 +21,7 @@ static String baselineExpression(String input) { case "tt_decimal", "tt_unix_time" -> "n"; case "tt_decimal_array" -> "a"; case "tt_json_array" -> "a"; + case "tt_json_map" -> "m"; case "tt_timestamp" -> "ts"; default -> "s"; }; @@ -34,6 +35,7 @@ static String baselineType(String input) { case "tt_unix_time" -> "BIGINT"; case "tt_decimal_array" -> "ARRAY"; case "tt_json_array" -> "ARRAY"; + case "tt_json_map" -> "MAP"; case "tt_timestamp" -> "TIMESTAMP(9)"; default -> "STRING"; }; @@ -48,7 +50,17 @@ static TableEnvironment environment( payload(unicode ? " |\u4e2daB\ud83d\ude00| " : " |abCd| efGh| ", bytes), payload(unicode ? " |\u00e9dE\ud83d\ude42| " : " |deFg| abCd| ", bytes) }; - if (input.equals("tt_json_array")) { + if (input.equals("tt_json_map")) { + tables.createTemporaryView("inputs", env.fromSequence(0, rows - 1) + .map(i -> { + java.util.Map map = new java.util.LinkedHashMap<>(); + map.put("first", text[(int) (i % 2)]); map.put("absent", null); map.put("last", "tail"); + return Row.of(isNull(i, nullEvery) ? null : map, "{\"items\":[1,2,3]}"); + }).returns(Types.ROW_NAMED(new String[] {"m", "s"}, + Types.MAP(Types.STRING, Types.STRING), Types.STRING)), + Schema.newBuilder().column("m", DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.STRING())) + .column("s", DataTypes.STRING()).build()); + } else if (input.equals("tt_json_array")) { tables.createTemporaryView( "inputs", env.fromSequence(0, rows - 1)