From ae79f97f625fe0b7a8b043c2302ac287796ab4b4 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 28 Aug 2026 18:05:03 -0400 Subject: [PATCH 01/10] Build the engine against a second Flink line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flink's streaming planner is not a stable API, so targeting more than one release line normally means duplicating the planner integration. Measuring the actual divergence first showed it is far smaller than that: across everything StreamFusion links against, the 2.1 and 2.2 planners disagree on a handful of members, and most of those are renames that erase to the same bytecode. What genuinely differs is narrow — a watermark push-down argument added in 2.2, two changelog-normalize predicates that only exist in 2.2, a state-backend accessor absent from 2.1, and one helper whose Scala and Java collection return types are indistinguishable after erasure and so only surface when compiling. Concentrate that divergence in a single class compiled once per line and selected by a build profile, with the values it hands back carried in line-neutral holders. The rest of the tree compiles unchanged for either target, which keeps the seam reviewable and stops version drift from leaking into the matchers and execution nodes. The profile also pins the dependencies that track the Flink line rather than our own release cadence: the Kafka connector and the Delta connector publish a build per line, and Flink's Protobuf format generates against a different Protobuf runtime major on each, which fails at runtime inside Flink's own deserializer rather than at build time. --- pom.xml | 26 ++- .../planner/compat/FlinkCompat.java | 183 +++++++++++++++++ .../planner/compat/FlinkCompat.java | 189 ++++++++++++++++++ .../planner/ChangelogNormalizeMatcher.java | 5 +- .../planner/LookupJoinMatcher.java | 23 +-- .../planner/NativeLookupJoinExecNode.java | 62 +++--- .../planner/ScanWatermarkSpec.java | 6 +- .../StreamPhysicalNativeLookupJoin.java | 22 +- .../planner/compat/AsyncLookupOptions.java | 27 +++ .../planner/compat/ExpandedCalc.java | 31 +++ .../planner/compat/GeneratedAsyncFetcher.java | 32 +++ .../planner/compat/LookupKeys.java | 42 ++++ .../state/RocksDBNativeKeyedStateBackend.java | 5 +- streamfusion-core/pom.xml | 19 ++ streamfusion-delta/pom.xml | 4 +- streamfusion-runtime/pom.xml | 19 ++ 16 files changed, 622 insertions(+), 73 deletions(-) create mode 100644 src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java create mode 100644 src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java create mode 100644 src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java create mode 100644 src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java create mode 100644 src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java create mode 100644 src/main/java/tech/streamfusion/planner/compat/LookupKeys.java diff --git a/pom.xml b/pom.xml index e81ac0da..915b20ed 100644 --- a/pom.xml +++ b/pom.xml @@ -58,9 +58,15 @@ UTF-8 5.10.2 18.3.0 + 2.2.1 - 4.2.0 2.12 @@ -183,7 +189,7 @@ org.apache.flink flink-connector-kafka - 5.0.0-2.2 + ${flink.connector.kafka.version} provided + + flink-2.1 + + 2.1.3 + 2.1 + 5.0.0-2.1 + java-flink2.1 + + 3.21.7 + + diff --git a/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java b/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java new file mode 100644 index 00000000..06405b1a --- /dev/null +++ b/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java @@ -0,0 +1,183 @@ +package tech.streamfusion.planner.compat; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexProgram; +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.api.dag.Transformation; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.functions.AsyncTableFunction; +import org.apache.flink.table.functions.TableFunction; +import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; +import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; +import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import org.apache.flink.table.planner.plan.utils.FunctionCallUtils; +import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend; +import org.apache.flink.table.types.logical.RowType; + +/** + * The Flink 2.1 half of the planner seam. See the 2.2 copy for the contract; only the Flink-facing + * names and the two capabilities 2.1 does not have differ. + */ +public final class FlinkCompat { + + private FlinkCompat() {} + + // ---------------------------------------------------------------- lookup join + + public static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + Map keys = new HashMap<>(); + scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) + .asJava() + .forEach((index, param) -> keys.put((Integer) index, param)); + return LookupKeys.of(keys); + } + + public static @Nullable String unsupportedKeyShape(LookupKeys keys) { + for (Object param : keys.raw().values()) { + if (!(param instanceof FunctionCallUtils.FieldRef) + && !(param instanceof FunctionCallUtils.Constant)) { + return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); + } + } + return null; + } + + public static @Nullable AsyncLookupOptions asyncOptions(StreamPhysicalLookupJoin join) { + if (join.asyncOptions().isEmpty()) { + return null; + } + FunctionCallUtils.AsyncOptions options = join.asyncOptions().get(); + return new AsyncLookupOptions(options.asyncBufferCapacity, options.keyOrdered); + } + + public static GeneratedAsyncFetcher generateAsyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + AsyncTableFunction lookupFunction, + String tableName) { + LookupJoinCodeGenerator.GeneratedTableFunctionWithDataType> + generated = + LookupJoinCodeGenerator.generateAsyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName); + return new GeneratedAsyncFetcher(generated.tableFunc(), generated.dataType()); + } + + public static GeneratedFunction> generateSyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + TableFunction lookupFunction, + String tableName, + boolean objectReuseEnabled) { + return LookupJoinCodeGenerator.generateSyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName, + objectReuseEnabled); + } + + public static Transformation applyCustomShufflePartitioner( + PlannerBase planner, + RelOptTable temporalTable, + RowType probeType, + LookupKeys lookupKeys, + Transformation rows, + ChangelogMode inputChangelogMode, + TransformationMetadata metadata) { + return LookupJoinUtil.tryApplyCustomShufflePartitioner( + planner, temporalTable, probeType, rawKeys(lookupKeys), rows, inputChangelogMode, metadata); + } + + private static List orderedKeys(LookupKeys lookupKeys) { + Map keys = rawKeys(lookupKeys); + List ordered = new ArrayList<>(keys.size()); + for (int key : LookupJoinUtil.getOrderedLookupKeys(keys.keySet())) { + ordered.add(keys.get(key)); + } + return ordered; + } + + @SuppressWarnings("unchecked") + private static Map rawKeys(LookupKeys lookupKeys) { + return (Map) (Map) lookupKeys.raw(); + } + + // ------------------------------------------------------- changelog normalize + + /** Flink 2.1 has no source-reuse marking pass, so a normalize never shares a source. */ + public static boolean sharesSourceOrCommonFilter(StreamPhysicalChangelogNormalize normalize) { + return false; + } + + // ------------------------------------------------------------ watermark push-down + + /** + * Flink 2.1 does not carry a rowtime expression on the pushed spec, and its watermark generator is + * generated from the watermark expression alone, so there is nothing to cross-check against. The + * caller reads the rowtime column out of the watermark expression either way. + */ + public static Optional watermarkRowtimeExpr(WatermarkPushDownSpec spec) { + return Optional.empty(); + } + + // ---------------------------------------------------------------- dimension calc + + /** Flink 2.1 returns the projection as a Scala {@code Seq}; the erased type is identical. */ + public static ExpandedCalc expandCalcProgram(RexProgram calc) { + scala.Tuple2, scala.Option> expanded = + FlinkRexUtil.expandRexProgram(calc); + return new ExpandedCalc( + scala.collection.JavaConverters.seqAsJavaListConverter(expanded._1()).asJava(), + expanded._2().isDefined() ? expanded._2().get() : null); + } + + // ----------------------------------------------------------------- state backend + + /** + * Flink 2.1's keyed-state backend has no type identifier, so nothing on that line consumes this; + * the value names the backend this one delegates to. + */ + public static String backendTypeIdentifier(CheckpointableKeyedStateBackend delegate) { + return "rocksdb"; + } +} diff --git a/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java b/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java new file mode 100644 index 00000000..f713f7df --- /dev/null +++ b/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java @@ -0,0 +1,189 @@ +package tech.streamfusion.planner.compat; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexProgram; +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.api.dag.Transformation; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.functions.AsyncTableFunction; +import org.apache.flink.table.functions.TableFunction; +import org.apache.flink.table.planner.codegen.FunctionCallCodeGenerator; +import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; +import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; +import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; +import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend; +import org.apache.flink.table.types.logical.RowType; + +/** + * The Flink 2.2 half of the planner seam. + * + *

Everything Flink changed between the supported minor versions is reached through this class, so + * the rest of the planner compiles once against a single source tree. Only the file is swapped per + * Flink line — the signatures below are the contract both copies satisfy. + */ +public final class FlinkCompat { + + private FlinkCompat() {} + + // ---------------------------------------------------------------- lookup join + + /** Flink 2.2 renamed {@code FunctionCallUtils} to {@code FunctionCallUtil}; members are equal. */ + public static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + Map keys = new HashMap<>(); + scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) + .asJava() + .forEach((index, param) -> keys.put((Integer) index, param)); + return LookupKeys.of(keys); + } + + /** Returns a decline reason when any key is not a plain field reference or constant. */ + public static @Nullable String unsupportedKeyShape(LookupKeys keys) { + for (Object param : keys.raw().values()) { + if (!(param instanceof FunctionCallUtil.FieldRef) + && !(param instanceof FunctionCallUtil.Constant)) { + return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); + } + } + return null; + } + + public static @Nullable AsyncLookupOptions asyncOptions(StreamPhysicalLookupJoin join) { + if (join.asyncOptions().isEmpty()) { + return null; + } + FunctionCallUtil.AsyncOptions options = join.asyncOptions().get(); + return new AsyncLookupOptions(options.asyncBufferCapacity, options.keyOrdered); + } + + public static GeneratedAsyncFetcher generateAsyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + AsyncTableFunction lookupFunction, + String tableName) { + FunctionCallCodeGenerator.GeneratedTableFunctionWithDataType> + generated = + LookupJoinCodeGenerator.generateAsyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName); + return new GeneratedAsyncFetcher(generated.tableFunc(), generated.dataType()); + } + + public static GeneratedFunction> generateSyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + TableFunction lookupFunction, + String tableName, + boolean objectReuseEnabled) { + return LookupJoinCodeGenerator.generateSyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName, + objectReuseEnabled); + } + + /** The connector-owned partitioning SPI, which is typed on the renamed parameter map. */ + public static Transformation applyCustomShufflePartitioner( + PlannerBase planner, + RelOptTable temporalTable, + RowType probeType, + LookupKeys lookupKeys, + Transformation rows, + ChangelogMode inputChangelogMode, + TransformationMetadata metadata) { + return LookupJoinUtil.tryApplyCustomShufflePartitioner( + planner, temporalTable, probeType, rawKeys(lookupKeys), rows, inputChangelogMode, metadata); + } + + private static List orderedKeys(LookupKeys lookupKeys) { + Map keys = rawKeys(lookupKeys); + List ordered = new ArrayList<>(keys.size()); + for (int key : LookupJoinUtil.getOrderedLookupKeys(keys.keySet())) { + ordered.add(keys.get(key)); + } + return ordered; + } + + @SuppressWarnings("unchecked") + private static Map rawKeys(LookupKeys lookupKeys) { + return (Map) (Map) lookupKeys.raw(); + } + + // ------------------------------------------------------- changelog normalize + + /** + * Whether the rel carries the source-reuse marking Flink 2.2 added. The 2.2 optimizer runs a + * {@code FlinkMarkChangelogNormalizeProgram} pass that can share one normalize across reused + * sources and hoist a common filter; the native operator reproduces neither. + */ + public static boolean sharesSourceOrCommonFilter(StreamPhysicalChangelogNormalize normalize) { + return normalize.sourceReused() || normalize.commonFilter().length > 0; + } + + // ------------------------------------------------------------ watermark push-down + + /** + * The rowtime expression Flink 2.2 carries alongside the watermark expression. Flink 2.1 keeps + * only the watermark expression, so the caller falls back to deriving the rowtime field itself. + */ + public static Optional watermarkRowtimeExpr(WatermarkPushDownSpec spec) { + return spec.getRowtimeExpr(); + } + + // ---------------------------------------------------------------- dimension calc + + /** Flink 2.2 returns the projection as a {@code java.util.List}. */ + public static ExpandedCalc expandCalcProgram(RexProgram calc) { + scala.Tuple2, scala.Option> expanded = + FlinkRexUtil.expandRexProgram(calc); + return new ExpandedCalc( + expanded._1(), expanded._2().isDefined() ? expanded._2().get() : null); + } + + // ----------------------------------------------------------------- state backend + + /** Reported through the keyed-state backend interface from Flink 2.2 on. */ + public static String backendTypeIdentifier(CheckpointableKeyedStateBackend delegate) { + return delegate.getBackendTypeIdentifier(); + } +} diff --git a/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java b/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java index 59e014ce..69024b17 100644 --- a/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java +++ b/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java @@ -1,6 +1,7 @@ package tech.streamfusion.planner; import tech.streamfusion.operator.RowDataArrowConverter; +import tech.streamfusion.planner.compat.FlinkCompat; import org.apache.calcite.rel.RelNode; import org.apache.flink.table.planner.calcite.FlinkTypeFactory$; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; @@ -23,7 +24,7 @@ static boolean matches(StreamPhysicalChangelogNormalize node) { if (node.filterCondition() != null) { return false; // a pushed filter condition is not yet reproduced } - if (node.sourceReused() || node.commonFilter().length > 0) { + if (FlinkCompat.sharesSourceOrCommonFilter(node)) { return false; // the source-reuse rewrite changes the operator's contract } return RowDataArrowConverter.supports( @@ -42,7 +43,7 @@ static String unsupportedReason(StreamPhysicalChangelogNormalize node) { if (node.filterCondition() != null) { return "changelog normalize: a pushed filter condition is not supported"; } - if (node.sourceReused() || node.commonFilter().length > 0) { + if (FlinkCompat.sharesSourceOrCommonFilter(node)) { return "changelog normalize: the source-reuse variant is not supported"; } return "changelog normalize: needs a row type the Arrow conversion supports"; diff --git a/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java b/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java index 313efb12..d3f58b26 100644 --- a/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java +++ b/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java @@ -1,13 +1,12 @@ package tech.streamfusion.planner; -import java.util.HashMap; -import java.util.Map; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.core.JoinRelType; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; import org.apache.flink.table.planner.plan.schema.TableSourceTable; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.LookupKeys; /** * Recognizes the processing-time lookup joins the native operator runs: {@code probe JOIN dim FOR @@ -40,22 +39,12 @@ static String unsupportedReason(StreamPhysicalLookupJoin join) { if (!(unwrapTable(join.temporalTable()) instanceof TableSourceTable)) { return "lookup join: temporal table is not a (non-legacy) table source"; } - for (FunctionCallUtil.FunctionParam param : lookupKeys(join).values()) { - if (!(param instanceof FunctionCallUtil.FieldRef) - && !(param instanceof FunctionCallUtil.Constant)) { - return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); - } - } - return null; + return FlinkCompat.unsupportedKeyShape(lookupKeys(join)); } /** The dimension key → probe field/constant map the generated fetcher builds its key row from. */ - static Map lookupKeys(StreamPhysicalLookupJoin join) { - Map keys = new HashMap<>(); - scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) - .asJava() - .forEach((index, param) -> keys.put((Integer) index, param)); - return keys; + static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + return FlinkCompat.lookupKeys(join); } static boolean isLeftOuterJoin(StreamPhysicalLookupJoin join) { @@ -85,7 +74,7 @@ static RelNode substitute(StreamPhysicalLookupJoin join, PlanContext ctx) { join.finalPreFilterCondition().isDefined() ? join.finalPreFilterCondition().get() : null, join.finalRemainingCondition().isDefined() ? join.finalRemainingCondition().get() : null, LookupJoinMatcher.isLeftOuterJoin(join), - join.asyncOptions().isDefined() ? join.asyncOptions().get() : null, + FlinkCompat.asyncOptions(join), join.retryOptions().isDefined() ? join.retryOptions().get() : null, join.preferCustomShuffle(), join.inputChangelogMode()); diff --git a/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java b/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java index e594cb89..4ecdee0d 100644 --- a/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java +++ b/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java @@ -6,10 +6,12 @@ import tech.streamfusion.operator.NativeAsyncLookupJoinOperator; import tech.streamfusion.operator.NativeLookupJoinOperator; import tech.streamfusion.operator.RowDataToArrowOperator; -import java.util.ArrayList; +import tech.streamfusion.planner.compat.AsyncLookupOptions; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.GeneratedAsyncFetcher; +import tech.streamfusion.planner.compat.LookupKeys; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Optional; import javax.annotation.Nullable; import org.apache.calcite.plan.RelOptTable; @@ -19,7 +21,6 @@ import org.apache.flink.api.dag.Transformation; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.configuration.ReadableConfig; -import org.apache.flink.streaming.api.functions.async.AsyncFunction; import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.connector.ChangelogMode; @@ -32,7 +33,6 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.codegen.CodeGeneratorContext; import org.apache.flink.table.planner.codegen.FilterCodeGenerator; -import org.apache.flink.table.planner.codegen.FunctionCallCodeGenerator; import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; import org.apache.flink.table.planner.delegation.PlannerBase; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase; @@ -42,7 +42,6 @@ import org.apache.flink.table.planner.plan.nodes.exec.SingleTransformationTranslator; import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode; import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; import org.apache.flink.table.planner.utils.JavaScalaConversionUtil; import org.apache.flink.table.planner.utils.ShortcutUtils; @@ -81,13 +80,13 @@ public class NativeLookupJoinExecNode extends ExecNodeBase private final RelOptTable temporalTable; private final RowType probeType; - private final Map lookupKeys; + private final LookupKeys lookupKeys; private final @Nullable List projectionOnTemporalTable; private final @Nullable RexNode filterOnTemporalTable; private final @Nullable RexNode preFilterCondition; private final @Nullable RexNode remainingJoinCondition; private final boolean leftOuterJoin; - private final @Nullable FunctionCallUtil.AsyncOptions asyncOptions; + private final @Nullable AsyncLookupOptions asyncOptions; private final @Nullable LookupJoinUtil.RetryLookupOptions retryOptions; private final boolean preferCustomShuffle; private final ChangelogMode inputChangelogMode; @@ -99,13 +98,13 @@ public NativeLookupJoinExecNode( String description, RelOptTable temporalTable, RowType probeType, - Map lookupKeys, + LookupKeys lookupKeys, @Nullable List projectionOnTemporalTable, @Nullable RexNode filterOnTemporalTable, @Nullable RexNode preFilterCondition, @Nullable RexNode remainingJoinCondition, boolean leftOuterJoin, - @Nullable FunctionCallUtil.AsyncOptions asyncOptions, + @Nullable AsyncLookupOptions asyncOptions, @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, boolean preferCustomShuffle, ChangelogMode inputChangelogMode) { @@ -144,17 +143,13 @@ protected Transformation translateToPlanInternal( RowType resultRowType = (RowType) getOutputType(); String tableName = String.join(".", temporalTable.getQualifiedName()); - List orderedKeys = new ArrayList<>(lookupKeys.size()); - for (int key : LookupJoinUtil.getOrderedLookupKeys(lookupKeys.keySet())) { - orderedKeys.add(lookupKeys.get(key)); - } boolean async = asyncOptions != null; ResultRetryStrategy retryStrategy = retryOptions == null ? ResultRetryStrategy.NO_RETRY_STRATEGY : retryOptions.toRetryStrategy(); UserDefinedFunction lookupFunction = LookupJoinUtil.getLookupFunction( temporalTable, - lookupKeys.keySet(), + lookupKeys.indexes(), classLoader, async, retryStrategy, @@ -173,7 +168,7 @@ protected Transformation translateToPlanInternal( input.getParallelism(), false); rows = - LookupJoinUtil.tryApplyCustomShufflePartitioner( + FlinkCompat.applyCustomShufflePartitioner( planner, temporalTable, probeType, @@ -222,18 +217,17 @@ protected Transformation translateToPlanInternal( OneInputStreamOperator operator; if (async) { - FunctionCallCodeGenerator.GeneratedTableFunctionWithDataType> - generatedFetcher = - LookupJoinCodeGenerator.generateAsyncLookupFunction( - config, - classLoader, - dataTypeFactory, - probeType, - tableSourceRowType, - resultRowType, - orderedKeys, - (AsyncTableFunction) lookupFunction, - tableName); + GeneratedAsyncFetcher generatedFetcher = + FlinkCompat.generateAsyncFetcher( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + lookupKeys, + (AsyncTableFunction) lookupFunction, + tableName); GeneratedResultFuture> generatedResultFuture = LookupJoinCodeGenerator.generateTableAsyncCollector( config, @@ -249,35 +243,35 @@ protected Transformation translateToPlanInternal( AsyncLookupJoinRunner runner = generatedCalc != null ? new AsyncLookupJoinWithCalcRunner( - generatedFetcher.tableFunc(), + generatedFetcher.tableFunction(), fetcherConverter, generatedCalc, generatedResultFuture, generatedPreFilter, InternalSerializers.create(rightRowType), leftOuterJoin, - asyncOptions.asyncBufferCapacity) + asyncOptions.bufferCapacity()) : new AsyncLookupJoinRunner( - generatedFetcher.tableFunc(), + generatedFetcher.tableFunction(), fetcherConverter, generatedResultFuture, generatedPreFilter, InternalSerializers.create(rightRowType), leftOuterJoin, - asyncOptions.asyncBufferCapacity); + asyncOptions.bufferCapacity()); operator = new NativeAsyncLookupJoinOperator( - runner, probeType, resultRowType, asyncOptions.keyOrdered); + runner, probeType, resultRowType, asyncOptions.keyOrdered()); } else { GeneratedFunction> generatedFetcher = - LookupJoinCodeGenerator.generateSyncLookupFunction( + FlinkCompat.generateSyncFetcher( config, classLoader, dataTypeFactory, probeType, tableSourceRowType, resultRowType, - orderedKeys, + lookupKeys, (TableFunction) lookupFunction, tableName, planner.getExecEnv().getConfig().isObjectReuseEnabled()); diff --git a/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java b/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java index 0d0bfb38..0810bba8 100644 --- a/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java +++ b/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java @@ -12,6 +12,7 @@ import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec; import org.apache.flink.table.planner.plan.abilities.source.SourceWatermarkSpec; import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import tech.streamfusion.planner.compat.FlinkCompat; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalTableSourceScan; import org.apache.flink.table.planner.plan.schema.TableSourceTable; import org.apache.flink.table.planner.utils.ShortcutUtils; @@ -75,8 +76,9 @@ static ScanWatermarkSpec of(StreamPhysicalTableSourceScan scan) { // computed rowtime); it must be one of the supported terms and agree with the watermark // expression's column. Integer rowtimeFromExpr = null; - if (pushed.getRowtimeExpr().isPresent()) { - Integer index = rowtimeTerm(stripReinterpret(pushed.getRowtimeExpr().get())); + var declaredRowtime = FlinkCompat.watermarkRowtimeExpr(pushed); + if (declaredRowtime.isPresent()) { + Integer index = rowtimeTerm(stripReinterpret(declaredRowtime.get())); if (index == null) { return UNSUPPORTED; } diff --git a/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java b/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java index 128f6938..fd85300b 100644 --- a/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java +++ b/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java @@ -1,7 +1,6 @@ package tech.streamfusion.planner; import java.util.List; -import java.util.Map; import javax.annotation.Nullable; import org.apache.calcite.plan.RelOptCluster; import org.apache.calcite.plan.RelOptTable; @@ -13,11 +12,13 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory$; import org.apache.flink.table.planner.plan.nodes.exec.ExecNode; import org.apache.flink.table.planner.plan.nodes.exec.InputProperty; -import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; import org.apache.flink.table.planner.utils.ShortcutUtils; import org.apache.flink.table.connector.ChangelogMode; +import tech.streamfusion.planner.compat.AsyncLookupOptions; +import tech.streamfusion.planner.compat.ExpandedCalc; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.LookupKeys; /** * Physical node standing in for a processing-time lookup join the native operator runs. Columnar on @@ -31,12 +32,12 @@ public class StreamPhysicalNativeLookupJoin extends StreamPhysicalNativeSingleRe implements ColumnarInput, ColumnarOutput { private final RelOptTable temporalTable; - private final Map lookupKeys; + private final LookupKeys lookupKeys; private final @Nullable RexProgram calcOnTemporalTable; private final @Nullable RexNode preFilterCondition; private final @Nullable RexNode remainingJoinCondition; private final boolean leftOuterJoin; - private final @Nullable FunctionCallUtil.AsyncOptions asyncOptions; + private final @Nullable AsyncLookupOptions asyncOptions; private final @Nullable LookupJoinUtil.RetryLookupOptions retryOptions; private final boolean preferCustomShuffle; private final ChangelogMode inputChangelogMode; @@ -47,12 +48,12 @@ public StreamPhysicalNativeLookupJoin( RelNode input, RelDataType outputRowType, RelOptTable temporalTable, - Map lookupKeys, + LookupKeys lookupKeys, @Nullable RexProgram calcOnTemporalTable, @Nullable RexNode preFilterCondition, @Nullable RexNode remainingJoinCondition, boolean leftOuterJoin, - @Nullable FunctionCallUtil.AsyncOptions asyncOptions, + @Nullable AsyncLookupOptions asyncOptions, @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, boolean preferCustomShuffle, ChangelogMode inputChangelogMode) { @@ -100,10 +101,9 @@ public ExecNode translateToExecNode() { List projectionOnTemporalTable = null; RexNode filterOnTemporalTable = null; if (calcOnTemporalTable != null) { - scala.Tuple2, scala.Option> expanded = - FlinkRexUtil.expandRexProgram(calcOnTemporalTable); - projectionOnTemporalTable = expanded._1(); - filterOnTemporalTable = expanded._2().isDefined() ? expanded._2().get() : null; + ExpandedCalc expanded = FlinkCompat.expandCalcProgram(calcOnTemporalTable); + projectionOnTemporalTable = expanded.projection(); + filterOnTemporalTable = expanded.filter(); } return new NativeLookupJoinExecNode( ShortcutUtils.unwrapTableConfig(this), diff --git a/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java b/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java new file mode 100644 index 00000000..ebb324be --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java @@ -0,0 +1,27 @@ +package tech.streamfusion.planner.compat; + +/** + * The async-lookup settings a lookup join was planned with, carried opaquely. + * + *

Flink renamed the enclosing utility between 2.1 and 2.2 but kept the fields, so the two values + * the native operator needs are copied out here and the original is retained only for the codegen + * call that still requires it. + */ +public final class AsyncLookupOptions { + + private final int bufferCapacity; + private final boolean keyOrdered; + + public AsyncLookupOptions(int bufferCapacity, boolean keyOrdered) { + this.bufferCapacity = bufferCapacity; + this.keyOrdered = keyOrdered; + } + + public int bufferCapacity() { + return bufferCapacity; + } + + public boolean keyOrdered() { + return keyOrdered; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java b/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java new file mode 100644 index 00000000..f74e1d44 --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java @@ -0,0 +1,31 @@ +package tech.streamfusion.planner.compat; + +import java.util.List; +import javax.annotation.Nullable; +import org.apache.calcite.rex.RexNode; + +/** + * A dimension-side calc split into its projection and optional filter. + * + *

Flink returns the projection as a Scala {@code Seq} on 2.1 and a {@code java.util.List} on 2.2. + * The two erase to the same descriptor, so the difference is invisible to bytecode comparison and + * only shows up when compiling — hence this holder rather than the raw tuple. + */ +public final class ExpandedCalc { + + private final List projection; + private final @Nullable RexNode filter; + + public ExpandedCalc(List projection, @Nullable RexNode filter) { + this.projection = projection; + this.filter = filter; + } + + public List projection() { + return projection; + } + + public @Nullable RexNode filter() { + return filter; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java b/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java new file mode 100644 index 00000000..83e11b91 --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java @@ -0,0 +1,32 @@ +package tech.streamfusion.planner.compat; + +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.table.types.DataType; + +/** + * The generated async lookup fetcher and the data type its results arrive in. + * + *

Flink moved the class that pairs these two between 2.1 and 2.2; both members are unchanged, so + * they are carried here instead and shared planner code never names the moved type. + */ +public final class GeneratedAsyncFetcher { + + private final GeneratedFunction> tableFunction; + private final DataType dataType; + + public GeneratedAsyncFetcher( + GeneratedFunction> tableFunction, DataType dataType) { + this.tableFunction = tableFunction; + this.dataType = dataType; + } + + public GeneratedFunction> tableFunction() { + return tableFunction; + } + + public DataType dataType() { + return dataType; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java b/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java new file mode 100644 index 00000000..cefc3aff --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java @@ -0,0 +1,42 @@ +package tech.streamfusion.planner.compat; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; + +/** + * The dimension-key → probe field/constant map a lookup join builds its key row from, carried + * opaquely. + * + *

Flink renamed the enclosing utility (and therefore the parameter type) between 2.1 and 2.2 + * without changing any member. Shared planner code passes this holder around and never names the + * Flink type; {@code FlinkCompat} — the one class compiled per Flink line — is the only place that + * unwraps it. + */ +public final class LookupKeys { + + private final Map byIndex; + + private LookupKeys(Map byIndex) { + this.byIndex = Collections.unmodifiableMap(byIndex); + } + + /** Wraps Flink's already-extracted parameter map. Called only from {@code FlinkCompat}. */ + public static LookupKeys of(Map byIndex) { + return new LookupKeys(new LinkedHashMap<>(byIndex)); + } + + public Set indexes() { + return byIndex.keySet(); + } + + public int size() { + return byIndex.size(); + } + + /** The raw Flink parameters. Callers outside {@code FlinkCompat} must treat these as opaque. */ + public Map raw() { + return byIndex; + } +} diff --git a/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java b/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java index 0f4c8c37..766b35ea 100644 --- a/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java +++ b/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java @@ -432,8 +432,9 @@ public boolean isSafeToReuseKVState() { return delegateUnchecked().isSafeToReuseKVState(); } - @Override + // Declared on the backend interface only from Flink 2.2; unused on 2.1. public String getBackendTypeIdentifier() { - return delegateUnchecked().getBackendTypeIdentifier(); + return tech.streamfusion.planner.compat.FlinkCompat.backendTypeIdentifier( + delegateUnchecked()); } } diff --git a/streamfusion-core/pom.xml b/streamfusion-core/pom.xml index d97c304b..5c097d1e 100644 --- a/streamfusion-core/pom.xml +++ b/streamfusion-core/pom.xml @@ -16,6 +16,25 @@ ${project.basedir}/../src/main/java + + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + add-source + + + ${project.basedir}/../src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-compiler-plugin diff --git a/streamfusion-delta/pom.xml b/streamfusion-delta/pom.xml index 650b3d95..410b3321 100644 --- a/streamfusion-delta/pom.xml +++ b/streamfusion-delta/pom.xml @@ -45,11 +45,11 @@ io.delta - delta-flink_2.2 + delta-flink_${flink.line} ${delta.version} provided - org.apache.logging.log4j diff --git a/streamfusion-runtime/pom.xml b/streamfusion-runtime/pom.xml index 911a7139..f20c75e2 100644 --- a/streamfusion-runtime/pom.xml +++ b/streamfusion-runtime/pom.xml @@ -42,6 +42,25 @@ + + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + add-source + + + ${project.basedir}/../src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-compiler-plugin From e2a5ee0c7f46cbdf0672b6b4f0186213f48c175b Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 28 Aug 2026 18:05:18 -0400 Subject: [PATCH 02/10] Serve either planner ABI from one loader artifact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loader shadows a private Flink class, so it has to match that class's exact signatures — and the two supported lines disagree on one. The accessor exposing the component classloader was narrowed to a concrete type in 2.2, so each line's callers are compiled against a different descriptor. Declaring either type alone turns a batch adaptive-join path into a missing-method failure on the other line. Declare the wider type in a base class and narrow it in the shim so the compiler emits a bridge, leaving one class that answers both descriptors. That keeps the shadow a signature detail rather than a second copy of the file to maintain. The fail-closed version whitelist is now compiled per line as well, so a build admits only the patch releases it was actually validated against instead of inheriting a list that happens to be true for a different target. --- divergences/23-flink-planner-loader-shadow.md | 16 +++++++-- streamfusion-loader/pom.xml | 34 +++++++++++++++++- .../loader/SupportedFlinkVersions.java | 14 ++++++++ .../loader/SupportedFlinkVersions.java | 17 +++++++++ .../table/planner/loader/PlannerModule.java | 5 +-- .../planner/loader/PlannerModuleCompat.java | 35 +++++++++++++++++++ 6 files changed, 115 insertions(+), 6 deletions(-) create mode 100644 streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java create mode 100644 streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java create mode 100644 streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java diff --git a/divergences/23-flink-planner-loader-shadow.md b/divergences/23-flink-planner-loader-shadow.md index c52612b5..ffe7e1a9 100644 --- a/divergences/23-flink-planner-loader-shadow.md +++ b/divergences/23-flink-planner-loader-shadow.md @@ -28,6 +28,16 @@ Flink still performs all standard planning and execution. The only injected beha scan stage, after which unsupported plan shapes retain their normal Flink nodes. The cost is a version-sensitive private-class seam. The shim supports exactly the tested Flink -**2.2.0 and 2.2.1** planner ABIs and fails closed for unknown or unversioned artifacts. It -rejects incompatible packaged versions at startup. A public upstream planner-extension API would -replace this file and remove the class-name shadow. +**2.1.3, 2.2.0 and 2.2.1** planner ABIs and fails closed for unknown or unversioned artifacts. It +rejects incompatible packaged versions at startup. Because the whitelist is compiled per Flink line, +a build admits only the patch versions it was actually validated against. + +Shadowing a private class also means matching its exact signatures, and the lines disagree on one: +the accessor exposing the component classloader was narrowed to a concrete type in 2.2, so the two +lines compile call sites against different descriptors and a single declared return type would be a +missing-method failure on whichever line lost the coin toss. Rather than fork the whole shim, it +declares the wider type in a base class and narrows it in the subclass, which makes the compiler +emit a bridge so one class satisfies both lines. That keeps the seam a signature detail instead of +a second copy of the file to maintain. + +A public upstream planner-extension API would replace this file and remove the class-name shadow. diff --git a/streamfusion-loader/pom.xml b/streamfusion-loader/pom.xml index 6317b8d3..edf38823 100644 --- a/streamfusion-loader/pom.xml +++ b/streamfusion-loader/pom.xml @@ -48,6 +48,8 @@ admitted planner ABI patch versions. --> 2.2.0 2.2.1 + 5.0.0-2.2 + java-flink2.2 5.10.2 @@ -99,7 +101,7 @@ org.apache.flink flink-connector-kafka - 5.0.0-2.2 + ${flink.connector.kafka.version} test @@ -130,6 +132,25 @@ + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + + add-source + + + + ${project.basedir}/src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-dependency-plugin @@ -171,6 +192,17 @@ + + + flink-2.1 + + 2.1.3 + 2.1.3 + 5.0.0-2.1 + java-flink2.1 + + diff --git a/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java b/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java new file mode 100644 index 00000000..e208a3d4 --- /dev/null +++ b/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java @@ -0,0 +1,14 @@ +package org.apache.flink.table.planner.loader; + +import java.util.Set; + +/** + * The Flink patch versions this loader build has been validated against. See the 2.2 copy for the + * contract. + */ +final class SupportedFlinkVersions { + + static final Set VERSIONS = Set.of("2.1.3"); + + private SupportedFlinkVersions() {} +} diff --git a/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java b/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java new file mode 100644 index 00000000..52a52cb3 --- /dev/null +++ b/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java @@ -0,0 +1,17 @@ +package org.apache.flink.table.planner.loader; + +import java.util.Set; + +/** + * The Flink patch versions this loader build has been validated against. + * + *

The loader shadows a Flink-internal class name, so it fails closed rather than cross an + * unverified planner ABI. The set is per Flink line and must only list versions the parity and + * upstream suites have actually run against. + */ +final class SupportedFlinkVersions { + + static final Set VERSIONS = Set.of("2.2.0", "2.2.1"); + + private SupportedFlinkVersions() {} +} diff --git a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java index 9733c07f..0325fe48 100644 --- a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java +++ b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java @@ -55,10 +55,10 @@ * other planner behavior continues through Flink's normal implementation. */ @Internal -public class PlannerModule { +public class PlannerModule extends PlannerModuleCompat { static final String FLINK_TABLE_PLANNER_FAT_JAR = "flink-table-planner.jar"; - private static final Set SUPPORTED_FLINK_VERSIONS = Set.of("2.2.0", "2.2.1"); + private static final Set SUPPORTED_FLINK_VERSIONS = SupportedFlinkVersions.VERSIONS; private static final String STREAMFUSION_PLANNER_JAR = "streamfusion-planner.jar"; private static final String[] STREAMFUSION_EXTENSION_PREFIXES = { "streamfusion-kafka-", @@ -129,6 +129,7 @@ private PlannerModule() { } } + @Override public URLClassLoader getSubmoduleClassLoader() { return submoduleClassLoader; } diff --git a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java new file mode 100644 index 00000000..80c320ae --- /dev/null +++ b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.loader; + +import org.apache.flink.annotation.Internal; + +/** + * Carries the Flink 2.1 shape of {@code getSubmoduleClassLoader}. + * + *

Flink 2.2 narrowed the return type to {@code URLClassLoader}, so its call sites are compiled + * against a descriptor Flink 2.1's are not: 2.1 calls {@code ()Ljava/lang/ClassLoader;} and 2.2 + * calls {@code ()Ljava/net/URLClassLoader;}. Declaring the wider type here and narrowing it in the + * subclass makes the compiler emit a bridge, so the loader exposes both descriptors and satisfies + * either line from one artifact. + */ +@Internal +abstract class PlannerModuleCompat { + + public abstract ClassLoader getSubmoduleClassLoader(); +} From 79f1f82e35be37b39d12816d2fec2dbe0ffc7d45 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 28 Aug 2026 18:05:32 -0400 Subject: [PATCH 03/10] Fail the upstream suite when the engine was never installed The suite installs StreamFusion by matching Flink classes by name at load time. A miss has always been silent: the suite runs stock Flink end to end, reports every test green, and proves nothing. That is tolerable while one Flink release is targeted and the names are known good, but it becomes actively misleading the moment the suite runs against more than one line, where a renamed or relocated injection point is exactly the failure being looked for. Record whether the injection point was instrumented and whether it was actually entered, then abort at shutdown if Flink's planner factory was loaded without it. A run that never installed the engine now fails loudly instead of passing. --- .../suite/StreamFusionSuiteAgent.java | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java b/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java index 856b733f..94f7f4ab 100644 --- a/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java +++ b/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java @@ -2,12 +2,18 @@ import java.lang.instrument.Instrumentation; import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; +import java.util.List; import java.util.Set; import java.util.WeakHashMap; import java.util.concurrent.atomic.AtomicBoolean; import net.bytebuddy.agent.builder.AgentBuilder; import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.dynamic.DynamicType; +import net.bytebuddy.utility.JavaModule; import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.takesArguments; @@ -32,6 +38,9 @@ public final class StreamFusionSuiteAgent { private static final AtomicBoolean NATIVE_MEMORY_STATE_REPORTED = new AtomicBoolean(); private static final AtomicBoolean ROCKSDB_STATE_REPORTED = new AtomicBoolean(); private static final AtomicBoolean NATIVE_PARQUET_WRITER_REPORTED = new AtomicBoolean(); + private static final Set TRANSFORMED_TYPES = Collections.synchronizedSet(new HashSet<>()); + private static final AtomicBoolean PLANNER_ADVICE_ENTERED = new AtomicBoolean(); + private static volatile Instrumentation INSTRUMENTATION; private static final Set INSTALLED_CONFIGS = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>())); private static final ThreadLocal UNMODIFIED_PLAN_SETUP = new ThreadLocal<>(); @@ -41,8 +50,13 @@ public final class StreamFusionSuiteAgent { private StreamFusionSuiteAgent() {} public static void premain(String arguments, Instrumentation instrumentation) { + INSTRUMENTATION = instrumentation; + Runtime.getRuntime() + .addShutdownHook( + new Thread(StreamFusionSuiteAgent::auditInterception, "streamfusion-suite-audit")); new AgentBuilder.Default() .with(AgentBuilder.Listener.StreamWriting.toSystemError().withTransformationsOnly()) + .with(new InterceptionAudit()) .type(named(PLANNER_FACTORY)) .transform( (builder, type, classLoader, module, protectionDomain) -> @@ -124,6 +138,7 @@ private InstallStreamFusion() {} @Advice.OnMethodEnter static void enter(@Advice.Argument(0) Object context) { + StreamFusionSuiteAgent.recordPlannerAdviceEntered(); try { if (requiresUnmodifiedFlinkPlan()) { return; @@ -277,4 +292,72 @@ static void enter() { } } } + + public static void recordPlannerAdviceEntered() { + PLANNER_ADVICE_ENTERED.set(true); + } + + /** + * Fails the JVM when the planner interception never took effect. + * + *

The advice is attached by class and method name, so on an untested Flink version a rename or + * a signature change attaches nothing at all and the upstream suite then passes while running + * stock Flink — a green result that proves nothing. Loaded-but-never-instrumented is unambiguous + * and halts; instrumented-but-never-entered only warns, because a suite can load the factory + * without ever building a table environment. + */ + static void auditInterception() { + Instrumentation instrumentation = INSTRUMENTATION; + if (instrumentation == null) { + return; + } + boolean plannerLoaded = isLoaded(instrumentation, PLANNER_FACTORY); + boolean delegateLoaded = isLoaded(instrumentation, DELEGATE_PLANNER_FACTORY); + if (!plannerLoaded && !delegateLoaded) { + return; // no table stack in this JVM, so no interception was expected + } + List problems = new ArrayList<>(); + if (plannerLoaded && !TRANSFORMED_TYPES.contains(PLANNER_FACTORY)) { + problems.add(PLANNER_FACTORY + " was loaded but never instrumented"); + } + if (delegateLoaded && !TRANSFORMED_TYPES.contains(DELEGATE_PLANNER_FACTORY)) { + problems.add(DELEGATE_PLANNER_FACTORY + " was loaded but never instrumented"); + } + if (!problems.isEmpty()) { + System.err.println( + "FATAL: the StreamFusion suite agent did not instrument the Flink planner factory." + + " This run exercised stock Flink and any pass it reported is meaningless."); + problems.forEach(problem -> System.err.println(" - " + problem)); + System.err.flush(); + Runtime.getRuntime().halt(70); + } + if (!PLANNER_ADVICE_ENTERED.get()) { + System.err.println( + "WARNING: the StreamFusion suite agent instrumented the Flink planner factory, but its" + + " create(..) advice never ran — check that the method matcher still applies."); + } + } + + private static boolean isLoaded(Instrumentation instrumentation, String className) { + for (Class loaded : instrumentation.getAllLoadedClasses()) { + if (className.equals(loaded.getName())) { + return true; + } + } + return false; + } + + /** Records which interception targets actually took effect. */ + private static final class InterceptionAudit extends AgentBuilder.Listener.Adapter { + + @Override + public void onTransformation( + TypeDescription typeDescription, + ClassLoader classLoader, + JavaModule module, + boolean loaded, + DynamicType dynamicType) { + TRANSFORMED_TYPES.add(typeDescription.getName()); + } + } } From 0cffad9f66b226d7b67df3e25b9455e5bcbc7781 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 28 Aug 2026 18:05:45 -0400 Subject: [PATCH 04/10] Point the upstream suite at the Flink line under test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The suite harness pinned one release throughout: the tag it cloned, the connector release it resolved, and the profile StreamFusion itself was built with were all fixed independently. Running it against another line therefore risked the worst possible outcome — an engine compiled for one planner exercised against another, producing either a confusing failure or, worse, a green run that measured a mismatched pair. Derive all of it from the requested Flink version instead, so choosing a line selects the source tag, the matching connector release, and the build profile together. The harness can now be pointed at any supported line without editing it. --- bin/flink-suite.sh | 11 +++++++++++ dev/flink-suite/classpath-pom.xml | 7 ++++++- docs/upstream-flink-suite.md | 8 ++++++-- 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/bin/flink-suite.sh b/bin/flink-suite.sh index 84b09389..a680df99 100755 --- a/bin/flink-suite.sh +++ b/bin/flink-suite.sh @@ -5,6 +5,14 @@ set -uo pipefail readonly REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" readonly FLINK_VERSION="${FLINK_VERSION:-2.2.1}" readonly FLINK_TAG="release-${FLINK_VERSION}" +# The suite must build StreamFusion for the same Flink line it is about to run against. +readonly FLINK_LINE="${FLINK_VERSION%.*}" +readonly FLINK_KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}-${FLINK_LINE}" +if [[ "${FLINK_LINE}" == "2.2" ]]; then + readonly SF_FLINK_PROFILE_ARG="" +else + readonly SF_FLINK_PROFILE_ARG="-Pflink-${FLINK_LINE}" +fi readonly KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}" readonly KAFKA_CONNECTOR_TAG="v${KAFKA_CONNECTOR_VERSION}" readonly SUITE_ROOT="${FLINK_SUITE_ROOT:-${REPO_ROOT}/.flink-suite}" @@ -198,11 +206,14 @@ else echo "Building and installing StreamFusion and its supported connector/format modules against the source-suite planner..." mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \ -Dstreamfusion.flink-source-suite \ + ${SF_FLINK_PROFILE_ARG} \ -f "${STREAMFUSION_BUILD_ROOT}/pom.xml" \ -pl :streamfusion-core,:streamfusion-kafka,:streamfusion-json,:streamfusion-csv,:streamfusion-raw,:streamfusion-avro,:streamfusion-avro-confluent-registry,:streamfusion-protobuf,:streamfusion-parquet \ -am -DskipTests clean install || exit $? mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \ -f "${REPO_ROOT}/dev/flink-suite/classpath-pom.xml" \ + -Dflink.version="${FLINK_VERSION}" \ + -Dflink.connector.kafka.version="${FLINK_KAFKA_CONNECTOR_VERSION}" \ dependency:build-classpath -Dmdep.outputFile="${CLASSPATH_FILE}" || exit $? if [[ "${SUITE_MODE}" == "formats" || "${SUITE_MODE}" == "parquet" ]]; then diff --git a/dev/flink-suite/classpath-pom.xml b/dev/flink-suite/classpath-pom.xml index b46aefea..ea0fd9cf 100644 --- a/dev/flink-suite/classpath-pom.xml +++ b/dev/flink-suite/classpath-pom.xml @@ -6,6 +6,11 @@ tech.streamfusion streamfusion-flink-suite-classpath 1.0-SNAPSHOT + + + 2.2.1 + 5.0.0-2.2 + tech.streamfusion @@ -58,7 +63,7 @@ org.apache.flink flink-connector-kafka - 5.0.0-2.2 + ${flink.connector.kafka.version} diff --git a/docs/upstream-flink-suite.md b/docs/upstream-flink-suite.md index 5fdaebf4..8e9101d6 100644 --- a/docs/upstream-flink-suite.md +++ b/docs/upstream-flink-suite.md @@ -55,8 +55,12 @@ coverage; their `Calc` versus `NativeCalc`-style diffs are diagnostic output, no The checkout is cached between runs. Set `FLINK_SUITE_ROOT` to put it elsewhere, or tune local test parallelism with `FLINK_SUITE_UNIT_FORKS` and `FLINK_SUITE_IT_FORKS`. The runner uses only public -artifact repositories, independent of developer-specific Maven mirrors. `FLINK_VERSION` is pinned by -the harness and should only be changed after validating the injection point against that release. +artifact repositories, independent of developer-specific Maven mirrors. `FLINK_VERSION` selects the +Flink line under test and drives everything derived from it — the release tag cloned, the matching +connector release, and the build profile StreamFusion itself is compiled with. Setting it to a +release whose injection point has not been validated is the one thing that needs review first; +building StreamFusion for a different line than the planner it is about to run against would +otherwise produce a green suite that proves nothing. After a successful build, skip the StreamFusion and Flink rebuild while iterating on test selection: From 6362f4f497fcaeb578b171a28f1a8bf3490ead03 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 28 Aug 2026 18:06:00 -0400 Subject: [PATCH 05/10] Skip parity cases the host planner cannot express Parity tests assert that a query produces identical results on stock Flink and on the native plan. That assertion is undefined when Flink itself cannot plan the query: an older line rejects a left-joined ordinality unnest outright, in its own optimizer, before the engine is ever installed. The failure looks like a StreamFusion defect and is not one. Gate such a case on the Flink release that fixed it, reading the version from the running Flink rather than a build property so a test can never be excused on a version it did not actually load. The exclusion is deliberately narrow: it covers only behavior the host lacks, never a coverage gap of ours, which must keep failing. Newer lines continue to run the case with nothing skipped. --- .../streamfusion/EnabledIfFlinkAtLeast.java | 26 ++++++++++++++++ .../FlinkUnnestSqlHarnessTest.java | 4 +++ .../streamfusion/FlinkVersionCondition.java | 30 +++++++++++++++++++ .../tech/streamfusion/HostFlinkVersion.java | 25 ++++++++++++++++ 4 files changed, 85 insertions(+) create mode 100644 src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java create mode 100644 src/test/java/tech/streamfusion/FlinkVersionCondition.java create mode 100644 src/test/java/tech/streamfusion/HostFlinkVersion.java diff --git a/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java b/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java new file mode 100644 index 00000000..46d503df --- /dev/null +++ b/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java @@ -0,0 +1,26 @@ +package tech.streamfusion; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import org.junit.jupiter.api.extension.ExtendWith; + +/** + * Skips a test whose behaviour the Flink under test cannot exhibit yet. + * + *

Reserved for cases where the host itself lacks the behaviour, so there is no Flink result to + * be identical to. A StreamFusion coverage gap must never be hidden behind this. + */ +@Target({ElementType.TYPE, ElementType.METHOD}) +@Retention(RetentionPolicy.RUNTIME) +@ExtendWith(FlinkVersionCondition.class) +@interface EnabledIfFlinkAtLeast { + + int major(); + + int minor(); + + /** Why the older line cannot run it, ideally the upstream issue key. */ + String reason(); +} diff --git a/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java index 699eac20..94411a83 100644 --- a/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java @@ -78,6 +78,10 @@ void leftUnnestMatchesHost() throws Exception { } @Test + @EnabledIfFlinkAtLeast( + major = 2, + minor = 2, + reason = "FLINK-33217; earlier planners cannot type this query at all") void leftUnnestWithOrdinalityMatchesHost() throws Exception { // A LEFT null-pad row carries a null ordinal too. NativeParity.assertParity( diff --git a/src/test/java/tech/streamfusion/FlinkVersionCondition.java b/src/test/java/tech/streamfusion/FlinkVersionCondition.java new file mode 100644 index 00000000..29910939 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkVersionCondition.java @@ -0,0 +1,30 @@ +package tech.streamfusion; + +import java.util.Optional; +import org.junit.jupiter.api.extension.ConditionEvaluationResult; +import org.junit.jupiter.api.extension.ExecutionCondition; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.platform.commons.support.AnnotationSupport; + +class FlinkVersionCondition implements ExecutionCondition { + + @Override + public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { + Optional required = + AnnotationSupport.findAnnotation(context.getElement(), EnabledIfFlinkAtLeast.class); + if (required.isEmpty()) { + return ConditionEvaluationResult.enabled("no Flink version requirement"); + } + EnabledIfFlinkAtLeast annotation = required.get(); + if (HostFlinkVersion.atLeast(annotation.major(), annotation.minor())) { + return ConditionEvaluationResult.enabled("host Flink is new enough"); + } + return ConditionEvaluationResult.disabled( + "needs Flink %d.%d or newer (%s); host is %s" + .formatted( + annotation.major(), + annotation.minor(), + annotation.reason(), + HostFlinkVersion.current())); + } +} diff --git a/src/test/java/tech/streamfusion/HostFlinkVersion.java b/src/test/java/tech/streamfusion/HostFlinkVersion.java new file mode 100644 index 00000000..3fd5e67c --- /dev/null +++ b/src/test/java/tech/streamfusion/HostFlinkVersion.java @@ -0,0 +1,25 @@ +package tech.streamfusion; + +import org.apache.flink.runtime.util.EnvironmentInformation; + +/** + * The Flink line the tests are executing against. + * + *

Read from the running Flink rather than a build property so a test can never be gated on a + * version different from the one it actually loaded. + */ +final class HostFlinkVersion { + + private HostFlinkVersion() {} + + static String current() { + return EnvironmentInformation.getVersion(); + } + + static boolean atLeast(int major, int minor) { + String[] parts = current().split("[.-]"); + int hostMajor = Integer.parseInt(parts[0]); + int hostMinor = Integer.parseInt(parts[1]); + return hostMajor != major ? hostMajor > major : hostMinor >= minor; + } +} From 6aedbfe517c345f2e4304639f55ee9f78a21b916 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 4 Sep 2026 00:01:52 -0700 Subject: [PATCH 06/10] Warm the host cast path with a value its input type admits Casts are evaluated through Flink's own cast rules, and the code generated for a non-nullable input dereferences the value with no null guard. Priming that code with a null therefore aborted operator startup rather than surfacing the per-row error the query expects, turning every cast of a non-nullable string into a job failure. Prime with a value the declared input type admits instead. This was invisible locally because the harnesses only ever exercised nullable columns; non-nullable types reach the engine through SQL literals, so only the upstream suite covered the shape that broke. --- .../planner/HostCastFunction.java | 39 ++++++++++++- .../planner/HostCastFunctionTest.java | 57 +++++++++++++++++++ 2 files changed, 95 insertions(+), 1 deletion(-) create mode 100644 src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java diff --git a/src/main/java/tech/streamfusion/planner/HostCastFunction.java b/src/main/java/tech/streamfusion/planner/HostCastFunction.java index bf3eb571..e1c7464e 100644 --- a/src/main/java/tech/streamfusion/planner/HostCastFunction.java +++ b/src/main/java/tech/streamfusion/planner/HostCastFunction.java @@ -88,12 +88,49 @@ private void initializeExecutor(ClassLoader classLoader) { // java.lang.reflect switches from its native accessor to a generated accessor after a small // invocation threshold. Force that transition while Flink's job classloader is open; otherwise // a long-running native batch can cross the threshold after the safety wrapper was retired. + Object sample = warmupValue(inputType); for (int i = 0; i < 20; i++) { - executor.cast(null); + try { + executor.cast(sample); + } catch (Throwable warmupFailure) { + // Warming is an optimization, never a precondition: a rejected sample must not fail startup. + break; + } } } } + /** + * A value the executor can actually consume. {@code null} is not legal input for a NOT NULL type, + * whose generated cast dereferences the argument without a guard. + */ + private static Object warmupValue(LogicalType type) { + switch (type.getTypeRoot()) { + case CHAR: + case VARCHAR: + return StringData.fromString("0"); + case BOOLEAN: + return Boolean.FALSE; + case TINYINT: + return (byte) 0; + case SMALLINT: + return (short) 0; + case INTEGER: + return 0; + case BIGINT: + return 0L; + case FLOAT: + return 0f; + case DOUBLE: + return 0d; + case DECIMAL: + DecimalType decimal = (DecimalType) type; + return DecimalData.fromBigDecimal(BigDecimal.ZERO, decimal.getPrecision(), decimal.getScale()); + default: + return null; + } + } + /** The upcall marshals external values (String/BigDecimal/boxed numbers); the executor speaks * Flink's internal data. */ private Object toInternal(Object value) { diff --git a/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java b/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java new file mode 100644 index 00000000..7e5398f7 --- /dev/null +++ b/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java @@ -0,0 +1,57 @@ +package tech.streamfusion.planner; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.List; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.DoubleType; +import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.SmallIntType; +import org.apache.flink.table.types.logical.TinyIntType; +import org.apache.flink.table.types.logical.VarCharType; +import org.junit.jupiter.api.Test; + +class HostCastFunctionTest { + + private static final List NOT_NULL_STRINGS = + List.of( + new CharType(false, 3), + new VarCharType(false, 5), + new VarCharType(false, VarCharType.MAX_LENGTH)); + + private static final List NUMBERS = + List.of( + new TinyIntType(), + new SmallIntType(), + new IntType(), + new BigIntType(), + new FloatType(), + new DoubleType()); + + /** + * A NOT NULL input type's generated cast has no null guard — it trims the argument directly — so + * warming the executor with a null failed the operator's open() instead of any row. + */ + @Test + void opensForNotNullStringToNumberCasts() { + for (LogicalType input : NOT_NULL_STRINGS) { + for (LogicalType target : NUMBERS) { + assertDoesNotThrow( + () -> new HostCastFunction(input, target).open(null), input + " -> " + target); + } + } + } + + /** Warming must not consume the executor: the first real row still casts. */ + @Test + void castsAfterWarmup() { + HostCastFunction function = + new HostCastFunction(new VarCharType(false, 5), new IntType()); + function.open(null); + assertEquals(-7, function.eval("-7")); + } +} From f8460d0de5ef4f1a60eacd0ffb1d5089fbe24c30 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 4 Sep 2026 00:02:04 -0700 Subject: [PATCH 07/10] Accept either Arrow encoding of a day-time interval Day-time intervals cross the columnar boundary in two shapes. The engine canonicalises on integer milliseconds, but the query engine emits a native interval array whenever an expression's result type is an interval rather than a timestamp. The native side already absorbed both encodings; the Java boundary did not, so an interval-typed expression failed at runtime instead of returning a value. Read the native encoding as milliseconds at the boundary, mirroring how the TIME type already absorbs several Arrow encodings behind one SQL type. Boundary errors now name the Arrow vector and its type alongside the SQL type: dispatch happens on the vector, so the SQL type alone cannot explain a mismatch, and that blind spot hid this bug and an earlier one in the same code. --- .../streamfusion/arrow/ArrowConversion.java | 100 +++++++++++------- .../vectors/ArrowIntervalDayColumnVector.java | 56 ++++++++++ .../ArrowIntervalDayColumnVectorTest.java | 61 +++++++++++ 3 files changed, 176 insertions(+), 41 deletions(-) create mode 100644 src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java create mode 100644 src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java diff --git a/src/main/java/tech/streamfusion/arrow/ArrowConversion.java b/src/main/java/tech/streamfusion/arrow/ArrowConversion.java index 65abcf17..44b84ba3 100644 --- a/src/main/java/tech/streamfusion/arrow/ArrowConversion.java +++ b/src/main/java/tech/streamfusion/arrow/ArrowConversion.java @@ -18,48 +18,12 @@ package tech.streamfusion.arrow; -import tech.streamfusion.arrow.vectors.ArrowArrayColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBigIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBinaryColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBooleanColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDateColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDecimalColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDoubleColumnVector; -import tech.streamfusion.arrow.vectors.ArrowFloatColumnVector; -import tech.streamfusion.arrow.vectors.ArrowIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowMapColumnVector; -import tech.streamfusion.arrow.vectors.ArrowNullColumnVector; -import tech.streamfusion.arrow.vectors.ArrowRowColumnVector; -import tech.streamfusion.arrow.vectors.ArrowSmallIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTimeColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTimestampColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTinyIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowVarBinaryColumnVector; -import tech.streamfusion.arrow.vectors.ArrowVarCharColumnVector; -import tech.streamfusion.arrow.writers.ArrayWriter; -import tech.streamfusion.arrow.writers.ArrowFieldWriter; -import tech.streamfusion.arrow.writers.BigIntWriter; -import tech.streamfusion.arrow.writers.BinaryWriter; -import tech.streamfusion.arrow.writers.BooleanWriter; -import tech.streamfusion.arrow.writers.DateWriter; -import tech.streamfusion.arrow.writers.DecimalWriter; -import tech.streamfusion.arrow.writers.DoubleWriter; -import tech.streamfusion.arrow.writers.FloatWriter; -import tech.streamfusion.arrow.writers.IntWriter; -import tech.streamfusion.arrow.writers.MapWriter; -import tech.streamfusion.arrow.writers.NullWriter; -import tech.streamfusion.arrow.writers.RowWriter; -import tech.streamfusion.arrow.writers.SmallIntWriter; -import tech.streamfusion.arrow.writers.TimeWriter; -import tech.streamfusion.arrow.writers.TimestampWriter; -import tech.streamfusion.arrow.writers.TinyIntWriter; -import tech.streamfusion.arrow.writers.VarBinaryWriter; -import tech.streamfusion.arrow.writers.VarCharWriter; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.stream.Collectors; + import org.apache.arrow.vector.BigIntVector; import org.apache.arrow.vector.BitVector; import org.apache.arrow.vector.DateDayVector; @@ -69,6 +33,7 @@ import org.apache.arrow.vector.Float4Vector; import org.apache.arrow.vector.Float8Vector; import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.IntervalDayVector; import org.apache.arrow.vector.NullVector; import org.apache.arrow.vector.SmallIntVector; import org.apache.arrow.vector.TimeMicroVector; @@ -100,8 +65,8 @@ import org.apache.flink.table.types.logical.BooleanType; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.DateType; -import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DayTimeIntervalType; +import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; import org.apache.flink.table.types.logical.IntType; @@ -119,6 +84,45 @@ import org.apache.flink.table.types.logical.YearMonthIntervalType; import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor; +import tech.streamfusion.arrow.vectors.ArrowArrayColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBigIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBinaryColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBooleanColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDateColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDecimalColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDoubleColumnVector; +import tech.streamfusion.arrow.vectors.ArrowFloatColumnVector; +import tech.streamfusion.arrow.vectors.ArrowIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowIntervalDayColumnVector; +import tech.streamfusion.arrow.vectors.ArrowMapColumnVector; +import tech.streamfusion.arrow.vectors.ArrowNullColumnVector; +import tech.streamfusion.arrow.vectors.ArrowRowColumnVector; +import tech.streamfusion.arrow.vectors.ArrowSmallIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTimeColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTimestampColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTinyIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowVarBinaryColumnVector; +import tech.streamfusion.arrow.vectors.ArrowVarCharColumnVector; +import tech.streamfusion.arrow.writers.ArrayWriter; +import tech.streamfusion.arrow.writers.ArrowFieldWriter; +import tech.streamfusion.arrow.writers.BigIntWriter; +import tech.streamfusion.arrow.writers.BinaryWriter; +import tech.streamfusion.arrow.writers.BooleanWriter; +import tech.streamfusion.arrow.writers.DateWriter; +import tech.streamfusion.arrow.writers.DecimalWriter; +import tech.streamfusion.arrow.writers.DoubleWriter; +import tech.streamfusion.arrow.writers.FloatWriter; +import tech.streamfusion.arrow.writers.IntWriter; +import tech.streamfusion.arrow.writers.MapWriter; +import tech.streamfusion.arrow.writers.NullWriter; +import tech.streamfusion.arrow.writers.RowWriter; +import tech.streamfusion.arrow.writers.SmallIntWriter; +import tech.streamfusion.arrow.writers.TimeWriter; +import tech.streamfusion.arrow.writers.TimestampWriter; +import tech.streamfusion.arrow.writers.TinyIntWriter; +import tech.streamfusion.arrow.writers.VarBinaryWriter; +import tech.streamfusion.arrow.writers.VarCharWriter; + /** * The Arrow ↔ {@link RowData} type mapping, reader factory, and writer factory, ported (and trimmed) from * Flink's {@code org.apache.flink.table.runtime.arrow.ArrowUtils}. Vendored rather than depended on @@ -278,6 +282,8 @@ static ColumnVector createColumnVector(ValueVector vector, LogicalType fieldType || vector instanceof TimeMicroVector || vector instanceof TimeNanoVector) { return new ArrowTimeColumnVector(vector); + } else if (vector instanceof IntervalDayVector) { + return new ArrowIntervalDayColumnVector((IntervalDayVector) vector); } else if (vector instanceof TimeStampVector) { return new ArrowTimestampColumnVector(vector); } else if (vector instanceof MapVector) { @@ -306,7 +312,11 @@ static ColumnVector createColumnVector(ValueVector vector, LogicalType fieldType } else if (vector instanceof NullVector) { return ArrowNullColumnVector.INSTANCE; } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } @@ -374,7 +384,11 @@ private static ArrowFieldWriter createArrowFieldWriterForRow( } else if (vector instanceof NullVector) { return new NullWriter<>((NullVector) vector); } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } @@ -444,7 +458,11 @@ private static ArrowFieldWriter createArrowFieldWriterForArray( } else if (vector instanceof NullVector) { return new NullWriter<>((NullVector) vector); } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } diff --git a/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java b/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java new file mode 100644 index 00000000..d0209ae2 --- /dev/null +++ b/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package tech.streamfusion.arrow.vectors; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.columnar.vector.LongColumnVector; +import org.apache.flink.util.Preconditions; + +import org.apache.arrow.vector.IntervalDayVector; + +/** + * Arrow column vector for a day-time INTERVAL carried as Arrow's own {@code Interval(DAY_TIME)}. + * + *

StreamFusion's canonical form for these is a signed millisecond {@code Int64} — Flink's internal + * representation — but DataFusion produces a native interval array when an expression's result type + * is an interval rather than a timestamp. Both encodings therefore reach this boundary, exactly as + * the four Arrow time encodings do for {@code TIME}. + */ +@Internal +public final class ArrowIntervalDayColumnVector implements LongColumnVector { + + private static final long MILLIS_PER_DAY = 86_400_000L; + + private final IntervalDayVector valueVector; + + public ArrowIntervalDayColumnVector(IntervalDayVector valueVector) { + this.valueVector = Preconditions.checkNotNull(valueVector); + } + + @Override + public long getLong(int i) { + return IntervalDayVector.getDays(valueVector.getDataBuffer(), i) * MILLIS_PER_DAY + + IntervalDayVector.getMilliseconds(valueVector.getDataBuffer(), i); + } + + @Override + public boolean isNullAt(int i) { + return valueVector.isNull(i); + } +} diff --git a/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java b/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java new file mode 100644 index 00000000..d79af622 --- /dev/null +++ b/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java @@ -0,0 +1,61 @@ +package tech.streamfusion.arrow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collections; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.IntervalDayVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.IntervalUnit; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.flink.table.types.logical.DayTimeIntervalType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; + +/** + * A day-time INTERVAL reaches the boundary either as the Int64 millis StreamFusion canonicalises on, + * or as Arrow's own {@code Interval(DAY_TIME)} when the native result type is an interval rather than + * a timestamp. Both must read back as Flink's internal millisecond long. + */ +class ArrowIntervalDayColumnVectorTest { + + private static final RowType SCHEMA = + RowType.of( + new LogicalType[] { + new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.SECOND) + }, + new String[] {"i"}); + + @Test + void readsArrowDayTimeIntervalAsMillis() { + Field field = + new Field( + "i", + FieldType.nullable(new ArrowType.Interval(IntervalUnit.DAY_TIME)), + Collections.emptyList()); + try (BufferAllocator allocator = new RootAllocator(); + VectorSchemaRoot root = + VectorSchemaRoot.create(new Schema(Collections.singletonList(field)), allocator)) { + IntervalDayVector vector = (IntervalDayVector) root.getVector("i"); + vector.allocateNew(3); + vector.set(0, 0, 6_000); + // The days component must be folded in, not dropped. + vector.set(1, 2, 500); + vector.setNull(2); + vector.setValueCount(3); + root.setRowCount(3); + + ArrowReader reader = ArrowConversion.createArrowReader(root, SCHEMA); + + assertEquals(6_000L, reader.read(0).getLong(0)); + assertEquals(2 * 86_400_000L + 500L, reader.read(1).getLong(0)); + assertTrue(reader.read(2).isNullAt(0)); + } + } +} From dea78829acfa7b7ff17dfb5ff0890335a3d8b1f1 Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Fri, 4 Sep 2026 00:02:18 -0700 Subject: [PATCH 08/10] Leave a plan alone when the host means to reject it Flink validates a forced delta-join strategy after our pass has run, and raises an error only when an ordinary join survived optimization. Substituting that join away erased the very evidence the validation looks for, so a query the host intends to refuse ran silently instead. Any acceleration that changes whether a query is legal is a correctness bug, not a coverage question. Decline such plans wholesale. The condition mirrors the host's exactly rather than declining whenever the strategy is forced, so acceleration is retained for every plan the validation would have passed, including one that mixes a delta join with joins we do accelerate. --- .../planner/PhysicalPlanScan.java | 32 +++++ .../planner/DeltaJoinForceGateTest.java | 115 ++++++++++++++++++ 2 files changed, 147 insertions(+) create mode 100644 src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java diff --git a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java index bdf8f525..a4659834 100644 --- a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java +++ b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java @@ -22,6 +22,7 @@ import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalGroupAggregate; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalGroupWindowAggregate; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalIntervalJoin; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLimit; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLocalGroupAggregate; @@ -91,6 +92,14 @@ private RelNode optimizeConfigured(RelNode root) { LOG.info("StreamFusion native acceleration is disabled; the plan runs on Flink"); return root; } + // Flink runs its FORCE delta-join validation after this pass, and only complains when a regular + // join survives. Substituting one away would silently run a plan the host means to reject. + if (deltaJoinForceWouldReject(root)) { + fallbackReasons.add( + "delta join: table.optimizer.delta-join.strategy is FORCE but the plan has no delta join"); + LOG.info("StreamFusion declined the plan so Flink can enforce its FORCE delta-join strategy"); + return root; + } RelNode optimized = substitute(root); // The one always-on plan-time summary; -Dstreamfusion.logFallbackReasons=true itemizes the // reasons and explainSummary() carries them into explain output. @@ -721,6 +730,29 @@ private void record(RelNode node) { } } + /** Mirrors {@code StreamPhysicalDeltaJoinForceValidator}, which spares a plan that has any delta join. */ + private static boolean deltaJoinForceWouldReject(RelNode root) { + if (ShortcutUtils.unwrapTableConfig(root) + .get(OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY) + != OptimizerConfigOptions.DeltaJoinStrategy.FORCE) { + return false; + } + return contains(root, StreamPhysicalJoin.class) + && !contains(root, StreamPhysicalDeltaJoin.class); + } + + private static boolean contains(RelNode node, Class relType) { + if (relType.isInstance(node)) { + return true; + } + for (RelNode input : node.getInputs()) { + if (contains(input, relType)) { + return true; + } + } + return false; + } + /** Operator types seen in the optimized physical plans, in traversal order. */ public List operatorTypes() { return List.copyOf(operatorTypes); diff --git a/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java b/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java new file mode 100644 index 00000000..5088bb72 --- /dev/null +++ b/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java @@ -0,0 +1,115 @@ +package tech.streamfusion.planner; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.streaming.api.datastream.DataStream; +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.ValidationException; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.config.OptimizerConfigOptions; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +/** + * Flink validates its FORCE delta-join strategy after our pass runs, and rejects a plan only when a + * regular join survived. Substituting that join away would turn a query the host means to reject + * into one that silently runs, so the pass declines such plans wholesale. The gate mirrors the + * host's condition exactly rather than declining on FORCE alone — acceleration is kept for every + * plan the validator would have passed. + */ +class DeltaJoinForceGateTest { + + private static final String JOIN_QUERY = + "SELECT a.k, a.v, b.w FROM A AS a JOIN B AS b ON a.k = b.k"; + + @Test + void forceWithoutADeltaJoinLeavesThePlanForFlinkToReject() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.FORCE); + + ValidationException failure = + assertThrows(ValidationException.class, () -> tEnv.explainSql(JOIN_QUERY)); + + assertTrue( + failure.getMessage().contains("delta join"), + "expected Flink's own FORCE rejection, got: " + failure.getMessage()); + assertEquals(0, scan.substitutions(), scan::explainSummary); + assertTrue( + scan.fallbackReasons().stream().anyMatch(reason -> reason.startsWith("delta join:")), + "the decline must be reported as a fallback reason, saw: " + scan.fallbackReasons()); + } + + @Test + void forceStillAcceleratesAPlanWithoutAJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.FORCE); + + tEnv.explainSql("SELECT k, v * 2 FROM A"); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + @Test + void theDefaultStrategyAcceleratesTheSameJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + + tEnv.explainSql(JOIN_QUERY); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + @Test + void noneStrategyAcceleratesTheSameJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.NONE); + + tEnv.explainSql(JOIN_QUERY); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + private static TableEnvironment environment() { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + DataStream a = + env.fromData( + Types.ROW_NAMED(new String[] {"k", "v"}, Types.LONG, Types.LONG), + Row.of(1L, 10L), + Row.of(2L, 20L)); + DataStream b = + env.fromData( + Types.ROW_NAMED(new String[] {"k", "w"}, Types.LONG, Types.LONG), + Row.of(1L, 100L), + Row.of(2L, 200L)); + tEnv.createTemporaryView( + "A", + a, + Schema.newBuilder().column("k", DataTypes.BIGINT()).column("v", DataTypes.BIGINT()).build()); + tEnv.createTemporaryView( + "B", + b, + Schema.newBuilder().column("k", DataTypes.BIGINT()).column("w", DataTypes.BIGINT()).build()); + return tEnv; + } +} From 766c5be0ebb5bcc059cb52cf5271ca7bec4ac98e Mon Sep 17 00:00:00 2001 From: Devine Chinemere Date: Sat, 5 Sep 2026 15:04:02 -0700 Subject: [PATCH 09/10] Pin Calcite per Flink line The upstream-suite harness hardcoded a single Calcite version while Flink pins one per release line, so on the older line the engine was built against a parser the host never uses and whole suite modes failed before a single test ran. Take the version from the Flink checkout under test rather than restating it, and fail loudly if it cannot be found. The generated classpath is named per version as well, so a reused build can no longer silently run one line's tests against the other line's jars. --- bin/flink-suite.sh | 14 +++++++++++++- pom.xml | 6 +++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/bin/flink-suite.sh b/bin/flink-suite.sh index a680df99..553eeed2 100755 --- a/bin/flink-suite.sh +++ b/bin/flink-suite.sh @@ -21,7 +21,9 @@ readonly KAFKA_CONNECTOR_ROOT="${SUITE_ROOT}/flink-connector-kafka-${KAFKA_CONNE readonly STREAMFUSION_BUILD_ROOT="${SUITE_ROOT}/streamfusion-source" readonly AGENT_ROOT="${REPO_ROOT}/dev/flink-suite/agent" readonly AGENT_JAR="${AGENT_ROOT}/target/streamfusion-flink-suite-agent-1.0-SNAPSHOT.jar" -readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath.txt" +# Per line: the two lines resolve different Flink, Calcite and connector jars, so a shared file lets +# a reused build run one line's tests against the other line's classpath. +readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath-${FLINK_VERSION}.txt" readonly MAVEN_SETTINGS="${REPO_ROOT}/dev/flink-suite/settings.xml" readonly SUITE_MAVEN_REPO="${SUITE_ROOT}/m2" readonly UNSHADED_PLANNER_JAR="${SUITE_ROOT}/flink-table-planner-${FLINK_VERSION}-unshaded.jar" @@ -204,9 +206,19 @@ else ) || exit $? echo "Building and installing StreamFusion and its supported connector/format modules against the source-suite planner..." + # Flink pins Calcite per line, and the source-suite classpath must agree with it: a planner + # compiled against one Calcite cannot initialise its convertlet table against another. + readonly FLINK_TABLE_POM="${FLINK_ROOT}/flink-table/pom.xml" + CALCITE_VERSION="$(sed -n 's:.*\(.*\).*:\1:p' "${FLINK_TABLE_POM}" | head -1)" + if [[ -z "${CALCITE_VERSION}" ]]; then + echo "Could not read calcite.version from ${FLINK_TABLE_POM}" >&2 + exit 1 + fi + echo "Flink ${FLINK_VERSION} pins Calcite ${CALCITE_VERSION}." mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \ -Dstreamfusion.flink-source-suite \ ${SF_FLINK_PROFILE_ARG} \ + -Dcalcite.version="${CALCITE_VERSION}" \ -f "${STREAMFUSION_BUILD_ROOT}/pom.xml" \ -pl :streamfusion-core,:streamfusion-kafka,:streamfusion-json,:streamfusion-csv,:streamfusion-raw,:streamfusion-avro,:streamfusion-avro-confluent-registry,:streamfusion-protobuf,:streamfusion-parquet \ -am -DskipTests clean install || exit $? diff --git a/pom.xml b/pom.xml index 915b20ed..81f08f2d 100644 --- a/pom.xml +++ b/pom.xml @@ -72,6 +72,9 @@ 2.12 4.32.1 + + 1.36.0 4.4.0 @@ -481,6 +484,7 @@ java-flink2.1 3.21.7 + 1.34.0 2.2.1 + 2.2 5.0.0-2.2 tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-kafka + streamfusion-kafka-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-json + streamfusion-json-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-csv + streamfusion-csv-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-raw + streamfusion-raw-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-avro + streamfusion-avro-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-avro-confluent-registry + streamfusion-avro-confluent-registry-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-protobuf + streamfusion-protobuf-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-parquet + streamfusion-parquet-flink${flink.line} 0.1.0-rc2 2.2.0 2.2.1 + 2.2 5.0.0-2.2 java-flink2.2 5.10.2 @@ -64,7 +65,7 @@ are separate lib-directory JARs, matching Flink's connector packaging model. --> tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} runtime @@ -166,7 +167,7 @@ tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} runtime streamfusion-planner.jar @@ -199,6 +200,7 @@ 2.1.3 2.1.3 + 2.1 5.0.0-2.1 java-flink2.1 diff --git a/streamfusion-parquet/pom.xml b/streamfusion-parquet/pom.xml index b6c9c513..e11d0634 100644 --- a/streamfusion-parquet/pom.xml +++ b/streamfusion-parquet/pom.xml @@ -10,14 +10,14 @@ ${revision} - streamfusion-parquet + streamfusion-parquet-flink${flink.line} StreamFusion Parquet true tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} provided diff --git a/streamfusion-protobuf/pom.xml b/streamfusion-protobuf/pom.xml index 7198f567..1b311ef7 100644 --- a/streamfusion-protobuf/pom.xml +++ b/streamfusion-protobuf/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-protobuf + streamfusion-protobuf-flink${flink.line} StreamFusion Protobuf true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/protobuftech/streamfusion/native/protobuf**/libstreamfusion_protobuf.so**/libstreamfusion_protobuf.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/protobuf/**/*.javatech/streamfusion/planner/ProtobufDescriptors.java diff --git a/streamfusion-raw/pom.xml b/streamfusion-raw/pom.xml index 2fd2ce01..7e2737d8 100644 --- a/streamfusion-raw/pom.xml +++ b/streamfusion-raw/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-raw + streamfusion-raw-flink${flink.line} StreamFusion Raw true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/rawtech/streamfusion/native/raw**/libstreamfusion_raw.so**/libstreamfusion_raw.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/raw/**/*.java diff --git a/streamfusion-runtime/pom.xml b/streamfusion-runtime/pom.xml index f20c75e2..ca448b51 100644 --- a/streamfusion-runtime/pom.xml +++ b/streamfusion-runtime/pom.xml @@ -13,6 +13,11 @@ streamfusion-runtime StreamFusion Runtime + + true + ${project.basedir}/../src/main/java