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 1ea8cb0e4e5 [FLINK-40167][table] Add EARLY_FIRE join hint surface and
option validation (#28353)
1ea8cb0e4e5 is described below
commit 1ea8cb0e4e53462d3196cee35eba954930847bb0
Author: Weiqing Yang <[email protected]>
AuthorDate: Fri Jul 24 18:12:25 2026 -0700
[FLINK-40167][table] Add EARLY_FIRE join hint surface and option validation
(#28353)
---
.../docs/util/ConfigurationOptionLocator.java | 3 +-
.../table/api/config/EarlyFireJoinHintOptions.java | 81 +++++++++++
.../planner/hint/CapitalizeQueryHintsShuttle.java | 3 +-
.../table/planner/hint/FlinkHintStrategies.java | 58 ++++++++
.../flink/table/planner/hint/JoinStrategy.java | 13 ++
.../planner/plan/optimize/QueryHintsResolver.java | 7 +
.../planner/plan/hints/batch/JoinHintTestBase.java | 5 +-
.../plan/hints/stream/EarlyFireJoinHintTest.java | 153 +++++++++++++++++++++
.../plan/hints/stream/EarlyFireJoinHintTest.xml | 51 +++++++
9 files changed, 370 insertions(+), 4 deletions(-)
diff --git
a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
index 95dae776d7d..ed2d987be50 100644
---
a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
+++
b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
@@ -105,7 +105,8 @@ public class ConfigurationOptionLocator {
"org.apache.flink.state.rocksdb.PredefinedOptions",
"org.apache.flink.python.PythonConfig",
"org.apache.flink.cep.configuration.SharedBufferCacheConfig",
-
"org.apache.flink.table.api.config.LookupJoinHintOptions"));
+
"org.apache.flink.table.api.config.LookupJoinHintOptions",
+
"org.apache.flink.table.api.config.EarlyFireJoinHintOptions"));
private static final String DEFAULT_PATH_PREFIX = "src/main/java";
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
new file mode 100644
index 00000000000..6c919f318f2
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java
@@ -0,0 +1,81 @@
+/*
+ * 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.api.config;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava33.com.google.common.collect.ImmutableSet;
+
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds hint option name definitions for EARLY_FIRE join hints
based on {@link
+ * org.apache.flink.configuration.ConfigOption}.
+ */
+@PublicEvolving
+public class EarlyFireJoinHintOptions {
+
+ public static final ConfigOption<Duration> DELAY =
+ key("delay")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "The delay between the time an unmatched outer row
becomes eligible to"
+ + " be emitted with null padding and the
time it is actually"
+ + " emitted. Must be at least 1
millisecond.");
+
+ public static final ConfigOption<TimeMode> TIME_MODE =
+ key("time-mode")
+ .enumType(TimeMode.class)
+ .noDefaultValue()
+ .withDescription(
+ "The time domain that drives the early-fire delay,
can be 'rowtime' or"
+ + " 'proctime'. If not set, it defaults to
the time domain of"
+ + " the interval join.");
+
+ private static final Set<ConfigOption<?>> requiredKeys = new HashSet<>();
+ private static final Set<ConfigOption<?>> supportedKeys = new HashSet<>();
+
+ static {
+ requiredKeys.add(DELAY);
+
+ supportedKeys.add(DELAY);
+ supportedKeys.add(TIME_MODE);
+ }
+
+ public static ImmutableSet<ConfigOption> getRequiredOptions() {
+ return ImmutableSet.copyOf(requiredKeys);
+ }
+
+ public static ImmutableSet<ConfigOption> getSupportedOptions() {
+ return ImmutableSet.copyOf(supportedKeys);
+ }
+
+ /** The time domain that drives the early-fire delay. */
+ @PublicEvolving
+ public enum TimeMode {
+ ROWTIME,
+ PROCTIME
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/CapitalizeQueryHintsShuttle.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/CapitalizeQueryHintsShuttle.java
index 006dab4b832..6da4452010d 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/CapitalizeQueryHintsShuttle.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/CapitalizeQueryHintsShuttle.java
@@ -46,7 +46,8 @@ public class CapitalizeQueryHintsShuttle extends
QueryHintsRelShuttle {
changed.set(true);
if
(JoinStrategy.isJoinStrategy(capitalHintName)) {
- if
(JoinStrategy.isLookupHint(hint.hintName)) {
+ if
(JoinStrategy.isLookupHint(hint.hintName)
+ ||
JoinStrategy.isEarlyFireHint(hint.hintName)) {
return
RelHint.builder(capitalHintName)
.hintOptions(hint.kvOptions)
.inheritPath(hint.inheritPath)
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 5978f98017d..204c23a97d8 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
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.hint;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.config.EarlyFireJoinHintOptions;
import org.apache.flink.table.api.config.LookupJoinHintOptions;
import org.apache.flink.table.factories.FactoryUtil;
import
org.apache.flink.table.planner.plan.rules.logical.WrapJsonAggFunctionArgumentsRule;
@@ -36,6 +37,9 @@ import org.apache.calcite.util.Litmus;
import java.time.Duration;
import java.util.Collections;
import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
/**
* A collection of Flink style {@link HintStrategy}s.
@@ -135,6 +139,11 @@ public abstract class FlinkHintStrategies {
HintPredicates.JOIN,
HintPredicates.AGGREGATE))
.optionChecker(STATE_TTL_NON_EMPTY_KV_OPTION_CHECKER)
.build())
+ .hintStrategy(
+ JoinStrategy.EARLY_FIRE.getJoinHintName(),
+ HintStrategy.builder(HintPredicates.JOIN)
+ .optionChecker(EARLY_FIRE_KV_OPTION_CHECKER)
+ .build())
.build();
}
@@ -253,6 +262,55 @@ public abstract class FlinkHintStrategies {
return true;
};
+ private static final HintOptionChecker EARLY_FIRE_KV_OPTION_CHECKER =
+ (earlyFireHint, litmus) -> {
+ litmus.check(
+ earlyFireHint.listOptions.size() == 0,
+ "Invalid list options in EARLY_FIRE hint, only support
key-value options.");
+
+ Configuration conf =
Configuration.fromMap(earlyFireHint.kvOptions);
+ ImmutableSet<ConfigOption> requiredKeys =
+ EarlyFireJoinHintOptions.getRequiredOptions();
+ litmus.check(
+ requiredKeys.stream().allMatch(conf::contains),
+ "Invalid EARLY_FIRE hint: incomplete required
option(s): {}",
+ requiredKeys);
+
+ ImmutableSet<ConfigOption> supportedKeys =
+ EarlyFireJoinHintOptions.getSupportedOptions();
+ Set<String> supportedKeyNames =
+
supportedKeys.stream().map(ConfigOption::key).collect(Collectors.toSet());
+ Set<String> unknownKeys =
+ earlyFireHint.kvOptions.keySet().stream()
+ .filter(key ->
!supportedKeyNames.contains(key))
+
.collect(Collectors.toCollection(TreeSet::new));
+ litmus.check(
+ unknownKeys.isEmpty(),
+ "Unsupported EARLY_FIRE hint option(s) {}, supported
options are {}.",
+ unknownKeys,
+ new TreeSet<>(supportedKeyNames));
+ litmus.check(
+ earlyFireHint.kvOptions.size() <= supportedKeys.size(),
+ "Too many EARLY_FIRE hint options {} beyond max number
of supported options {}",
+ earlyFireHint.kvOptions.size(),
+ supportedKeys.size());
+
+ try {
+ // try to validate all hint options by parsing them
+ supportedKeys.forEach(conf::get);
+ } catch (IllegalArgumentException e) {
+ litmus.fail("Invalid EARLY_FIRE hint options: {}",
e.getMessage());
+ }
+
+ Duration delay = conf.get(EarlyFireJoinHintOptions.DELAY);
+ litmus.check(
+ null != delay && delay.toMillis() > 0,
+ "Invalid EARLY_FIRE hint option: {} value should be at
least 1 millisecond but was {}",
+ EarlyFireJoinHintOptions.DELAY.key(),
+ delay);
+ return true;
+ };
+
private static final HintOptionChecker
STATE_TTL_NON_EMPTY_KV_OPTION_CHECKER =
(ttlHint, litmus) -> {
litmus.check(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/JoinStrategy.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/JoinStrategy.java
index 0ded47cb222..d9e3acff7a2 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/JoinStrategy.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/JoinStrategy.java
@@ -50,6 +50,12 @@ public enum JoinStrategy {
/** Instructs the optimizer to use lookup join strategy. Only accept
key-value hint options. */
LOOKUP("LOOKUP"),
+ /**
+ * Instructs an outer interval join to emit unmatched outer rows with null
padding after a
+ * configurable delay. Only accept key-value hint options.
+ */
+ EARLY_FIRE("EARLY_FIRE"),
+
/**
* Instructs the optimizer to use multi-way join strategy for streaming
queries. This hint
* allows specifying multiple tables to be joined together in a single
{@link
@@ -89,6 +95,7 @@ public enum JoinStrategy {
case NEST_LOOP:
return options.size() > 0;
case LOOKUP:
+ case EARLY_FIRE:
return null == options || options.size() == 0;
case MULTI_JOIN:
return options.size() > 0;
@@ -101,4 +108,10 @@ public enum JoinStrategy {
return isJoinStrategy(formalizedHintName)
&& JoinStrategy.valueOf(formalizedHintName) == LOOKUP;
}
+
+ public static boolean isEarlyFireHint(String hintName) {
+ String formalizedHintName = hintName.toUpperCase(Locale.ROOT);
+ return isJoinStrategy(formalizedHintName)
+ && JoinStrategy.valueOf(formalizedHintName) == EARLY_FIRE;
+ }
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/QueryHintsResolver.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/QueryHintsResolver.java
index b288f111e30..b153ef2b885 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/QueryHintsResolver.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/QueryHintsResolver.java
@@ -146,6 +146,13 @@ public class QueryHintsResolver extends
QueryHintsRelShuttle {
updateInfoForOptionCheck(hint.hintName, rightName);
newHints.add(hint);
}
+ } else if (JoinStrategy.isEarlyFireHint(hint.hintName)) {
+ // EARLY_FIRE carries only key-value options and is not bound
to a specific input
+ // side, so it is passed through unchanged once its options
are validated by the
+ // hint option checker.
+ allHints.add(trimInheritPath(hint));
+ validHints.add(trimInheritPath(hint));
+ newHints.add(hint);
} else if (JoinStrategy.isJoinStrategy(hint.hintName)) {
allHints.add(trimInheritPath(hint));
// add options about this hint for finally checking
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/batch/JoinHintTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/batch/JoinHintTestBase.java
index 24ae0e4025d..217e3c2b8af 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/batch/JoinHintTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/batch/JoinHintTestBase.java
@@ -61,8 +61,9 @@ public abstract class JoinHintTestBase extends TableTestBase {
private final List<String> allJoinHintNames =
Lists.newArrayList(JoinStrategy.values()).stream()
- // LOOKUP hint has different kv-options against other join
hints
- .filter(hint -> hint != JoinStrategy.LOOKUP)
+ // LOOKUP and EARLY_FIRE hints only support key-value
options, unlike the
+ // list-option join hints exercised here
+ .filter(hint -> hint != JoinStrategy.LOOKUP && hint !=
JoinStrategy.EARLY_FIRE)
.map(JoinStrategy::getJoinHintName)
.collect(Collectors.toList());
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
new file mode 100644
index 00000000000..447c198eef2
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.plan.hints.stream;
+
+import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.PlanKind;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import scala.Enumeration;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for the EARLY_FIRE join hint surface and option validation. */
+class EarlyFireJoinHintTest extends TableTestBase {
+
+ protected StreamTableTestUtil util;
+
+ @BeforeEach
+ void before() {
+ util = streamTestUtil(TableConfig.getDefault());
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE MyTable (\n"
+ + " a INT,\n"
+ + " b VARCHAR,\n"
+ + " c BIGINT,\n"
+ + " proctime AS PROCTIME(),\n"
+ + " rowtime TIMESTAMP(3),\n"
+ + " WATERMARK FOR rowtime AS rowtime\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false'\n"
+ + ")");
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE MyTable2 (\n"
+ + " a INT,\n"
+ + " b VARCHAR,\n"
+ + " c BIGINT,\n"
+ + " proctime AS PROCTIME(),\n"
+ + " rowtime TIMESTAMP(3),\n"
+ + " WATERMARK FOR rowtime AS rowtime\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false'\n"
+ + ")");
+ }
+
+ @Test
+ void testEarlyFireMissingDelay() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('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.rowtime BETWEEN t2.rowtime - INTERVAL '10'
SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+ assertThatThrownBy(() -> verify(sql)).hasMessageContaining("incomplete
required option(s)");
+ }
+
+ @Test
+ void testEarlyFireNonPositiveDelay() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('delay'='0s') */ 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("value should be at least 1
millisecond");
+ }
+
+ @Test
+ void testEarlyFireSubMillisecondDelay() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('delay'='1ns') */ 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("value should be at least 1
millisecond");
+ }
+
+ @Test
+ void testEarlyFireInvalidTimeMode() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='unknown') */
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("Invalid EARLY_FIRE hint options");
+ }
+
+ @Test
+ void testEarlyFireUnknownOption() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('delay'='5s', 'timemode'='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))
+ .hasMessageContaining("Unsupported EARLY_FIRE hint option(s)
[timemode]");
+ }
+
+ @Test
+ void testEarlyFireListOptionsRejected() {
+ String sql =
+ "SELECT /*+ EARLY_FIRE('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("only support key-value options");
+ }
+
+ @Test
+ void testEarlyFireLowerCaseHintNamePreservesOptions() {
+ 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.rowtime BETWEEN t2.rowtime - INTERVAL '10'
SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+ verify(sql);
+ }
+
+ private void verify(String sql) {
+ util.doVerifyPlan(
+ sql,
+ new ExplainDetail[] {},
+ false,
+ new Enumeration.Value[] {PlanKind.AST(), PlanKind.OPT_EXEC()},
+ false);
+ }
+}
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
new file mode 100644
index 00000000000..df5bc8675e7
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.xml
@@ -0,0 +1,51 @@
+<?xml version="1.0" ?>
+<!--
+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.
+-->
+<Root>
+ <TestCase name="testEarlyFireLowerCaseHintNamePreservesOptions">
+ <Resource name="sql">
+ <![CDATA[SELECT /*+ early_fire('delay'='5s', 'time-mode'='rowtime') */
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,
time-mode=rowtime}]]])
+ :- 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])
+ :- 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>
+</Root>
