This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 78358442b68 [FLINK-40168][table] Thread the EARLY_FIRE hint into the
interval join
78358442b68 is described below
commit 78358442b68bfc4ee99e02f7f6aaac17e20cc175
Author: weiqingy <[email protected]>
AuthorDate: Sat Jul 18 15:11:00 2026 -0700
[FLINK-40168][table] Thread the EARLY_FIRE hint into the interval join
---
.../nodes/exec/stream/StreamExecIntervalJoin.java | 25 ++
.../stream/StreamPhysicalIntervalJoinRule.java | 59 ++-
.../stream/StreamPhysicalIntervalJoin.scala | 13 +-
.../plan/hints/stream/EarlyFireJoinHintTest.java | 60 +++
.../plan/hints/stream/EarlyFireJoinHintTest.xml | 70 +++-
.../testEarlyFireJsonPlanRoundTrip.out | 446 +++++++++++++++++++++
6 files changed, 669 insertions(+), 4 deletions(-)
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 2ac0af781e0..20b676af785 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.OneInputTransformation;
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.util.Preconditions;
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<RowData>
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 class StreamExecIntervalJoin extends
ExecNodeBase<RowData>
ExecNodeContext.newContext(StreamExecIntervalJoin.class),
ExecNodeContext.newPersistedConfig(StreamExecIntervalJoin.class, tableConfig),
intervalJoinSpec,
+ earlyFireDelay,
+ earlyFireTimeMode,
Lists.newArrayList(leftInputProperty, rightInputProperty),
outputType,
description);
@@ -118,12 +138,17 @@ public class StreamExecIntervalJoin extends
ExecNodeBase<RowData>
@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<InputProperty>
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 6354f04de07..d7cbad27d0c 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.RelOptRule;
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 class StreamPhysicalIntervalJoinRule
RelTraitSet providedTraitSet) {
Tuple2<Option<IntervalJoinSpec.WindowBounds>, Option<RexNode>> 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 class StreamPhysicalIntervalJoinRule
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<RelHint> 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 4916d653b2e..d3401783989 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 447c198eef2..88aac40078f 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 @@ class EarlyFireJoinHintTest extends TableTestBase {
+ " '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 @@ class EarlyFireJoinHintTest extends TableTestBase {
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 df5bc8675e7..3df60f3cb8d 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])
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[a, b])
-+- 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])
++- 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])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEarlyFireOnProcTimeLeftOuterJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b
+FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON
+ t1.a = t2.a AND
+ t1.proctime BETWEEN t2.proctime - INTERVAL '1' HOUR AND t2.proctime +
INTERVAL '1' HOUR]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$6])
++- LogicalJoin(condition=[AND(=($0, $5), >=($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]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[a, b])
++- IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=false,
leftLowerBound=-3600000, leftUpperBound=3600000, leftTimeIndex=1,
rightTimeIndex=2], where=[((a = a0) AND (proctime >= (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])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEarlyFireOnRowTimeLeftOuterJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b
+FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON
+ t1.a = t2.a AND
+ t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime +
INTERVAL '1' HOUR]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$6])
++- LogicalJoin(condition=[AND(=($0, $5), >=($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]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[a, b])
++- 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])
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 00000000000..2b98b28751e
--- /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
