From a8564016f2d169410d25442b6781de467e0b159d Mon Sep 17 00:00:00 2001 From: weiqingy Date: Sat, 18 Jul 2026 15:11:00 -0700 Subject: [PATCH 1/2] [FLINK-40168][table] Thread the EARLY_FIRE hint into the interval join --- .../exec/stream/StreamExecIntervalJoin.java | 25 + .../StreamPhysicalIntervalJoinRule.java | 59 ++- .../stream/StreamPhysicalIntervalJoin.scala | 13 +- .../hints/stream/EarlyFireJoinHintTest.java | 60 +++ .../hints/stream/EarlyFireJoinHintTest.xml | 70 ++- .../testEarlyFireJsonPlanRoundTrip.out | 446 ++++++++++++++++++ 6 files changed, 669 insertions(+), 4 deletions(-) create mode 100644 flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest_jsonplan/testEarlyFireJsonPlanRoundTrip.out diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java index 2ac0af781e0c66..20b676af78562e 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java @@ -28,6 +28,7 @@ import org.apache.flink.streaming.api.transformations.TwoInputTransformation; import org.apache.flink.streaming.api.transformations.UnionTransformation; import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.config.EarlyFireJoinHintOptions; import org.apache.flink.table.api.config.ExecutionConfigOptions; import org.apache.flink.table.data.RowData; import org.apache.flink.table.planner.delegation.PlannerBase; @@ -59,11 +60,14 @@ import org.apache.flink.shaded.guava33.com.google.common.collect.Lists; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.util.List; /** {@link StreamExecNode} for a time interval stream join. */ @@ -91,13 +95,27 @@ public class StreamExecIntervalJoin extends ExecNodeBase public static final String INTERVAL_JOIN_TRANSFORMATION = "interval-join"; public static final String FIELD_NAME_INTERVAL_JOIN_SPEC = "intervalJoinSpec"; + public static final String FIELD_NAME_EARLY_FIRE_DELAY = "earlyFireDelay"; + public static final String FIELD_NAME_EARLY_FIRE_TIME_MODE = "earlyFireTimeMode"; @JsonProperty(FIELD_NAME_INTERVAL_JOIN_SPEC) private final IntervalJoinSpec intervalJoinSpec; + @Nullable + @JsonProperty(FIELD_NAME_EARLY_FIRE_DELAY) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final Long earlyFireDelay; + + @Nullable + @JsonProperty(FIELD_NAME_EARLY_FIRE_TIME_MODE) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode; + public StreamExecIntervalJoin( ReadableConfig tableConfig, IntervalJoinSpec intervalJoinSpec, + @Nullable Long earlyFireDelay, + @Nullable EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode, InputProperty leftInputProperty, InputProperty rightInputProperty, RowType outputType, @@ -107,6 +125,8 @@ public StreamExecIntervalJoin( ExecNodeContext.newContext(StreamExecIntervalJoin.class), ExecNodeContext.newPersistedConfig(StreamExecIntervalJoin.class, tableConfig), intervalJoinSpec, + earlyFireDelay, + earlyFireTimeMode, Lists.newArrayList(leftInputProperty, rightInputProperty), outputType, description); @@ -118,12 +138,17 @@ public StreamExecIntervalJoin( @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context, @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig, @JsonProperty(FIELD_NAME_INTERVAL_JOIN_SPEC) IntervalJoinSpec intervalJoinSpec, + @Nullable @JsonProperty(FIELD_NAME_EARLY_FIRE_DELAY) Long earlyFireDelay, + @Nullable @JsonProperty(FIELD_NAME_EARLY_FIRE_TIME_MODE) + EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode, @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List inputProperties, @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType, @JsonProperty(FIELD_NAME_DESCRIPTION) String description) { super(id, context, persistedConfig, inputProperties, outputType, description); Preconditions.checkArgument(inputProperties.size() == 2); this.intervalJoinSpec = Preconditions.checkNotNull(intervalJoinSpec); + this.earlyFireDelay = earlyFireDelay; + this.earlyFireTimeMode = earlyFireTimeMode; } @Override diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java index 6354f04de07485..d7cbad27d0c5aa 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java @@ -19,9 +19,13 @@ package org.apache.flink.table.planner.plan.rules.physical.stream; import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.Configuration; import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.config.EarlyFireJoinHintOptions; +import org.apache.flink.table.api.config.EarlyFireJoinHintOptions.TimeMode; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; +import org.apache.flink.table.planner.hint.JoinStrategy; import org.apache.flink.table.planner.plan.nodes.FlinkRelNode; import org.apache.flink.table.planner.plan.nodes.exec.spec.IntervalJoinSpec; import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin; @@ -32,11 +36,16 @@ import org.apache.calcite.plan.RelOptRuleCall; import org.apache.calcite.plan.RelTraitSet; import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.hint.RelHint; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexNode; import org.immutables.value.Value; +import javax.annotation.Nullable; + +import java.time.Duration; import java.util.Collection; +import java.util.List; import java.util.function.Function; import java.util.stream.Collectors; @@ -133,6 +142,8 @@ public FlinkRelNode transform( RelTraitSet providedTraitSet) { Tuple2, Option> tuple2 = extractWindowBounds(join); + boolean isEventTime = tuple2.f0.get().isEventTime(); + EarlyFire earlyFire = extractEarlyFire(join.getHints(), isEventTime); return new StreamPhysicalIntervalJoin( join.getCluster(), providedTraitSet, @@ -141,7 +152,53 @@ public FlinkRelNode transform( join.getJoinType(), join.getCondition(), tuple2.f1.getOrElse(() -> join.getCluster().getRexBuilder().makeLiteral(true)), - tuple2.f0.get()); + tuple2.f0.get(), + earlyFire.delay, + earlyFire.timeMode); + } + + private static EarlyFire extractEarlyFire(List hints, boolean isEventTime) { + RelHint earlyFireHint = null; + for (RelHint hint : hints) { + if (JoinStrategy.isEarlyFireHint(hint.hintName)) { + earlyFireHint = hint; + break; + } + } + if (earlyFireHint == null) { + return new EarlyFire(null, null); + } + + Configuration conf = Configuration.fromMap(earlyFireHint.kvOptions); + Duration delay = conf.get(EarlyFireJoinHintOptions.DELAY); + TimeMode timeMode = conf.get(EarlyFireJoinHintOptions.TIME_MODE); + if (timeMode == null) { + timeMode = isEventTime ? TimeMode.ROWTIME : TimeMode.PROCTIME; + } + + if (!isEventTime && timeMode == TimeMode.ROWTIME) { + throw new ValidationException( + "EARLY_FIRE hint requested row-time triggering on a processing-time interval" + + " join. Row-time triggering requires a row-time interval join."); + } + if (isEventTime && timeMode == TimeMode.PROCTIME) { + // Processing-time triggering on an event-time interval join is not supported. + throw new TableException( + "EARLY_FIRE hint requested processing-time triggering on a row-time interval" + + " join, which is not yet supported."); + } + + return new EarlyFire(delay == null ? null : delay.toMillis(), timeMode); + } + + private static final class EarlyFire { + @Nullable private final Long delay; + @Nullable private final TimeMode timeMode; + + EarlyFire(@Nullable Long delay, @Nullable TimeMode timeMode) { + this.delay = delay; + this.timeMode = timeMode; + } } /** Configuration for {@link StreamPhysicalIntervalJoinRule}. */ diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala index 4916d653b2e236..d3401783989d9f 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala @@ -18,6 +18,7 @@ package org.apache.flink.table.planner.plan.nodes.physical.stream import org.apache.flink.table.api.TableException +import org.apache.flink.table.api.config.EarlyFireJoinHintOptions.TimeMode import org.apache.flink.table.planner.calcite.FlinkTypeFactory import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty} import org.apache.flink.table.planner.plan.nodes.exec.spec.IntervalJoinSpec @@ -45,7 +46,9 @@ class StreamPhysicalIntervalJoin( val originalCondition: RexNode, // remaining join condition contains all of join condition except window bounds remainingCondition: RexNode, - windowBounds: WindowBounds) + windowBounds: WindowBounds, + earlyFireDelay: java.lang.Long, + earlyFireTimeMode: TimeMode) extends CommonPhysicalJoin(cluster, traitSet, leftRel, rightRel, remainingCondition, joinType) with StreamPhysicalRel { @@ -76,7 +79,9 @@ class StreamPhysicalIntervalJoin( joinType, originalCondition, conditionExpr, - windowBounds) + windowBounds, + earlyFireDelay, + earlyFireTimeMode) } override def explainTerms(pw: RelWriter): RelWriter = { @@ -98,12 +103,16 @@ class StreamPhysicalIntervalJoin( preferExpressionFormat(pw), pw.getDetailLevel)) .item("select", getRowType.getFieldNames.mkString(", ")) + .itemIf("earlyFireDelay", earlyFireDelay, earlyFireDelay != null) + .itemIf("earlyFireTimeMode", earlyFireTimeMode, earlyFireTimeMode != null) } override def translateToExecNode(): ExecNode[_] = { new StreamExecIntervalJoin( unwrapTableConfig(this), new IntervalJoinSpec(joinSpec, windowBounds), + earlyFireDelay, + earlyFireTimeMode, InputProperty.DEFAULT, InputProperty.DEFAULT, FlinkTypeFactory.toLogicalRowType(getRowType), diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java index 447c198eef22ed..88aac40078f8e2 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java @@ -65,6 +65,14 @@ void before() { + " 'connector' = 'values',\n" + " 'bounded' = 'false'\n" + ")"); + util.tableEnv() + .executeSql( + "CREATE TABLE MySink (\n" + + " a INT,\n" + + " b VARCHAR\n" + + ") WITH (\n" + + " 'connector' = 'values'\n" + + ")"); } @Test @@ -142,6 +150,58 @@ void testEarlyFireLowerCaseHintNamePreservesOptions() { verify(sql); } + @Test + void testEarlyFireOnRowTimeLeftOuterJoin() { + String sql = + "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; + verify(sql); + } + + @Test + void testEarlyFireRowTimeOnProcTimeJoin() { + String sql = + "SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='rowtime') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.proctime BETWEEN t2.proctime - INTERVAL '1' HOUR AND t2.proctime + INTERVAL '1' HOUR"; + assertThatThrownBy(() -> verify(sql)) + .hasStackTraceContaining("requires a row-time interval join"); + } + + @Test + void testEarlyFireProcTimeOnRowTimeJoin() { + String sql = + "SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='proctime') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; + assertThatThrownBy(() -> verify(sql)).hasStackTraceContaining("not yet supported"); + } + + @Test + void testEarlyFireOnProcTimeLeftOuterJoin() { + String sql = + "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.proctime BETWEEN t2.proctime - INTERVAL '1' HOUR AND t2.proctime + INTERVAL '1' HOUR"; + verify(sql); + } + + @Test + void testEarlyFireJsonPlanRoundTrip() { + String insert = + "INSERT INTO MySink\n" + + "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; + util.verifyJsonPlan(insert); + } + private void verify(String sql) { util.doVerifyPlan( sql, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml index df5bc8675e72a6..3df60f3cb8d756 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml @@ -38,7 +38,75 @@ LogicalProject(a=[$0], b=[$6]) = (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0]) ++- IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-10000, leftUpperBound=3600000, leftTimeIndex=1, rightTimeIndex=2], where=[((a = a0) AND (rowtime >= (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[ROWTIME]) + :- Exchange(distribution=[hash[a]]) + : +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime]) + +- Exchange(distribution=[hash[a]]) + +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime]) +]]> + + + + + + + + =($3, -($8, 3600000:INTERVAL HOUR)), <=($3, +($8, 3600000:INTERVAL HOUR)))], joinType=[left], joinHints=[[[EARLY_FIRE inheritPath:[0] options:{delay=5s}]]]) + :- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + : +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]]) +]]> + + + = (proctime0 - 3600000:INTERVAL HOUR)) AND (proctime <= (proctime0 + 3600000:INTERVAL HOUR)))], select=[a, proctime, a0, b, proctime0], earlyFireDelay=[5000], earlyFireTimeMode=[PROCTIME]) + :- Exchange(distribution=[hash[a]]) + : +- Calc(select=[a, proctime]) + : +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + : +- Calc(select=[a, PROCTIME() AS proctime, rowtime]) + : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime]) + +- Exchange(distribution=[hash[a]]) + +- Calc(select=[a, b, proctime]) + +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + +- Calc(select=[a, b, PROCTIME() AS proctime, rowtime]) + +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime]) +]]> + + + + + + + + =($4, -($9, 10000:INTERVAL SECOND)), <=($4, +($9, 3600000:INTERVAL HOUR)))], joinType=[left], joinHints=[[[EARLY_FIRE inheritPath:[0] options:{delay=5s}]]]) + :- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + : +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]]) +]]> + + + = (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[ROWTIME]) :- Exchange(distribution=[hash[a]]) : +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime]) diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest_jsonplan/testEarlyFireJsonPlanRoundTrip.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest_jsonplan/testEarlyFireJsonPlanRoundTrip.out new file mode 100644 index 00000000000000..2b98b28751eb25 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest_jsonplan/testEarlyFireJsonPlanRoundTrip.out @@ -0,0 +1,446 @@ +{ + "flinkVersion" : "", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MyTable`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "a", + "dataType" : "INT" + }, { + "name" : "b", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "c", + "dataType" : "BIGINT" + }, { + "name" : "proctime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$PROCTIME$1", + "type" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "PROCTIME" + } + }, + "serializableString" : "PROCTIME()" + } + }, { + "name" : "rowtime", + "dataType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + }, + "options" : { + "bounded" : "false", + "connector" : "values" + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 0 ], [ 3 ] ], + "producedType" : "ROW<`a` INT, `rowtime` TIMESTAMP(3)> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`a` INT, `rowtime` TIMESTAMP(3)> NOT NULL" + } ] + }, + "outputType" : "ROW<`a` INT, `rowtime` TIMESTAMP(3)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime])" + }, { + "id" : 2, + "type" : "stream-exec-watermark-assigner_1", + "watermarkExpr" : { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "TIMESTAMP(3)" + }, + "rowtimeFieldIndex" : 1, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "a", + "fieldType" : "INT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "a", + "fieldType" : "INT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[a]])" + }, { + "id" : 4, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MyTable2`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "a", + "dataType" : "INT" + }, { + "name" : "b", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "c", + "dataType" : "BIGINT" + }, { + "name" : "proctime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$PROCTIME$1", + "type" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "PROCTIME" + } + }, + "serializableString" : "PROCTIME()" + } + }, { + "name" : "rowtime", + "dataType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "rowtime", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 4, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`rowtime`" + } + } ] + }, + "options" : { + "bounded" : "false", + "connector" : "values" + } + } + }, + "abilities" : [ { + "type" : "ProjectPushDown", + "projectedFields" : [ [ 0 ], [ 1 ], [ 3 ] ], + "producedType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)> NOT NULL" + }, { + "type" : "ReadingMetadata", + "metadataKeys" : [ ], + "producedType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)> NOT NULL" + } ] + }, + "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime])" + }, { + "id" : 5, + "type" : "stream-exec-watermark-assigner_1", + "watermarkExpr" : { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "TIMESTAMP(3)" + }, + "rowtimeFieldIndex" : 2, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "a", + "fieldType" : "INT" + }, { + "name" : "b", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "a", + "fieldType" : "INT" + }, { + "name" : "b", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[a]])" + }, { + "id" : 7, + "type" : "stream-exec-interval-join_1", + "intervalJoinSpec" : { + "joinSpec" : { + "joinType" : "LEFT", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "windowBounds" : { + "isEventTime" : true, + "leftLowerBound" : -10000, + "leftUpperBound" : 3600000, + "leftTimeIndex" : 1, + "rightTimeIndex" : 2 + } + }, + "earlyFireDelay" : 5000, + "earlyFireTimeMode" : "ROWTIME", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "a", + "fieldType" : "INT" + }, { + "name" : "rowtime", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, { + "name" : "a0", + "fieldType" : "INT" + }, { + "name" : "b", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "rowtime0", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-10000, leftUpperBound=3600000, leftTimeIndex=1, rightTimeIndex=2], where=[((a = a0) AND (rowtime >= (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[ROWTIME])" + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "VARCHAR(2147483647)" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647)>", + "description" : "Calc(select=[a, b])" + }, { + "id" : 9, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MySink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "a", + "dataType" : "INT" + }, { + "name" : "b", + "dataType" : "VARCHAR(2147483647)" + } ] + }, + "options" : { + "connector" : "values" + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647)>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, b])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 4, + "target" : 5, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file From 0ef163a55b2583ade7d4c7717255c4d6d527cf7e Mon Sep 17 00:00:00 2001 From: weiqingy Date: Sat, 18 Jul 2026 13:00:28 -0700 Subject: [PATCH 2/2] [FLINK-40169][table] Add target option to the EARLY_FIRE hint Add an optional `target` option to the EARLY_FIRE join hint. Only `interval_join` is accepted today; any other value fails planning. An omitted target is equivalent to `interval_join`, so existing hints keep their meaning. `target` scopes the hint to a single operator kind. The interval-join rule consumes the hint only when it targets the interval join and leaves a hint aimed at another operator kind untouched, so an untargeted hint never silently expands its scope. --- .../api/config/EarlyFireJoinHintOptions.java | 13 ++++++++ .../planner/hint/FlinkHintStrategies.java | 8 +++++ .../StreamPhysicalIntervalJoinRule.java | 6 ++++ .../hints/stream/EarlyFireJoinHintTest.java | 21 ++++++++++++ .../hints/stream/EarlyFireJoinHintTest.xml | 32 +++++++++++++++++++ 5 files changed, 80 insertions(+) diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java index 6c919f318f2469..f8ecb172bd7f44 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java @@ -36,6 +36,18 @@ @PublicEvolving public class EarlyFireJoinHintOptions { + /** The only operator kind the EARLY_FIRE hint applies to. */ + public static final String INTERVAL_JOIN = "interval_join"; + + public static final ConfigOption TARGET = + key("target") + .stringType() + .noDefaultValue() + .withDescription( + "The operator kind that the EARLY_FIRE hint applies to. Currently only" + + " 'interval_join' is supported. When omitted, the hint applies" + + " to the interval join."); + public static final ConfigOption DELAY = key("delay") .durationType() @@ -60,6 +72,7 @@ public class EarlyFireJoinHintOptions { static { requiredKeys.add(DELAY); + supportedKeys.add(TARGET); supportedKeys.add(DELAY); supportedKeys.add(TIME_MODE); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java index 204c23a97d84a5..55018dec75fc9e 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java @@ -308,6 +308,14 @@ private static HintOptionChecker fixedSizeListOptionChecker(int size) { "Invalid EARLY_FIRE hint option: {} value should be at least 1 millisecond but was {}", EarlyFireJoinHintOptions.DELAY.key(), delay); + + String target = conf.get(EarlyFireJoinHintOptions.TARGET); + litmus.check( + null == target || EarlyFireJoinHintOptions.INTERVAL_JOIN.equals(target), + "Invalid EARLY_FIRE hint option: {} value '{}' is not supported, only '{}' is supported currently", + EarlyFireJoinHintOptions.TARGET.key(), + target, + EarlyFireJoinHintOptions.INTERVAL_JOIN); return true; }; diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java index d7cbad27d0c5aa..5b848e4f20c2de 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java @@ -170,6 +170,12 @@ private static EarlyFire extractEarlyFire(List hints, boolean isEventTi } Configuration conf = Configuration.fromMap(earlyFireHint.kvOptions); + // target scopes the hint to one operator kind: this rule applies it only when it targets + // the interval join, and leaves a hint aimed at any other operator kind untouched. + String target = conf.get(EarlyFireJoinHintOptions.TARGET); + if (target != null && !EarlyFireJoinHintOptions.INTERVAL_JOIN.equals(target)) { + return new EarlyFire(null, null); + } Duration delay = conf.get(EarlyFireJoinHintOptions.DELAY); TimeMode timeMode = conf.get(EarlyFireJoinHintOptions.TIME_MODE); if (timeMode == null) { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java index 88aac40078f8e2..1db3dc6e8c241f 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java @@ -140,6 +140,17 @@ void testEarlyFireListOptionsRejected() { .hasMessageContaining("only support key-value options"); } + @Test + void testEarlyFireUnsupportedTarget() { + String sql = + "SELECT /*+ EARLY_FIRE('target'='window_join', 'delay'='5s') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; + assertThatThrownBy(() -> verify(sql)) + .hasMessageContaining("target value 'window_join' is not supported"); + } + @Test void testEarlyFireLowerCaseHintNamePreservesOptions() { String sql = @@ -160,6 +171,16 @@ void testEarlyFireOnRowTimeLeftOuterJoin() { verify(sql); } + @Test + void testEarlyFireExplicitTargetIntervalJoin() { + String sql = + "SELECT /*+ EARLY_FIRE('target'='interval_join', 'delay'='5s') */ t1.a, t2.b\n" + + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n" + + " t1.a = t2.a AND\n" + + " t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR"; + verify(sql); + } + @Test void testEarlyFireRowTimeOnProcTimeJoin() { String sql = diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml index 3df60f3cb8d756..8daae577559eeb 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml @@ -16,6 +16,38 @@ See the License for the specific language governing permissions and limitations under the License. --> + + + + + + =($4, -($9, 10000:INTERVAL SECOND)), <=($4, +($9, 3600000:INTERVAL HOUR)))], joinType=[left], joinHints=[[[EARLY_FIRE inheritPath:[0] options:{delay=5s, target=interval_join}]]]) + :- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + : +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]]) + +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4]) + +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3]) + +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]]) +]]> + + + = (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[ROWTIME]) + :- Exchange(distribution=[hash[a]]) + : +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime]) + +- Exchange(distribution=[hash[a]]) + +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime]) + +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime]) +]]> + +