This is an automated email from the ASF dual-hosted git repository.
fhueske 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 8fbca66d516 [FLINK-39782][table] Add SNAPSHOT built-in function
definition and type inference (#28396)
8fbca66d516 is described below
commit 8fbca66d516e220d7174255ce8ff1760ffa33644
Author: Fabian Hueske <[email protected]>
AuthorDate: Mon Jun 22 12:46:09 2026 +0200
[FLINK-39782][table] Add SNAPSHOT built-in function definition and type
inference (#28396)
* Add SNAPSHOT function definition.
* Function is added without a runtime implementation.
* We will later add planner rules to rewrite it to the
LateralSnapshotJoinOperator..
* Provide input and output type strategies for function
* Add tests for SNAPSHOT input and output type strategies
* Add plan and execution tests using the SNAPSHOT function
* Execution tests are failing as expected because we don't provide an
implementation for the function.
Generated-by: Claude Opus 4.8 (1M context)
---
.../functions/BuiltInFunctionDefinitions.java | 42 ++++
.../strategies/LateralSnapshotTypeStrategy.java | 234 +++++++++++++++++++++
.../strategies/SpecificInputTypeStrategies.java | 4 +
.../strategies/SpecificTypeStrategies.java | 4 +
.../LateralSnapshotInputTypeStrategyTest.java | 177 ++++++++++++++++
.../LateralSnapshotOutputTypeStrategyTest.java | 117 +++++++++++
.../plan/stream/sql/SnapshotTableFunctionTest.java | 171 +++++++++++++++
.../plan/stream/sql/SnapshotTableFunctionTest.xml | 73 +++++++
8 files changed, 822 insertions(+)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 1d842767252..0828f2bb8b4 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -112,6 +112,7 @@ import static
org.apache.flink.table.types.inference.strategies.SpecificInputTyp
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.FROM_CHANGELOG_INPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.INDEX;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.JSON_ARGUMENT;
+import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ML_PREDICT_INPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.TO_CHANGELOG_INPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.TWO_EQUALS_COMPARABLE;
@@ -120,6 +121,7 @@ import static
org.apache.flink.table.types.inference.strategies.SpecificInputTyp
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.percentageArray;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.ARRAY_APPEND_PREPEND;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY;
+import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.ML_PREDICT_OUTPUT_TYPE_STRATEGY;
import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.TO_CHANGELOG_OUTPUT_TYPE_STRATEGY;
@@ -908,6 +910,46 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.ptf.FromChangelogFunction")
.build();
+ /**
+ * Built-in proxy function for the LATERAL SNAPSHOT temporal join.
+ *
+ * <p>The function itself has no runtime — it is a planner placeholder. A
dedicated optimizer
+ * rule recognizes calls of this function inside a {@code LATERAL} context
and rewrites the
+ * surrounding correlate/join into a specialized stream operator that
joins probe-side records
+ * against an updating temporal build-side table.
+ */
+ public static final BuiltInFunctionDefinition SNAPSHOT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("SNAPSHOT")
+ .kind(PROCESS_TABLE)
+ .staticArguments(
+ StaticArgument.table(
+ "input",
+ Row.class,
+ false,
+ EnumSet.of(
+ StaticArgumentTrait.TABLE,
+
StaticArgumentTrait.ROW_SEMANTIC_TABLE,
+
StaticArgumentTrait.SUPPORT_UPDATES,
+
StaticArgumentTrait.REQUIRE_UPDATE_BEFORE,
+
StaticArgumentTrait.REQUIRE_FULL_DELETE)),
+ StaticArgument.scalar(
+ "load_completed_condition",
DataTypes.STRING(), true),
+ StaticArgument.scalar(
+ "load_completed_time",
DataTypes.TIMESTAMP_LTZ(3), true),
+ StaticArgument.scalar(
+ "load_completed_idle_timeout",
+ DataTypes.INTERVAL(DataTypes.SECOND()),
+ true),
+ StaticArgument.scalar(
+ "state_ttl",
DataTypes.INTERVAL(DataTypes.SECOND()), true))
+ .inputTypeStrategy(LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .outputTypeStrategy(LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY)
+ .runtimeProvided()
+ // TODO: disableSystemArguments(true), once we have a
dedicated translation rule
+ .notDeterministic()
+ .build();
+
public static final BuiltInFunctionDefinition GREATEST =
BuiltInFunctionDefinition.newBuilder()
.name("GREATEST")
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
new file mode 100644
index 00000000000..8ec2d4899c9
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
@@ -0,0 +1,234 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Type strategies for the {@code SNAPSHOT} table function used by the {@code
LATERAL SNAPSHOT}
+ * temporal join.
+ *
+ * <p>Validates the named arguments
+ *
+ * <ul>
+ * <li>{@code input} (TABLE, required)
+ * <li>{@code load_completed_condition} (STRING literal, optional, default
{@code 'compile_time'},
+ * allowed values: {@code 'compile_time'}, {@code 'user_time'})
+ * <li>{@code load_completed_time} (TIMESTAMP_LTZ(3), optional)
+ * <li>{@code load_completed_idle_timeout} (INTERVAL SECOND, optional)
+ * <li>{@code state_ttl} (INTERVAL SECOND, optional)
+ * </ul>
+ *
+ * <p>and ensures cross-argument consistency:
+ *
+ * <ul>
+ * <li>{@code load_completed_condition='user_time'} requires {@code
load_completed_time}.
+ * <li>{@code load_completed_condition='compile_time'} (or unset) forbids
{@code
+ * load_completed_time}.
+ * </ul>
+ *
+ * <p>The output type forwards the input table's row type, but materializes
any rowtime attribute
+ * indicator into a regular timestamp.
+ */
+@Internal
+public final class LateralSnapshotTypeStrategy {
+
+ /** Argument index of the {@code input} TABLE. */
+ public static final int INPUT_ARG_INDEX = 0;
+
+ /** Argument index of the {@code load_completed_condition} STRING. */
+ public static final int LOAD_COMPLETED_CONDITION_ARG_INDEX = 1;
+
+ /** Argument index of the {@code load_completed_time} TIMESTAMP_LTZ. */
+ public static final int LOAD_COMPLETED_TIME_ARG_INDEX = 2;
+
+ /** Argument index of the {@code load_completed_idle_timeout} INTERVAL. */
+ public static final int LOAD_COMPLETED_IDLE_TIMEOUT_ARG_INDEX = 3;
+
+ /** Argument index of the {@code state_ttl} INTERVAL. */
+ public static final int STATE_TTL_ARG_INDEX = 4;
+
+ /** Default value for {@code load_completed_condition}. */
+ public static final String LOAD_COMPLETED_CONDITION_COMPILE_TIME =
"compile_time";
+
+ /**
+ * Allowed value for {@code load_completed_condition} that requires {@code
load_completed_time}.
+ */
+ public static final String LOAD_COMPLETED_CONDITION_USER_TIME =
"user_time";
+
+ private static final Set<String> VALID_LOAD_COMPLETED_CONDITIONS =
+ Set.of(LOAD_COMPLETED_CONDITION_COMPILE_TIME,
LOAD_COMPLETED_CONDITION_USER_TIME);
+
+ /** Stable, human-readable rendering of {@link
#VALID_LOAD_COMPLETED_CONDITIONS}. */
+ private static final String VALID_LOAD_COMPLETED_CONDITIONS_DESC =
+ String.format(
+ "'%s', '%s'",
+ LOAD_COMPLETED_CONDITION_COMPILE_TIME,
LOAD_COMPLETED_CONDITION_USER_TIME);
+
+ //
--------------------------------------------------------------------------------------------
+ // Input validation
+ //
--------------------------------------------------------------------------------------------
+
+ public static final InputTypeStrategy INPUT_TYPE_STRATEGY =
+ new InputTypeStrategy() {
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.between(1, 5);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean
throwOnFailure) {
+ return validateInputs(callContext, throwOnFailure);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(final
FunctionDefinition definition) {
+ return List.of(
+ Signature.of(
+ Argument.of("input", "TABLE"),
+ Argument.of("load_completed_condition",
"STRING"),
+ Argument.of("load_completed_time",
"TIMESTAMP_LTZ(3)"),
+ Argument.of("load_completed_idle_timeout",
"INTERVAL SECOND"),
+ Argument.of("state_ttl", "INTERVAL
SECOND")));
+ }
+ };
+
+ //
--------------------------------------------------------------------------------------------
+ // Output type inference: forward the input table row type with time
attributes materialized.
+ //
--------------------------------------------------------------------------------------------
+
+ public static final TypeStrategy OUTPUT_TYPE_STRATEGY =
+ callContext -> {
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(INPUT_ARG_INDEX)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "Argument 'input' of
SNAPSHOT must be a table."));
+ return
Optional.of(materializeTimeAttributes(semantics.dataType()));
+ };
+
+ //
--------------------------------------------------------------------------------------------
+ // Helpers
+ //
--------------------------------------------------------------------------------------------
+
+ /**
+ * Rebuilds {@code inputTableType} as a ROW with all fields identical to
the input, except for
+ * time attributes which get stripped off their time indicator property.
+ */
+ private static DataType materializeTimeAttributes(final DataType
inputTableType) {
+ final List<DataType> fieldTypes =
DataType.getFieldDataTypes(inputTableType);
+ final List<String> fieldNames = DataType.getFieldNames(inputTableType);
+ final List<Field> fields =
+ IntStream.range(0, fieldTypes.size())
+ .mapToObj(
+ pos ->
+ DataTypes.FIELD(
+ fieldNames.get(pos),
+
DataTypeUtils.removeTimeAttribute(
+ fieldTypes.get(pos))))
+ .collect(Collectors.toList());
+ return DataTypes.ROW(fields);
+ }
+
+ private static Optional<List<DataType>> validateInputs(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (callContext.getTableSemantics(INPUT_ARG_INDEX).isEmpty()) {
+ return callContext.fail(
+ throwOnFailure, "Argument 'input' of SNAPSHOT must be a
table.");
+ }
+
+ // Reject non-literal load_completed_condition explicitly: the planner
needs the value
+ // at compile time to decide between 'compile_time' and 'user_time'.
+ final boolean hasLoadCompletedCondition =
+ isArgumentProvided(callContext,
LOAD_COMPLETED_CONDITION_ARG_INDEX);
+ if (isProvidedNonLiteral(callContext,
LOAD_COMPLETED_CONDITION_ARG_INDEX)) {
+ return callContext.fail(
+ throwOnFailure,
+ "Argument 'load_completed_condition' of SNAPSHOT must be a
STRING literal.");
+ }
+ // Get condition and default to 'compile_time' if not provided
+ final String condition =
+ hasLoadCompletedCondition
+ ? callContext
+
.getArgumentValue(LOAD_COMPLETED_CONDITION_ARG_INDEX, String.class)
+ .orElse(LOAD_COMPLETED_CONDITION_COMPILE_TIME)
+ : LOAD_COMPLETED_CONDITION_COMPILE_TIME;
+ // Reject invalid condition value
+ if (!VALID_LOAD_COMPLETED_CONDITIONS.contains(condition)) {
+ return callContext.fail(
+ throwOnFailure,
+ "Argument 'load_completed_condition' of SNAPSHOT must be
one of %s but was '%s'.",
+ VALID_LOAD_COMPLETED_CONDITIONS_DESC,
+ condition);
+ }
+
+ final boolean hasLoadCompletedTime =
+ isArgumentProvided(callContext, LOAD_COMPLETED_TIME_ARG_INDEX);
+
+ // Cross-argument consistency: condition <-> load_completed_time
+ if (LOAD_COMPLETED_CONDITION_USER_TIME.equals(condition) &&
!hasLoadCompletedTime) {
+ return callContext.fail(
+ throwOnFailure,
+ "SNAPSHOT requires 'load_completed_time' when "
+ + "'load_completed_condition' is 'user_time'.");
+ }
+ if (!LOAD_COMPLETED_CONDITION_USER_TIME.equals(condition) &&
hasLoadCompletedTime) {
+ return callContext.fail(
+ throwOnFailure,
+ "SNAPSHOT does not accept 'load_completed_time' when "
+ + "'load_completed_condition' is not
'user_time'.");
+ }
+
+ return Optional.of(callContext.getArgumentDataTypes());
+ }
+
+ private static boolean isArgumentProvided(final CallContext callContext,
final int index) {
+ return callContext.getArgumentDataTypes().size() > index
+ && !callContext.isArgumentNull(index);
+ }
+
+ private static boolean isProvidedNonLiteral(final CallContext callContext,
final int index) {
+ return isArgumentProvided(callContext, index) &&
!callContext.isArgumentLiteral(index);
+ }
+
+ private LateralSnapshotTypeStrategy() {}
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index dc7167a656d..d551ca89a55 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -130,6 +130,10 @@ public final class SpecificInputTypeStrategies {
public static final InputTypeStrategy FROM_CHANGELOG_INPUT_TYPE_STRATEGY =
FromChangelogTypeStrategy.INPUT_TYPE_STRATEGY;
+ /** Input strategy for {@link BuiltInFunctionDefinitions#SNAPSHOT}. */
+ public static final InputTypeStrategy LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY
=
+ LateralSnapshotTypeStrategy.INPUT_TYPE_STRATEGY;
+
/** See {@link ExtractInputTypeStrategy}. */
public static final InputTypeStrategy EXTRACT = new
ExtractInputTypeStrategy();
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
index 06b6cbb2d3b..77712114442 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
@@ -206,6 +206,10 @@ public final class SpecificTypeStrategies {
public static final TypeStrategy FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY =
FromChangelogTypeStrategy.OUTPUT_TYPE_STRATEGY;
+ /** Type strategy specific for {@link
BuiltInFunctionDefinitions#SNAPSHOT}. */
+ public static final TypeStrategy LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY =
+ LateralSnapshotTypeStrategy.OUTPUT_TYPE_STRATEGY;
+
private SpecificTypeStrategies() {
// no instantiation
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
new file mode 100644
index 00000000000..05012a3b277
--- /dev/null
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY;
+
+/**
+ * Tests for {@link
SpecificInputTypeStrategies#LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY}.
+ *
+ * <p>Validates the named-argument signature of the {@code SNAPSHOT} table
function, including the
+ * cross-argument consistency between {@code load_completed_condition} and
{@code
+ * load_completed_time}.
+ */
+class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase
{
+
+ private static final DataType TABLE_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.STRING()),
+ DataTypes.FIELD("v", DataTypes.INT()));
+
+ private static final DataType STRING_TYPE = DataTypes.STRING();
+ private static final DataType TIMESTAMP_TYPE = DataTypes.TIMESTAMP(3);
+ private static final DataType INTERVAL_TYPE =
DataTypes.INTERVAL(DataTypes.SECOND());
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ //
----------------------------------------------------------------------------
+ // Valid: just the build-side table.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Valid: input only (default condition)",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .expectArgumentTypes(TABLE_TYPE),
+
+ //
----------------------------------------------------------------------------
+ // Valid: explicit 'compile_time' condition without
load_completed_time.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Valid: condition='compile_time'",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "compile_time")
+ .expectArgumentTypes(TABLE_TYPE, STRING_TYPE),
+
+ //
----------------------------------------------------------------------------
+ // Valid: 'user_time' with a TIMESTAMP literal.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Valid: condition='user_time' +
load_completed_time",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "user_time")
+ .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
+ .expectArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_TYPE),
+
+ //
----------------------------------------------------------------------------
+ // Valid: full named-arg form with idle timeout and TTL.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy("Valid: full args",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(
+ TABLE_TYPE,
+ STRING_TYPE,
+ TIMESTAMP_TYPE,
+ INTERVAL_TYPE,
+ INTERVAL_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "user_time")
+ .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
+ .calledWithLiteralAt(3, Duration.ofSeconds(10))
+ .calledWithLiteralAt(4, Duration.ofDays(1))
+ .expectArgumentTypes(
+ TABLE_TYPE,
+ STRING_TYPE,
+ TIMESTAMP_TYPE,
+ INTERVAL_TYPE,
+ INTERVAL_TYPE),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: No arguments
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy("Invalid: no arguments",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes()
+ // Intentionally no arguments.
+ .expectErrorMessage("Invalid function call"),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: 'input' argument is not a table.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: input is not a table",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(STRING_TYPE)
+ // Intentionally no table type registered at position
0.
+ .expectErrorMessage("Argument 'input' of SNAPSHOT must
be a table."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: 'user_time' condition requires load_completed_time.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: condition='user_time' without
load_completed_time",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "user_time")
+ .expectErrorMessage(
+ "SNAPSHOT requires 'load_completed_time' when "
+ + "'load_completed_condition' is
'user_time'."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: load_completed_time requires 'user_time' condition.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: load_completed_time without explicit
'user_time'",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "compile_time")
+ .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
+ .expectErrorMessage(
+ "SNAPSHOT does not accept
'load_completed_time' when "
+ + "'load_completed_condition' is not
'user_time'."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: unknown condition value.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: unknown condition value",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, "invalid_condition")
+ .expectErrorMessage(
+ "Argument 'load_completed_condition' of
SNAPSHOT must be one of 'compile_time', 'user_time' but was
'invalid_condition'."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: load_completed_condition provided as a non-literal
expression.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: non-literal
load_completed_condition",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ // Intentionally no literal provided for
load_completed_condition
+ .expectErrorMessage(
+ "Argument 'load_completed_condition' of
SNAPSHOT must be a STRING literal."));
+ }
+}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotOutputTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotOutputTypeStrategyTest.java
new file mode 100644
index 00000000000..63a6df67be9
--- /dev/null
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotOutputTypeStrategyTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.TypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.utils.CallContextMock;
+import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+import org.apache.flink.table.types.logical.TimestampKind;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.table.types.utils.TypeConversions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.table.types.inference.strategies.LateralSnapshotTypeStrategy.INPUT_ARG_INDEX;
+import static
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link LateralSnapshotTypeStrategy#OUTPUT_TYPE_STRATEGY}.
+ *
+ * <p>The output type of the {@code SNAPSHOT} table function is the row type
of its {@code input}
+ * table argument, with any time-attribute indicators materialized into
regular timestamps.
+ */
+class LateralSnapshotOutputTypeStrategyTest extends TypeStrategiesTestBase {
+
+ // Input row exercising forwarding of field names, assorted types and both
nullabilities, a
+ // regular TIMESTAMP column (must be left untouched) and a
rowtime-attribute column.
+ private static final DataType INPUT_TABLE_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("currency", DataTypes.STRING().notNull()),
+ DataTypes.FIELD("rate", DataTypes.DECIMAL(10, 2)),
+ DataTypes.FIELD("valid_from", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD(
+ "rowtime",
+ TypeConversions.fromLogicalToDataType(
+ new TimestampType(true,
TimestampKind.ROWTIME, 3))));
+
+ private static final DataType OUTPUT_TABLE_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("currency", DataTypes.STRING().notNull()),
+ DataTypes.FIELD("rate", DataTypes.DECIMAL(10, 2)),
+ DataTypes.FIELD("valid_from", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("rowtime", DataTypes.TIMESTAMP(3)));
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ //
----------------------------------------------------------------------------
+ // Output is derived from the table semantics, forwards every
column unchanged
+ // (except for time-attributes which are materialized) and
ignores trailing scalar
+ // arguments.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Input table type is forwarded, trailing
scalar args do not affect output type",
+ LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY)
+ .inputTypes(INPUT_TABLE_TYPE, DataTypes.STRING(),
DataTypes.TIMESTAMP(3))
+ .calledWithTableSemanticsAt(
+ INPUT_ARG_INDEX, new
TableSemanticsMock(INPUT_TABLE_TYPE))
+ .expectDataType(OUTPUT_TABLE_TYPE),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: 'input' argument is not a table.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: input is not a table",
+ LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY)
+ .inputTypes(DataTypes.STRING())
+ // Intentionally no table semantics registered at the
input position.
+ .expectErrorMessage("Argument 'input' of SNAPSHOT must
be a table."));
+ }
+
+ /**
+ * Verifies that the output strips time-attribute indicators. Logical-type
equality ignores the
+ * time-attribute kind (see {@link TimestampType#equals}), so {@code
expectDataType} in {@link
+ * #testData()} cannot detect it; the kind is asserted explicitly here.
+ */
+ @Test
+ void stripsTimeAttributeIndicators() {
+ final CallContextMock callContext = new CallContextMock();
+ callContext.argumentDataTypes = List.of(INPUT_TABLE_TYPE);
+ callContext.tableSemantics =
+ Map.of(INPUT_ARG_INDEX, new
TableSemanticsMock(INPUT_TABLE_TYPE));
+
+ // Sanity: the input must actually carry a time attribute, otherwise
this test is vacuous.
+
assertThat(LogicalTypeChecks.getFieldTypes(INPUT_TABLE_TYPE.getLogicalType()))
+ .anyMatch(LogicalTypeChecks::isTimeAttribute);
+
+ final DataType output =
+
LATERAL_SNAPSHOT_OUTPUT_TYPE_STRATEGY.inferType(callContext).orElseThrow();
+
+ assertThat(LogicalTypeChecks.getFieldTypes(output.getLogicalType()))
+ .noneMatch(LogicalTypeChecks::isTimeAttribute);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
new file mode 100644
index 00000000000..b40ad42be4e
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.stream.sql;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Plan tests for the SNAPSHOT built-in process table function used by the
LATERAL SNAPSHOT temporal
+ * join (FLIP-579).
+ *
+ * <p>SNAPSHOT is a planner placeholder without a runtime implementation.
These tests therefore only
+ * verify that a call parses, that its named arguments pass type inference,
and that it survives
+ * logical optimization in both a plain {@code FROM} clause and a {@code
LATERAL} context. Verifying
+ * the optimized rel plan (rather than the exec plan) stops short of the
runtime translation that a
+ * future optimizer rule will provide by rewriting the call into a dedicated
temporal-join operator.
+ */
+public class SnapshotTableFunctionTest extends TableTestBase {
+
+ private TableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ // Probe side of the temporal join.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE Orders ("
+ + " order_id INT,"
+ + " currency STRING,"
+ + " amount INT,"
+ + " order_time TIMESTAMP(3),"
+ + " WATERMARK FOR order_time AS order_time"
+ + ") WITH ('connector' = 'datagen')");
+ // Build side: the updating dimension table that SNAPSHOT takes as its
table argument.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE Rates ("
+ + " currency STRING,"
+ + " rate INT,"
+ + " rate_time TIMESTAMP(3),"
+ + " WATERMARK FOR rate_time AS rate_time"
+ + ") WITH ('connector' = 'values')");
+ // Sinks used by the execution tests below.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE RatesSink ("
+ + " currency STRING,"
+ + " rate INT,"
+ + " rate_time TIMESTAMP(3)"
+ + ") WITH ('connector' = 'blackhole')");
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE JoinSink ("
+ + " order_id INT,"
+ + " amount INT,"
+ + " rate INT"
+ + ") WITH ('connector' = 'blackhole')");
+ }
+
+ @Test
+ void testFromContext() {
+ // SNAPSHOT used as a standalone table function with the full set of
named arguments
+ util.verifyRelPlan(
+ "SELECT * FROM SNAPSHOT("
+ + "input => TABLE Rates, "
+ + "load_completed_condition => 'user_time', "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00.001' AS TIMESTAMP_LTZ(3)), "
+ + "load_completed_idle_timeout => INTERVAL '10'
SECOND, "
+ + "state_ttl => INTERVAL '1' DAY)");
+ }
+
+ @Test
+ void testLateralContext() {
+ // SNAPSHOT used in a LATERAL context
+ util.verifyRelPlan(
+ "SELECT o.order_id, o.amount, r.rate "
+ + "FROM Orders AS o, LATERAL TABLE(SNAPSHOT(input =>
TABLE Rates)) AS r "
+ + "WHERE o.currency = r.currency");
+ }
+
+ @Test
+ @Disabled(
+ "SNAPSHOT should disable the implicit system arguments (on_time,
uid), but that is not "
+ + "wired up yet. disableSystemArguments(true) is only
legal for a PTF that is "
+ + "rewritten by its own optimizer rule before reaching "
+ + "StreamPhysicalProcessTableFunctionRule (which otherwise
rejects it with "
+ + "'Disabling system arguments is not supported for
user-defined PTF').")
+ void testSystemArgumentsNotAllowed() {
+ // SNAPSHOT must disable the implicit system arguments (e.g.
`on_time`). Passing one must be
+ // rejected because the argument is not allowed.
+ assertThatThrownBy(
+ () ->
+ util.verifyRelPlan(
+ "SELECT * FROM SNAPSHOT("
+ + "input => TABLE Rates, "
+ + "on_time =>
DESCRIPTOR(rate_time))"))
+ .satisfies(anyCauseMatches("on_time"));
+ }
+
+ @Test
+ void testPartitionByNotAllowed() {
+ // The table argument uses row semantics, so it does not accept a
PARTITION BY clause.
+ assertThatThrownBy(
+ () ->
+ util.verifyRelPlan(
+ "SELECT * FROM SNAPSHOT("
+ + "input => TABLE Rates
PARTITION BY currency)"))
+ .satisfies(
+ anyCauseMatches(
+ "Only tables with set semantics may be
partitioned. Invalid PARTITION BY clause in the 0-th operand of table function
'SNAPSHOT'"));
+ }
+
+ @Test
+ void testFromContextExecutionFails() {
+ // SNAPSHOT has no runtime implementation yet, so compiling the query
into the job's
+ // transformations fails. A future optimizer rule is expected to
rewrite the call before
+ // this stage.
+ assertThatThrownBy(
+ () ->
+ util.generateTransformations(
+ "INSERT INTO RatesSink "
+ + "SELECT * FROM
SNAPSHOT(input => TABLE Rates)"))
+ .satisfies(
+ anyCauseMatches(
+ "Could not find a runtime implementation for
built-in function 'SNAPSHOT'. "
+ + "The planner should have provided an
implementation."));
+ }
+
+ @Test
+ void testLateralContextExecutionFails() {
+ // SNAPSHOT has no runtime implementation yet, so compiling the query
into the job's
+ // transformations fails. A future optimizer rule is expected to
rewrite the call before
+ // this stage.
+ assertThatThrownBy(
+ () ->
+ util.generateTransformations(
+ "INSERT INTO JoinSink "
+ + "SELECT o.order_id,
o.amount, r.rate "
+ + "FROM Orders AS o, LATERAL
TABLE(SNAPSHOT(input => TABLE Rates)) AS r "
+ + "WHERE o.currency =
r.currency"))
+ .satisfies(
+ anyCauseMatches(
+ "Could not find a runtime implementation for
built-in function 'SNAPSHOT'. "
+ + "The planner should have provided an
implementation."));
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.xml
new file mode 100644
index 00000000000..290410f2e41
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.xml
@@ -0,0 +1,73 @@
+<?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="testFromContext">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM SNAPSHOT(input => TABLE Rates,
load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP
'2026-07-01 00:00:00.001' AS TIMESTAMP_LTZ(3)), load_completed_idle_timeout =>
INTERVAL '10' SECOND, state_ttl => INTERVAL '1' DAY)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(currency=[$0], rate=[$1], rate_time=[$2])
++- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_UTF-16LE'user_time', CAST(2026-07-01
00:00:00.001:TIMESTAMP(3)):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL,
10000:INTERVAL SECOND, 86400000:INTERVAL DAY, DEFAULT(), DEFAULT())],
rowType=[RecordType(VARCHAR(2147483647) currency, INTEGER rate, TIMESTAMP(3)
rate_time)])
+ +- LogicalProject(currency=[$0], rate=[$1], rate_time=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[rate_time], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
Rates]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+ProcessTableFunction(invocation=[SNAPSHOT(TABLE(#0), _UTF-16LE'user_time',
CAST(2026-07-01 00:00:00.001:TIMESTAMP(3)):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3)
NOT NULL, 10000:INTERVAL SECOND, 86400000:INTERVAL DAY, DEFAULT(), DEFAULT())],
uid=[null], select=[currency,rate,rate_time],
rowType=[RecordType(VARCHAR(2147483647) currency, INTEGER rate, TIMESTAMP(3)
rate_time)])
++- WatermarkAssigner(rowtime=[rate_time], watermark=[rate_time])
+ +- TableSourceScan(table=[[default_catalog, default_database, Rates]],
fields=[currency, rate, rate_time])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLateralContext">
+ <Resource name="sql">
+ <![CDATA[SELECT o.order_id, o.amount, r.rate FROM Orders AS o, LATERAL
TABLE(SNAPSHOT(input => TABLE Rates)) AS r WHERE o.currency = r.currency]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(order_id=[$0], amount=[$2], rate=[$5])
++- LogicalFilter(condition=[=($1, $4)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalWatermarkAssigner(rowtime=[order_time], watermark=[$3])
+ : +- LogicalTableScan(table=[[default_catalog, default_database,
Orders]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())],
rowType=[RecordType(VARCHAR(2147483647) currency, INTEGER rate, TIMESTAMP(3)
rate_time)])
+ +- LogicalProject(currency=[$0], rate=[$1], rate_time=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[rate_time], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
Rates]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[order_id, amount, rate])
++- Join(joinType=[InnerJoin], where=[=(currency, currency0)],
select=[order_id, currency, amount, currency0, rate],
leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[currency]])
+ : +- Calc(select=[order_id, currency, amount])
+ : +- WatermarkAssigner(rowtime=[order_time], watermark=[order_time])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
Orders]], fields=[order_id, currency, amount, order_time])
+ +- Exchange(distribution=[hash[currency]])
+ +- Calc(select=[currency, rate])
+ +- ProcessTableFunction(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], uid=[null],
select=[currency,rate,rate_time], rowType=[RecordType(VARCHAR(2147483647)
currency, INTEGER rate, TIMESTAMP(3) rate_time)])
+ +- WatermarkAssigner(rowtime=[rate_time], watermark=[rate_time])
+ +- TableSourceScan(table=[[default_catalog, default_database,
Rates]], fields=[currency, rate, rate_time])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
