You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2025/03/10 10:05:24 UTC
(flink) branch master updated: [FLINK-37302][table] Support PTFs with time and timer services
This is an automated email from the ASF dual-hosted git repository.
twalthr 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 7b35edeb3e3 [FLINK-37302][table] Support PTFs with time and timer services
7b35edeb3e3 is described below
commit 7b35edeb3e343d9d35e432bb2c08f43055b96590
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Mar 10 11:05:17 2025 +0100
[FLINK-37302][table] Support PTFs with time and timer services
This closes #26250.
---
.../table/test/program/FailingSqlTestStep.java | 63 ++++
.../flink/table/test/program/SqlTestStep.java | 2 +-
.../flink/table/test/program/TableTestProgram.java | 13 +
.../apache/flink/table/test/program/TestStep.java | 1 +
.../flink/table/annotation/ArgumentTrait.java | 38 +-
.../table/functions/ProcessTableFunction.java | 239 ++++++++++++-
.../table/functions/UserDefinedFunctionHelper.java | 2 +
.../table/types/inference/StaticArgumentTrait.java | 6 +
.../table/types/inference/SystemTypeInference.java | 185 +++++++++-
.../stream/StreamExecProcessTableFunction.java | 24 +-
.../stream/StreamPhysicalProcessTableFunction.java | 42 ++-
.../codegen/ProcessTableRunnerGenerator.scala | 220 ++++++++----
.../stream/ProcessTableFunctionSemanticTests.java | 15 +-
.../stream/ProcessTableFunctionTestPrograms.java | 387 +++++++++++++++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 302 +++++++++++++++-
.../nodes/exec/testutils/SemanticTestBase.java | 20 +-
.../plan/stream/sql/ProcessTableFunctionTest.java | 42 ++-
.../plan/stream/sql/ProcessTableFunctionTest.xml | 85 +++--
.../runtime/generated/ProcessTableRunner.java | 205 ++++++++++-
.../operators/process/ExternalTimeContext.java | 91 +++++
.../operators/process/PassAllCollector.java | 20 +-
.../process/PassPartitionKeysCollector.java | 24 +-
.../process/PassThroughCollectorBase.java | 29 +-
.../operators/process/ProcessTableOperator.java | 310 ++++++++++++-----
.../process/ReadableInternalTimeContext.java | 86 +++++
.../operators/process/RuntimeTableSemantics.java | 27 +-
.../runtime/operators/process/TimeConverter.java | 87 +++++
.../process/WritableInternalTimeContext.java | 186 ++++++++++
28 files changed, 2471 insertions(+), 280 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingSqlTestStep.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingSqlTestStep.java
new file mode 100644
index 00000000000..7a9e94a94f4
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingSqlTestStep.java
@@ -0,0 +1,63 @@
+/*
+ * 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.test.program;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.util.Preconditions;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Test step for executing SQL that will fail eventually with either {@link ValidationException}
+ * (during planning time) or {@link TableRuntimeException} (during execution time).
+ *
+ * <p>Note: Not every runner supports generic SQL statements. Sometimes the runner would like to
+ * enrich properties e.g. of a CREATE TABLE. Use this step with caution.
+ */
+public final class FailingSqlTestStep implements TestStep {
+
+ public final String sql;
+ public final Class<? extends Exception> expectedException;
+ public final String expectedErrorMessage;
+
+ FailingSqlTestStep(
+ String sql, Class<? extends Exception> expectedException, String expectedErrorMessage) {
+ Preconditions.checkArgument(
+ expectedException == ValidationException.class
+ || expectedException == TableRuntimeException.class,
+ "Usually a SQL query should fail with either validation or runtime exception. "
+ + "Otherwise this might require an update to the exception design.");
+ this.sql = sql;
+ this.expectedException = expectedException;
+ this.expectedErrorMessage = expectedErrorMessage;
+ }
+
+ @Override
+ public TestKind getKind() {
+ return TestKind.FAILING_SQL;
+ }
+
+ public void apply(TableEnvironment env) {
+ assertThatThrownBy(() -> env.executeSql(sql).await())
+ .satisfies(anyCauseMatches(expectedException, expectedErrorMessage));
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SqlTestStep.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SqlTestStep.java
index c6809dc16cf..3d9af685542 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SqlTestStep.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SqlTestStep.java
@@ -22,7 +22,7 @@ import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
/**
- * Test step for execution SQL.
+ * Test step for executing SQL.
*
* <p>Note: Not every runner supports generic SQL statements. Sometimes the runner would like to
* enrich properties e.g. of a CREATE TABLE. Use this step with caution.
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java
index 0d9d0f26c57..f911fc2309c 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.test.program;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.test.program.FunctionTestStep.FunctionBehavior;
@@ -324,6 +325,18 @@ public class TableTestProgram {
return this;
}
+ /**
+ * Run step for executing SQL that will fail eventually with a {@link
+ * TableRuntimeException}.
+ */
+ public Builder runFailingSql(
+ String sql,
+ Class<? extends Exception> expectedException,
+ String expectedErrorMessage) {
+ this.runSteps.add(new FailingSqlTestStep(sql, expectedException, expectedErrorMessage));
+ return this;
+ }
+
public Builder runTableApi(Function<TableEnvAccessor, Table> toTable, String sinkName) {
this.runSteps.add(new TableApiTestStep(toTable, sinkName));
return this;
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TestStep.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TestStep.java
index 293789a2353..5a50a634c86 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TestStep.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TestStep.java
@@ -50,6 +50,7 @@ public interface TestStep {
SINK_WITHOUT_DATA,
SINK_WITH_DATA,
SINK_WITH_RESTORE_DATA,
+ FAILING_SQL
}
TestKind getKind();
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
index 076c0c08f19..24a37330c60 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.annotation;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
import org.apache.flink.table.types.inference.StaticArgumentTrait;
import org.apache.flink.types.RowKind;
@@ -90,7 +91,7 @@ public enum ArgumentTrait {
* PARTITION BY clause are passed through.
*
* <p>Given a table t (containing columns k and v), and a PTF f() (producing columns c1 and c2),
- * the output of a {@code SELECT * FROM f(tableArg => TABLE t PARTITION BY k)} uses the
+ * the output of a {@code SELECT * FROM f(table_arg => TABLE t PARTITION BY k)} uses the
* following order:
*
* <pre>
@@ -101,6 +102,8 @@ public enum ArgumentTrait {
* <p>In case of multiple table arguments, pass-through columns are added according to the
* declaration order in the PTF signature.
*
+ * <p>Timers are not available when pass-through columns are enabled.
+ *
* <p>Note: This trait is valid for {@link #TABLE_AS_ROW} and {@link #TABLE_AS_SET} arguments.
*/
PASS_COLUMNS_THROUGH(false, StaticArgumentTrait.PASS_COLUMNS_THROUGH),
@@ -118,7 +121,7 @@ public enum ArgumentTrait {
* WITH UpdatingTable AS (
* SELECT COUNT(*) FROM (VALUES 1, 2, 3)
* )
- * SELECT * FROM f(tableArg => TABLE UpdatingTable)
+ * SELECT * FROM f(table_arg => TABLE UpdatingTable)
* </pre>
*
* <p>If updates should be supported, ensure that the data type of the table argument is chosen
@@ -130,9 +133,38 @@ public enum ArgumentTrait {
* streaming mode, results should be emitted based on watermarks and event-time. The trait
* {@link #PASS_COLUMNS_THROUGH} is not supported if this trait is declared.
*
+ * <p>Timers are not available when updates are enabled.
+ *
+ * <p>Note: This trait is valid for {@link #TABLE_AS_ROW} and {@link #TABLE_AS_SET} arguments.
+ */
+ SUPPORT_UPDATES(false, StaticArgumentTrait.SUPPORT_UPDATES),
+
+ /**
+ * Defines that an {@code on_time} argument must be provided, referencing a watermarked
+ * timestamp column in the given table.
+ *
+ * <p>The {@code on_time} argument indicates which column provides the event-time timestamp. In
+ * other words, it specifies the column that defines the timestamp for when a row was generated.
+ * This timestamp is used within the PTF for timers and time-based operations when the watermark
+ * progresses the logical clock.
+ *
+ * <p>By default, the {@code on_time} argument is optional. If no timestamp column is set for
+ * the PTF, the {@link TimeContext#time()} will return null. If the {@code on_time} argument is
+ * provided, {@link TimeContext#time()} will return it and the PTF will return a {@code rowtime}
+ * column in the output, allowing subsequent operations to access and propagate the resulting
+ * event-time timestamp.
+ *
+ * <p>For example:
+ *
+ * <pre>
+ * CREATE TABLE t (v STRING, ts TIMESTAMP_LTZ(3), WATERMARK FOR ts AS ts - INTERVAL '2' SECONDS);
+ *
+ * SELECT v, rowtime FROM f(table_arg => TABLE t, on_time => DESCRIPTOR(ts));
+ * </pre>
+ *
* <p>Note: This trait is valid for {@link #TABLE_AS_ROW} and {@link #TABLE_AS_SET} arguments.
*/
- SUPPORT_UPDATES(false, StaticArgumentTrait.SUPPORT_UPDATES);
+ REQUIRE_ON_TIME(false, StaticArgumentTrait.REQUIRE_ON_TIME);
private final boolean isRoot;
private final StaticArgumentTrait staticTrait;
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
index 2f5ad136bd0..6094c1a9957 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
@@ -29,6 +29,9 @@ import org.apache.flink.table.types.extraction.TypeInferenceExtractor;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.util.Collector;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
/**
* Base class for a user-defined process table function. A process table function (PTF) maps zero,
* one, or multiple tables to zero, one, or multiple rows (or structured types). Scalar arguments
@@ -285,6 +288,75 @@ import org.apache.flink.util.Collector;
* }
* }</pre>
*
+ * <h1>Time and Timers</h1>
+ *
+ * <p>A PTF supports event time natively. Time-based services are available via {@link
+ * Context#timeContext(Class)}.
+ *
+ * <h2>Time</h2>
+ *
+ * <p>Every PTF takes an optional {@code on_time} argument. The {@code on_time} argument in the
+ * function call declares the time attribute column for which a watermark has been declared. When
+ * processing a table's row, this timestamp can be accessed via {@link TimeContext#time()} and the
+ * watermark via {@link TimeContext#currentWatermark()} respectively.
+ *
+ * <p>Specifying an {@code on_time} argument in the function call instructs the framework to return
+ * a {@code rowtime} column in the function's output for subsequent time-based operations.
+ *
+ * <p>The {@link ArgumentTrait#REQUIRE_ON_TIME} makes the {@code on_time} argument mandatory if
+ * necessary.
+ *
+ * <h2>Timers</h2>
+ *
+ * <p>A PTF that takes set semantic tables can support timers. Timers allow for continuing the
+ * processing at a later point in time. This makes waiting, synchronization, or timeouts possible. A
+ * timer fires for the registered time when the watermark progresses the logical clock.
+ *
+ * <p>Timers can be named ({@link TimeContext#registerOnTime(String, Object)}) or unnamed ({@link
+ * TimeContext#registerOnTime(Object)}). The name of a timer can be useful for replacing or deleting
+ * an existing timer, or for identifying multiple timers via {@link OnTimerContext#currentTimer()}
+ * when they fire.
+ *
+ * <p>An {@code onTimer()} method must be declared next to the eval() method for reacting to timer
+ * events. The signature of the onTimer() method must contain an optional {@link OnTimerContext}
+ * followed by all state entries (as declared in the eval() method).
+ *
+ * <p>Flink takes care of storing and restoring timers during failures or restarts. Thus, timers are
+ * a special kind of state. Similarly, timers are scoped to a virtual processor defined by the
+ * PARTITION BY clause. A timer can only be registered and deleted in the current virtual processor.
+ *
+ * <pre>{@code
+ * // a function that waits for a second event or timeouts after 60 seconds
+ * class TimerFunction extends ProcessTableFunction<String> {
+ * public static class SeenState {
+ * public String seen = null;
+ * }
+ *
+ * public void eval(Context ctx, @StateHint SeenState memory, @ArgumentHint( { TABLE_AS_SET, REQUIRE_ON_TIME } ) Row input) {
+ * TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);
+ * if (memory.seen == null) {
+ * memory.seen = input.getField(0).toString();
+ * timeCtx.registerOnTimer("timeout", timeCtx.time().plusSeconds(60));
+ * } else {
+ * collect("Second event arrived for: " + memory.seen)
+ * ctx.clearAll();
+ * }
+ * }
+ *
+ * public void onTimer(SeenState memory) {
+ * collect("Timeout for: " + memory.seen)
+ * }
+ * }
+ * }</pre>
+ *
+ * <h2>Efficiency and Design Principles</h2>
+ *
+ * <p>Registering too many timers might affect performance. An ever-growing timer state can happen
+ * by an unlimited number of partitions (i.e. an open keyspace) or even within a partition. Thus,
+ * reduce the number of registered timers to a minimum and consider cleaning up timers if they are
+ * not needed anymore via {@link Context#clearAllTimers()} or {@link
+ * TimeContext#clearTimer(String)}.
+ *
* @param <T> The type of the output row. Either an explicit composite type or an atomic type that
* is implicitly wrapped into a row consisting of one field.
*/
@@ -329,6 +401,21 @@ public abstract class ProcessTableFunction<T> extends UserDefinedFunction {
@PublicEvolving
public interface Context {
+ /**
+ * Returns a specialized {@link TimeContext} to work with time and timers.
+ *
+ * <p>Time and timer timestamps can be represented as either {@link Instant}, {@link
+ * LocalDateTime}, or {@link Long}. Time and timers are based on milliseconds since epoch
+ * and do not take the local session timezone into consideration.
+ *
+ * <p>Note: A timer context is always scoped under the currently processed event (either the
+ * current input row or a firing timer).
+ *
+ * @param conversionClass representation of time and timer timestamps
+ * @return the context for time and timer services
+ */
+ <TimeType> TimeContext<TimeType> timeContext(Class<TimeType> conversionClass);
+
/**
* Returns additional information about the semantics of a table argument.
*
@@ -351,9 +438,159 @@ public abstract class ProcessTableFunction<T> extends UserDefinedFunction {
/**
* Clears all state entries within the virtual partition once the eval() method returns.
*
- * <p>Semantically this is equal to calling {@link #clearState(String)} on all state
+ * <p>Semantically, this is equal to calling {@link #clearState(String)} on all state
* entries.
*/
void clearAllState();
+
+ /** Clears all timers within the virtual partition. */
+ void clearAllTimers();
+
+ /** Clears the virtual partition including timers and state. */
+ void clearAll();
+ }
+
+ /**
+ * A context that gives access to Flink's concepts of time and timers.
+ *
+ * <p>An event can have an event-time timestamp assigned. The timestamp can be accessed using
+ * the {@link #time()} method.
+ *
+ * <p>Timers allow for continuing the processing at a later point in time. This makes waiting,
+ * synchronization, or timeouts possible. A timer fires for the registered time when the
+ * watermark progresses the logical clock.
+ *
+ * <p>Flink takes care of storing and restoring timers during failures or restarts. Thus, timers
+ * are a special kind of state. Similarly, timers are scoped to a virtual processor defined by
+ * the PARTITION BY clause. A timer can only be registered and deleted in the current virtual
+ * processor.
+ *
+ * @param <TimeType> conversion class of timestamps, see {@link Context#timeContext(Class)}
+ */
+ @PublicEvolving
+ public interface TimeContext<TimeType> {
+
+ /**
+ * Returns the timestamp of the currently processed event.
+ *
+ * <p>An event can be either the row of a table or a firing timer:
+ *
+ * <h1>Row event timestamp</h1>
+ *
+ * <p>The timestamp of the row currently being processed within the {@code eval()} method.
+ *
+ * <p>Powered by the function call's {@code on_time} argument, this method will return the
+ * content of the referenced time attribute column. Returns {@code null} if the {@code
+ * on_time} argument doesn't reference a time attribute column in the currently processed
+ * table.
+ *
+ * <h1>Timer event timestamp</h1>
+ *
+ * <p>The timestamp of the firing timer currently being processed within the {@code
+ * onTimer()} method.
+ *
+ * @return the event-time timestamp, or {@code null} if no timestamp is present
+ */
+ TimeType time();
+
+ /**
+ * Returns the current event-time watermark.
+ *
+ * <p>Watermarks are generated in sources and sent through the topology for advancing the
+ * logical clock in each Flink subtask. The current watermark of a Flink subtask is the
+ * global minimum watermark of all inputs (i.e. across all parallel inputs and table
+ * partitions).
+ *
+ * <p>This method returns the current watermark of the Flink subtask that evaluates the PTF.
+ * Thus, the returned timestamp represents the entire Flink subtask, independent of the
+ * currently processed partition. This behavior is similar to a call to {@code SELECT
+ * CURRENT_WATERMARK(...)} in SQL.
+ *
+ * <p>If a watermark was not received from all inputs, the method returns {@code null}.
+ *
+ * <p>In case this method is called within the {@code onTimer()} method, the returned
+ * watermark is the triggering watermark that currently fires the timer.
+ *
+ * @return the current watermark of the Flink subtask, or {@code null} if no common logical
+ * time could be determined from the inputs
+ */
+ TimeType currentWatermark();
+
+ /**
+ * Registers a timer under the given name.
+ *
+ * <p>The timer fires when the {@link #currentWatermark()} advances the logical clock of the
+ * Flink subtask to a timestamp later or equal to the desired timestamp. In other words: A
+ * timer only fires if a watermark was received from all inputs and the timestamp is smaller
+ * or equal to the minimum of all received watermarks.
+ *
+ * <p>Timers can be named for distinguishing them in the {@code onTimer()} method.
+ * Registering a timer under the same name twice will replace an existing timer.
+ *
+ * <p>Note: Because only PTFs taking set semantic tables support state, and timers are a
+ * special kind of state, at least one {@link ArgumentTrait#TABLE_AS_SET} table argument
+ * must be declared.
+ *
+ * @param name identifier of the timer
+ * @param time timestamp when the timer should fire
+ */
+ void registerOnTime(String name, TimeType time);
+
+ /**
+ * Registers a timer.
+ *
+ * <p>The timer fires when the {@link #currentWatermark()} advances the logical clock of the
+ * Flink subtask to a timestamp later or equal to the desired timestamp. In other words: A
+ * timer only fires if a watermark was received from all inputs and the timestamp is smaller
+ * or equal to the minimum of all received watermarks.
+ *
+ * <p>Only one timer can be registered for a given time.
+ *
+ * <p>Note: Because only PTFs taking set semantic tables support state, and timers are a
+ * special kind of state, at least one {@link ArgumentTrait#TABLE_AS_SET} table argument
+ * must be declared.
+ *
+ * @param time timestamp when the timer should fire
+ */
+ void registerOnTime(TimeType time);
+
+ /**
+ * Clears a timer that was previously registered under the given name.
+ *
+ * <p>The call is ignored if no timer can be found.
+ *
+ * @param name identifier of the timer
+ */
+ void clearTimer(String name);
+
+ /**
+ * Clears a timer that was previously registered for a given time.
+ *
+ * <p>The call is ignored if no timer can be found. Named timers cannot be deleted with this
+ * method.
+ *
+ * @param time timestamp when the timer should have fired
+ */
+ void clearTimer(TimeType time);
+
+ /** Deletes all timers within the virtual partition. */
+ void clearAllTimers();
+ }
+
+ /** Special {@link Context} that is available when the {@code onTimer()} method is called. */
+ @PublicEvolving
+ public interface OnTimerContext extends Context {
+
+ /**
+ * Returns the name of the timer currently being fired, if the timer is a named timer (i.e.
+ * registered using {@link TimeContext#registerOnTime(String, Object)}).
+ *
+ * <p>Note: The time of the firing timer is available via {@link
+ * OnTimerContext#timeContext(Class)}.
+ *
+ * @return the timer's name, or {@code null} if the timer is unnamed (i.e. registered using
+ * {@link TimeContext#registerOnTime(Object)})
+ */
+ String currentTimer();
}
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
index bd5e9e95cd6..25d7e4ea7cf 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
@@ -95,6 +95,8 @@ public final class UserDefinedFunctionHelper {
public static final String PROCESS_TABLE_EVAL = "eval";
+ public static final String PROCESS_TABLE_ON_TIMER = "onTimer";
+
public static final String DEFAULT_ACCUMULATOR_NAME = "acc";
/**
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
index fbcc7be7879..41490a26fb9 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StaticArgumentTrait.java
@@ -32,13 +32,19 @@ import java.util.stream.Collectors;
*/
@PublicEvolving
public enum StaticArgumentTrait {
+ // Roots
SCALAR(),
TABLE(),
MODEL(),
+
+ // For TABLE
TABLE_AS_ROW(TABLE),
TABLE_AS_SET(TABLE),
PASS_COLUMNS_THROUGH(TABLE),
SUPPORT_UPDATES(TABLE),
+ REQUIRE_ON_TIME(TABLE),
+
+ // For TABLE_AS_SET
OPTIONAL_PARTITION_BY(TABLE_AS_SET);
private final Set<StaticArgumentTrait> requirements;
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
index 038ab8c88aa..c6eee77ed1b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
@@ -28,13 +28,25 @@ import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.functions.TableSemantics;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.RowType.RowField;
+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.logical.utils.LogicalTypeMerging;
+import org.apache.flink.types.ColumnList;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Predicate;
@@ -57,7 +69,11 @@ import java.util.stream.Stream;
public class SystemTypeInference {
public static final List<StaticArgument> PROCESS_TABLE_FUNCTION_SYSTEM_ARGS =
- List.of(StaticArgument.scalar("uid", DataTypes.STRING(), true));
+ List.of(
+ StaticArgument.scalar("on_time", DataTypes.DESCRIPTOR(), true),
+ StaticArgument.scalar("uid", DataTypes.STRING(), true));
+
+ public static final String PROCESS_TABLE_FUNCTION_RESULT_ROWTIME = "rowtime";
/**
* Format of unique identifiers for {@link ProcessTableFunction}.
@@ -129,10 +145,10 @@ public class SystemTypeInference {
private static void checkReservedArgs(List<StaticArgument> staticArgs) {
final Set<String> declaredArgs =
staticArgs.stream().map(StaticArgument::getName).collect(Collectors.toSet());
- final Set<String> reservedArgs =
+ final List<String> reservedArgs =
PROCESS_TABLE_FUNCTION_SYSTEM_ARGS.stream()
.map(StaticArgument::getName)
- .collect(Collectors.toSet());
+ .collect(Collectors.toList());
if (reservedArgs.stream().anyMatch(declaredArgs::contains)) {
throw new ValidationException(
"Function signature must not declare system arguments. "
@@ -177,15 +193,18 @@ public class SystemTypeInference {
if (functionKind != FunctionKind.TABLE && functionKind != FunctionKind.PROCESS_TABLE) {
return outputStrategy;
}
- return new SystemOutputStrategy(staticArgs, outputStrategy);
+ return new SystemOutputStrategy(functionKind, staticArgs, outputStrategy);
}
private static class SystemOutputStrategy implements TypeStrategy {
+ private final FunctionKind functionKind;
private final List<StaticArgument> staticArgs;
private final TypeStrategy origin;
- private SystemOutputStrategy(List<StaticArgument> staticArgs, TypeStrategy origin) {
+ private SystemOutputStrategy(
+ FunctionKind functionKind, List<StaticArgument> staticArgs, TypeStrategy origin) {
+ this.functionKind = functionKind;
this.staticArgs = staticArgs;
this.origin = origin;
}
@@ -208,6 +227,7 @@ public class SystemTypeInference {
// this whole topic is kind of vendor specific already
fields.addAll(derivePassThroughFields(callContext));
fields.addAll(deriveFunctionOutputFields(functionDataType));
+ fields.addAll(deriveRowtimeField(callContext));
final List<Field> uniqueFields = makeFieldNamesUnique(fields);
@@ -233,7 +253,7 @@ public class SystemTypeInference {
}
private List<Field> derivePassThroughFields(CallContext callContext) {
- if (staticArgs == null) {
+ if (functionKind != FunctionKind.PROCESS_TABLE) {
return List.of();
}
final List<DataType> argDataTypes = callContext.getArgumentDataTypes();
@@ -275,6 +295,155 @@ public class SystemTypeInference {
.mapToObj(pos -> DataTypes.FIELD(fieldNames.get(pos), fieldTypes.get(pos)))
.collect(Collectors.toList());
}
+
+ private List<Field> deriveRowtimeField(CallContext callContext) {
+ if (this.functionKind != FunctionKind.PROCESS_TABLE) {
+ return List.of();
+ }
+ final List<DataType> args = callContext.getArgumentDataTypes();
+
+ // Check if on_time is defined and non-empty
+ final int onTimePos = args.size() - 2;
+ final Set<String> onTimeFields =
+ callContext
+ .getArgumentValue(onTimePos, ColumnList.class)
+ .map(ColumnList::getNames)
+ .map(Set::copyOf)
+ .orElse(Set.of());
+
+ final Set<String> usedOnTimeFields = new HashSet<>();
+
+ final List<LogicalType> onTimeColumns =
+ IntStream.range(0, staticArgs.size())
+ .mapToObj(
+ pos -> {
+ final StaticArgument staticArg = staticArgs.get(pos);
+ if (!staticArg.is(StaticArgumentTrait.TABLE)) {
+ return null;
+ }
+ final RowType rowType =
+ (RowType) args.get(pos).getLogicalType();
+ final int onTimeColumn =
+ findUniqueOnTimeColumn(
+ staticArg.getName(), rowType, onTimeFields);
+ if (onTimeColumn >= 0) {
+ usedOnTimeFields.add(
+ rowType.getFieldNames().get(onTimeColumn));
+ return rowType.getTypeAt(onTimeColumn);
+ }
+ if (staticArg.is(StaticArgumentTrait.REQUIRE_ON_TIME)) {
+ throw new ValidationException(
+ String.format(
+ "Table argument '%s' requires a time attribute. "
+ + "Please provide one using the implicit `on_time` argument. "
+ + "For example: myFunction(..., on_time => DESCRIPTOR(`my_timestamp`)",
+ staticArg.getName()));
+ }
+ return null;
+ })
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+
+ final Set<String> unusedOnTimeFields = new HashSet<>(onTimeFields);
+ unusedOnTimeFields.removeAll(usedOnTimeFields);
+ if (!unusedOnTimeFields.isEmpty()) {
+ throw new ValidationException(
+ "Invalid time attribute declaration. "
+ + "Each column in the `on_time` argument must reference at least one "
+ + "column in one of the table arguments. Unknown references: "
+ + unusedOnTimeFields);
+ }
+
+ if (onTimeColumns.isEmpty()) {
+ return List.of();
+ }
+
+ // Don't allow mixtures of time attribute roots
+ final Set<LogicalTypeRoot> onTimeRoots =
+ onTimeColumns.stream()
+ .map(LogicalType::getTypeRoot)
+ .collect(Collectors.toSet());
+ if (onTimeRoots.size() > 1) {
+ throw new ValidationException(
+ "Invalid time attribute declaration. "
+ + "All columns in the `on_time` argument must reference the same data type kind. "
+ + "But found: "
+ + onTimeRoots);
+ }
+
+ final LogicalType commonOnTimeType =
+ LogicalTypeMerging.findCommonType(onTimeColumns)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Unable to derive data type for PTF result time attribute."));
+
+ final LogicalType resultTimestamp =
+ forwardTimeAttribute(commonOnTimeType, onTimeColumns);
+
+ return List.of(
+ DataTypes.FIELD(
+ PROCESS_TABLE_FUNCTION_RESULT_ROWTIME, DataTypes.of(resultTimestamp)));
+ }
+
+ private static int findUniqueOnTimeColumn(
+ String tableArgName, RowType rowType, Set<String> onTimeFields) {
+ final List<RowField> fields = rowType.getFields();
+ int found = -1;
+ for (int pos = 0; pos < fields.size(); pos++) {
+ final RowField field = fields.get(pos);
+ if (!onTimeFields.contains(field.getName())) {
+ continue;
+ }
+ if (found != -1) {
+ throw new ValidationException(
+ String.format(
+ "Ambiguous time attribute found. "
+ + "The `on_time` argument must reference at most one column in a table argument. "
+ + "Currently, the columns in `on_time` point to both '%s' and '%s' in table argument '%s'.",
+ fields.get(found).getName(), field.getName(), tableArgName));
+ }
+ found = pos;
+ if (isUnsupportedOnTimeColumn(field.getType())) {
+ throw new ValidationException(
+ String.format(
+ "Unsupported data type for time attribute. "
+ + "The `on_time` argument must reference a TIMESTAMP or TIMESTAMP_LTZ column (up to precision 3). "
+ + "However, column '%s' in table argument '%s' has data type '%s'.",
+ field.getName(),
+ tableArgName,
+ field.getType().asSummaryString()));
+ }
+ }
+ return found;
+ }
+
+ private static LogicalType forwardTimeAttribute(
+ LogicalType timestampType, List<LogicalType> onTimeColumns) {
+ if (onTimeColumns.stream().noneMatch(LogicalTypeChecks::isTimeAttribute)) {
+ return timestampType.copy(false);
+ }
+ switch (timestampType.getTypeRoot()) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return new TimestampType(
+ false,
+ TimestampKind.ROWTIME,
+ LogicalTypeChecks.getPrecision(timestampType));
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return new LocalZonedTimestampType(
+ false,
+ TimestampKind.ROWTIME,
+ LogicalTypeChecks.getPrecision(timestampType));
+ default:
+ throw new IllegalStateException(
+ "Timestamp type expected for PTF result time attribute.");
+ }
+ }
+
+ private static boolean isUnsupportedOnTimeColumn(LogicalType type) {
+ return !LogicalTypeChecks.canBeTimeAttributeType(type)
+ || LogicalTypeChecks.getPrecision(type) > 3;
+ }
}
private static class SystemInputStrategy implements InputTypeStrategy {
@@ -312,8 +481,8 @@ public class SystemTypeInference {
+ "that is not overloaded and doesn't contain varargs.");
}
- checkUidArg(callContext);
checkTableArgTraits(staticArgs, callContext);
+ checkUidArg(callContext);
return Optional.of(inferredDataTypes);
}
@@ -329,7 +498,7 @@ public class SystemTypeInference {
final List<DataType> args = callContext.getArgumentDataTypes();
// Verify the uid format if provided
- int uidPos = args.size() - 1;
+ final int uidPos = args.size() - 1;
if (!callContext.isArgumentNull(uidPos)) {
final String uid = callContext.getArgumentValue(uidPos, String.class).orElse("");
if (isInvalidUidForProcessTableFunction(uid)) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
index 6ffe7211817..9a2f01b3625 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecProcessTableFunction.java
@@ -63,6 +63,7 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCre
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexNode;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -70,10 +71,13 @@ import org.checkerframework.checker.nullness.qual.Nullable;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.apache.flink.table.planner.codegen.ProcessTableRunnerGenerator.GeneratedRunnerResult;
+import static org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalProcessTableFunction.deriveOnTimeFields;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getFieldCount;
/**
@@ -167,6 +171,7 @@ public class StreamExecProcessTableFunction extends ExecNodeBase<RowData>
final List<Ord<StaticArgument>> providedInputArgs =
StreamPhysicalProcessTableFunction.getProvidedInputArgs(invocation);
final List<RexNode> operands = invocation.getOperands();
+ final Set<String> onTimeFields = deriveOnTimeFields(operands);
final List<RuntimeTableSemantics> runtimeTableSemantics =
providedInputArgs.stream()
.map(
@@ -174,7 +179,8 @@ public class StreamExecProcessTableFunction extends ExecNodeBase<RowData>
final RexTableArgCall tableArgCall =
(RexTableArgCall) operands.get(providedInputArg.i);
final StaticArgument tabledArg = providedInputArg.e;
- return createRuntimeTableSemantics(tabledArg, tableArgCall);
+ return createRuntimeTableSemantics(
+ tabledArg, tableArgCall, onTimeFields);
})
.collect(Collectors.toList());
@@ -264,7 +270,7 @@ public class StreamExecProcessTableFunction extends ExecNodeBase<RowData>
}
private RuntimeTableSemantics createRuntimeTableSemantics(
- StaticArgument tableArg, RexTableArgCall tableArgCall) {
+ StaticArgument tableArg, RexTableArgCall tableArgCall, Set<String> onTimeFields) {
final RowKind[] kinds =
inputChangelogModes
.get(tableArgCall.getInputIndex())
@@ -278,6 +284,17 @@ public class StreamExecProcessTableFunction extends ExecNodeBase<RowData>
} else {
dataType = DataTypes.of(FlinkTypeFactory.toLogicalRowType(tableArgCall.type));
}
+
+ final int timeColumn =
+ onTimeFields.stream()
+ .map(
+ onTimeField ->
+ tableArgCall.getType().getField(onTimeField, true, false))
+ .filter(Objects::nonNull)
+ .map(RelDataTypeField::getIndex)
+ .findFirst()
+ .orElse(-1);
+
return new RuntimeTableSemantics(
tableArg.getName(),
tableArgCall.getInputIndex(),
@@ -285,7 +302,8 @@ public class StreamExecProcessTableFunction extends ExecNodeBase<RowData>
tableArgCall.getPartitionKeys(),
expectedChanges,
tableArg.is(StaticArgumentTrait.PASS_COLUMNS_THROUGH),
- tableArg.is(StaticArgumentTrait.TABLE_AS_SET));
+ tableArg.is(StaticArgumentTrait.TABLE_AS_SET),
+ timeColumn);
}
private static RuntimeStateInfo createRuntimeStateInfo(
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java
index ab0b5168813..6a83ff4589a 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java
@@ -45,6 +45,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
@@ -53,6 +54,8 @@ import org.checkerframework.checker.nullness.qual.Nullable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
@@ -140,13 +143,15 @@ public class StreamPhysicalProcessTableFunction extends AbstractRelNode
.map(JavaScalaConversionUtil::toJava)
.map(optional -> optional.orElseThrow(IllegalStateException::new))
.collect(Collectors.toList());
+ final RexCall call = (RexCall) scan.getCall();
+ verifyTimeAttributes(getInputs(), call);
return new StreamExecProcessTableFunction(
unwrapTableConfig(this),
getInputs().stream().map(i -> InputProperty.DEFAULT).collect(Collectors.toList()),
FlinkTypeFactory.toLogicalRowType(rowType),
getRelDetailedDescription(),
uid,
- (RexCall) scan.getCall(),
+ call,
inputChangelogModes);
}
@@ -212,6 +217,31 @@ public class StreamPhysicalProcessTableFunction extends AbstractRelNode
return RexLiteral.stringValue(uidRexNode);
}
+ private static void verifyTimeAttributes(List<RelNode> inputs, RexCall call) {
+ final Set<String> onTimeFields = deriveOnTimeFields(call.getOperands());
+ inputs.stream()
+ .map(RelNode::getRowType)
+ .forEach(rowType -> verifyTimeAttribute(rowType, onTimeFields));
+ }
+
+ private static void verifyTimeAttribute(RelDataType rowType, Set<String> onTimeFields) {
+ onTimeFields.stream()
+ .map(onTimeField -> rowType.getField(onTimeField, true, false))
+ .filter(Objects::nonNull)
+ .forEach(StreamPhysicalProcessTableFunction::verifyTimeAttribute);
+ }
+
+ private static void verifyTimeAttribute(RelDataTypeField timeColumn) {
+ if (!FlinkTypeFactory.isTimeIndicatorType(timeColumn.getType())) {
+ throw new ValidationException(
+ String.format(
+ "Column '%s' is not a valid time attribute. "
+ + "Only columns with a watermark declaration qualify for the `on_time` argument. "
+ + "Also, make sure that the watermarked column is forwarded without any modification.",
+ timeColumn.getName()));
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Shared utilities
// --------------------------------------------------------------------------------------------
@@ -231,4 +261,14 @@ public class StreamPhysicalProcessTableFunction extends AbstractRelNode
.filter(arg -> operands.get(arg.i) instanceof RexTableArgCall)
.collect(Collectors.toList());
}
+
+ public static Set<String> deriveOnTimeFields(List<RexNode> operands) {
+ final RexCall onTimeOperand = (RexCall) operands.get(operands.size() - 2);
+ if (onTimeOperand.getKind() == SqlKind.DEFAULT) {
+ return Set.of();
+ }
+ return onTimeOperand.getOperands().stream()
+ .map(RexLiteral::stringValue)
+ .collect(Collectors.toSet());
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
index da542ee68c0..cfe84eab11a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
@@ -23,7 +23,7 @@ import org.apache.flink.table.connector.ChangelogMode
import org.apache.flink.table.data.RowData
import org.apache.flink.table.data.conversion.RowRowConverter
import org.apache.flink.table.functions.{FunctionDefinition, ProcessTableFunction, UserDefinedFunction, UserDefinedFunctionHelper}
-import org.apache.flink.table.functions.UserDefinedFunctionHelper.{validateClassForRuntime, PROCESS_TABLE_EVAL}
+import org.apache.flink.table.functions.UserDefinedFunctionHelper.{validateClassForRuntime, PROCESS_TABLE_EVAL, PROCESS_TABLE_ON_TIMER}
import org.apache.flink.table.planner.calcite.{FlinkTypeFactory, RexTableArgCall}
import org.apache.flink.table.planner.codegen.CodeGenUtils._
import org.apache.flink.table.planner.codegen.GeneratedExpression.{NEVER_NULL, NO_CODE}
@@ -43,6 +43,7 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeChecks
import org.apache.flink.types.Row
import org.apache.calcite.rex.{RexCall, RexCallBinding, RexNode}
+import org.apache.calcite.sql.SqlKind
import java.util
import java.util.Collections
@@ -92,10 +93,11 @@ object ProcessTableRunnerGenerator {
ctx.tableConfig,
new DefaultExpressionEvaluatorFactory(ctx.tableConfig, ctx.classLoader, rexFactory)
)
+ val functionTerm = ctx.addReusableFunction(udf)
val inference = udf.getTypeInference(dataTypeFactory)
+ val adaptedCallContext = TypeInferenceUtil.adaptArguments(inference, callContext, null)
// Enrich argument types with conversion class
- val adaptedCallContext = TypeInferenceUtil.adaptArguments(inference, callContext, null)
val enrichedArgumentDataTypes = toScala(adaptedCallContext.getArgumentDataTypes)
// Enrich output types with conversion class
@@ -111,7 +113,6 @@ object ProcessTableRunnerGenerator {
val stateDataTypes = stateInfos.asScala.values.map(_.getDataType).toSeq
stateDataTypes.foreach(ExtractionUtils.checkStateDataType)
- // Generate operands (state + args)
val stateToFunctionTerm = "stateToFunction"
val stateClearedTerm = "stateCleared"
val stateFromFunctionTerm = "stateFromFunction"
@@ -122,40 +123,35 @@ object ProcessTableRunnerGenerator {
externalStateOperands,
stateClearedTerm,
stateFromFunctionTerm)
- val tablesTerm = "tables"
- val callOperands = generateCallOperands(ctx, call, tablesTerm)
- val externalCallOperands =
- BridgingFunctionGenUtil.prepareExternalOperands(ctx, callOperands, enrichedArgumentDataTypes)
- val allExternalOperands = externalStateOperands ++ externalCallOperands
- val allDataTypes = stateDataTypes ++ enrichedArgumentDataTypes
- // Find and verify suitable runtime method
+ // Generate result collector
+ val resultCollectorTerm =
+ BridgingFunctionGenUtil.generateResultCollector(ctx, enrichedOutputDataType, returnType)
+ val setCollectorCode = s"$functionTerm.setCollector($resultCollectorTerm);"
+ ctx.addReusableOpenStatement(setCollectorCode)
- val withContext = verifyEvalImplementation(allDataTypes, udf, functionName)
- val contextTerm = if (withContext) {
- Some("runnerContext")
- } else {
- None
- }
-
- // Generate function call with operands (table/scalar arguments and state)
- val functionTerm = ctx.addReusableFunction(udf)
- val functionCall = BridgingFunctionGenUtil.generateTableFunctionCall(
+ // Generate call to eval()
+ val evalCallCode = generateEvalCode(
ctx,
+ call,
+ enrichedArgumentDataTypes,
+ externalStateOperands,
+ stateDataTypes,
+ udf,
+ functionName,
functionTerm,
- allExternalOperands,
- enrichedOutputDataType,
- returnType,
- skipIfArgsNull = false,
- contextTerm)
-
- // Generate setting result collector
- val resultCollectorTerm = functionCall.resultTerm
- val setCollectors =
- s"""
- |$resultCollectorTerm.setCollector(runnerCollector);
- |""".stripMargin
- ctx.addReusableOpenStatement(setCollectors)
+ resultCollectorTerm,
+ stateFromFunctionCode)
+
+ // Generate call to onTimer()
+ val onTimerCallCode = generateOnTimerCode(
+ externalStateOperands,
+ stateDataTypes,
+ udf,
+ functionName,
+ functionTerm,
+ resultCollectorTerm,
+ stateFromFunctionCode)
// Generate runner
val name = newName(ctx, "ProcessTableRunner")
@@ -164,8 +160,6 @@ object ProcessTableRunnerGenerator {
public final class $name
extends ${className[ProcessTableRunner]} {
- private transient $ROW_DATA[] $tablesTerm;
-
${ctx.reuseMemberCode()}
public $name(Object[] references) throws Exception {
@@ -175,23 +169,19 @@ object ProcessTableRunnerGenerator {
@Override
public void open(${className[OpenContext]} openContext) throws Exception {
${ctx.reuseOpenCode()}
- $tablesTerm = new $ROW_DATA[${inputChangelogModes.size()}];
}
@Override
- public void processElement(
- $ROW_DATA[] $stateToFunctionTerm,
- boolean[] $stateClearedTerm,
- $ROW_DATA[] $stateFromFunctionTerm,
- int inputIndex,
- $ROW_DATA row) throws Exception {
- ${className[util.Arrays]}.fill($tablesTerm, null);
- $tablesTerm[inputIndex] = row;
- ${ctx.reusePerRecordCode()}
+ public void callEval() throws Exception {
${ctx.reuseLocalVariableCode()}
- ${ctx.reuseInputUnboxingCode()}
- ${functionCall.code}
- ${stateFromFunctionCode.mkString("\n")}
+ $evalCallCode
+ $stateFromFunctionCode
+ }
+
+ @Override
+ public void callOnTimer() throws Exception {
+ ${ctx.reuseLocalVariableCode()}
+ $onTimerCallCode
}
@Override
@@ -213,8 +203,9 @@ object ProcessTableRunnerGenerator {
invocation: RexCall,
staticArgs: Seq[StaticArgument],
typeFactory: FlinkTypeFactory): RexCall = {
+ val operands = invocation.getOperands.asScala
// Remove system arguments
- val newOperands = invocation.getOperands.asScala
+ val newOperands = operands
.dropRight(SystemTypeInference.PROCESS_TABLE_FUNCTION_SYSTEM_ARGS.size())
// Remove system output columns
val prefixOutputSystemFields = newOperands.zipWithIndex.map {
@@ -227,7 +218,14 @@ object ProcessTableRunnerGenerator {
}
case _ => 0
}.sum
- val projectedFields = invocation.getType.getFieldList.asScala.drop(prefixOutputSystemFields)
+ val suffixOutputSystemFields = if (operands(operands.size - 2).getKind != SqlKind.DEFAULT) {
+ 1
+ } else {
+ 0
+ }
+ val projectedFields = invocation.getType.getFieldList.asScala
+ .drop(prefixOutputSystemFields)
+ .dropRight(suffixOutputSystemFields)
val newReturnType = typeFactory.createStructType(projectedFields.asJava)
invocation.clone(newReturnType, newOperands.asJava)
}
@@ -281,7 +279,7 @@ object ProcessTableRunnerGenerator {
stateDataTypes: Seq[DataType],
externalStateOperands: Seq[GeneratedExpression],
stateClearedTerm: String,
- stateFromFunctionTerm: String): Seq[String] = {
+ stateFromFunctionTerm: String): String = {
stateDataTypes.zipWithIndex
.map {
case (stateDataType, pos) =>
@@ -290,12 +288,14 @@ object ProcessTableRunnerGenerator {
s"$stateEntryTerm = $stateClearedTerm[$pos] ? null : " +
s"${genToInternalConverter(ctx, stateDataType)(externalStateOperandTerm)};"
}
+ .mkString("\n")
}
- private def generateCallOperands(
+ private def generateEvalOperands(
ctx: CodeGeneratorContext,
call: RexCall,
- tablesTerm: String): Seq[GeneratedExpression] = {
+ inputIndexTerm: String,
+ inputRowTerm: String): Seq[GeneratedExpression] = {
val generator = new ExprCodeGenerator(ctx, false)
call.getOperands.asScala
.map {
@@ -303,8 +303,8 @@ object ProcessTableRunnerGenerator {
val inputIndex = tableArgCall.getInputIndex
val tableType = FlinkTypeFactory.toLogicalType(call.getType).copy(true)
GeneratedExpression(
- s"$tablesTerm[$inputIndex]",
- s"$tablesTerm[$inputIndex] == null",
+ s"$inputRowTerm",
+ s"$inputIndexTerm != $inputIndex",
NO_CODE,
tableType)
case rexNode: RexNode =>
@@ -312,33 +312,125 @@ object ProcessTableRunnerGenerator {
}
}
- private def verifyEvalImplementation(
- argumentDataTypes: Seq[DataType],
+ private def verifyMethodImplementation(
+ methodName: String,
+ contextClass: Class[_],
+ parameterDataTypes: Seq[DataType],
udf: UserDefinedFunction,
- functionName: String): Boolean = {
- val contextClass = classOf[ProcessTableFunction.Context]
- val argumentClasses = argumentDataTypes.map(_.getConversionClass)
+ functionName: String,
+ optional: Boolean): Option[Boolean] = {
+ if (optional && ExtractionUtils.collectMethods(udf.getClass, methodName).isEmpty) {
+ return None
+ }
+ val argumentClasses = parameterDataTypes.map(_.getConversionClass)
try {
// Try with context first
val fullSignature = contextClass +: argumentClasses
validateClassForRuntime(
udf.getClass,
- PROCESS_TABLE_EVAL,
+ methodName,
fullSignature.toArray,
classOf[Unit],
functionName)
- true
+ Some(true)
} catch {
case _: ValidationException =>
// Try without context also to not force people to use a context
val minimalSignature = argumentClasses
validateClassForRuntime(
udf.getClass,
- PROCESS_TABLE_EVAL,
+ methodName,
minimalSignature.toArray,
classOf[Unit],
functionName)
- false
+ Some(false)
+ }
+ }
+
+ private def generateEvalCode(
+ ctx: CodeGeneratorContext,
+ call: RexCall,
+ enrichedArgumentDataTypes: Seq[DataType],
+ externalStateOperands: Seq[GeneratedExpression],
+ stateDataTypes: Seq[DataType],
+ udf: UserDefinedFunction,
+ functionName: String,
+ functionTerm: String,
+ resultCollectorTerm: String,
+ stateFromFunctionCode: String): String = {
+ val inputIndexTerm = "inputIndex"
+ val inputRowTerm = "inputRow"
+ val collectorTerm = "evalCollector"
+ val callOperands = generateEvalOperands(ctx, call, inputIndexTerm, inputRowTerm)
+ val externalCallOperands =
+ BridgingFunctionGenUtil.prepareExternalOperands(ctx, callOperands, enrichedArgumentDataTypes)
+ val allExternalOperands = externalStateOperands ++ externalCallOperands
+ val allDataTypes = stateDataTypes ++ enrichedArgumentDataTypes
+
+ // Find and verify suitable runtime method
+ val withContext = verifyMethodImplementation(
+ PROCESS_TABLE_EVAL,
+ classOf[ProcessTableFunction.Context],
+ allDataTypes,
+ udf,
+ functionName,
+ optional = false)
+ val contextOperand = if (withContext.get) {
+ Seq("runnerContext")
+ } else {
+ Seq()
+ }
+
+ val methodOperands = contextOperand ++ allExternalOperands.map(_.resultTerm)
+
+ val functionCallCode =
+ s"""
+ |$resultCollectorTerm.setCollector($collectorTerm);
+ |${allExternalOperands.map(_.code).mkString("\n")}
+ |$functionTerm.eval(${methodOperands.mkString(", ")});
+ |$stateFromFunctionCode
+ |""".stripMargin
+
+ functionCallCode
+ }
+
+ private def generateOnTimerCode(
+ externalStateOperands: Seq[GeneratedExpression],
+ stateDataTypes: Seq[DataType],
+ udf: UserDefinedFunction,
+ functionName: String,
+ functionTerm: String,
+ resultCollectorTerm: String,
+ stateFromFunctionCode: String): String = {
+ val collectorTerm = "onTimerCollector"
+
+ // Find and verify suitable runtime method
+ val withContext = verifyMethodImplementation(
+ PROCESS_TABLE_ON_TIMER,
+ classOf[ProcessTableFunction.OnTimerContext],
+ stateDataTypes,
+ udf,
+ functionName,
+ optional = true)
+ if (withContext.isEmpty) {
+ return ""
+ }
+ val contextOperand = if (withContext.get) {
+ Seq("runnerOnTimerContext")
+ } else {
+ Seq()
}
+
+ val methodOperands = contextOperand ++ externalStateOperands.map(_.resultTerm)
+
+ val functionCallCode =
+ s"""
+ |$resultCollectorTerm.setCollector($collectorTerm);
+ |${externalStateOperands.map(_.code).mkString("\n")}
+ |$functionTerm.onTimer(${methodOperands.mkString(", ")});
+ |$stateFromFunctionCode
+ |""".stripMargin
+
+ functionCallCode
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index 293583de8de..621b4e36dff 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -47,6 +47,19 @@ public class ProcessTableFunctionSemanticTests extends SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_MULTI_STATE,
ProcessTableFunctionTestPrograms.PROCESS_CLEARING_STATE,
ProcessTableFunctionTestPrograms.PROCESS_STATE_TTL,
- ProcessTableFunctionTestPrograms.PROCESS_DESCRIPTOR);
+ ProcessTableFunctionTestPrograms.PROCESS_DESCRIPTOR,
+ ProcessTableFunctionTestPrograms.PROCESS_TIME_CONVERSIONS,
+ ProcessTableFunctionTestPrograms.PROCESS_REGULAR_TIMESTAMP,
+ ProcessTableFunctionTestPrograms.PROCESS_NAMED_TIMERS,
+ ProcessTableFunctionTestPrograms.PROCESS_UNNAMED_TIMERS,
+ ProcessTableFunctionTestPrograms.PROCESS_LATE_EVENTS,
+ ProcessTableFunctionTestPrograms.PROCESS_SCALAR_ARGS_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_ON_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_POJO_STATE_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_INVALID_TABLE_AS_ROW_TIMERS,
+ ProcessTableFunctionTestPrograms.PROCESS_INVALID_PASS_THROUGH_TIMERS,
+ ProcessTableFunctionTestPrograms.PROCESS_INVALID_UPDATING_TIMERS);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index c9b3691bd75..29e2247b4bf 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -18,18 +18,30 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.AtomicTypeWrappingFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedSendingFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearStateFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.InvalidPassThroughTimersFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.InvalidTableAsRowTimersFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.InvalidUpdatingTimersFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.LateTimersFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.MultiStateFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NamedTimersFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalOnTimeFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalPartitionOnTimeFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoArgsFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoCreatingFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoStateFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoStateTimeFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoWithDefaultStateFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ScalarArgsFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ScalarArgsTimeFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowPassThroughFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetFunction;
@@ -37,13 +49,19 @@ import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetPassThroughFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsSetUpdatingArgFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TestProcessTableFunctionBase;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TimeConversionsFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TimeToLiveStateFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedTableAsRowFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedTableAsSetFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UnnamedTimersFunction;
import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
/** {@link TableTestProgram} definitions for testing {@link StreamExecProcessTableFunction}. */
public class ProcessTableFunctionTestPrograms {
@@ -61,9 +79,48 @@ public class ProcessTableFunctionTestPrograms {
+ "(VALUES ('Bob', 12), ('Alice', 42), ('Bob', 14)) AS T(name, score) "
+ "GROUP BY name";
+ private static final SourceTestStep TIMED_SOURCE =
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING",
+ "score INT",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND")
+ .producedValues(
+ Row.of("Bob", 1, Instant.ofEpochMilli(0)),
+ Row.of("Alice", 1, Instant.ofEpochMilli(1)),
+ Row.of("Bob", 2, Instant.ofEpochMilli(2)),
+ Row.of("Bob", 3, Instant.ofEpochMilli(3)),
+ Row.of("Bob", 4, Instant.ofEpochMilli(4)),
+ Row.of("Bob", 5, Instant.ofEpochMilli(5)),
+ Row.of("Bob", 6, Instant.ofEpochMilli(6)))
+ .build();
+
+ private static final SourceTestStep TIMED_SOURCE_LATE_EVENTS =
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING",
+ "score INT",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND")
+ .producedValues(
+ Row.of("Bob", 1, Instant.ofEpochMilli(0)),
+ Row.of("Alice", 1, Instant.ofEpochMilli(1)),
+ Row.of("Bob", 2, Instant.ofEpochMilli(99999)),
+ Row.of("Bob", 3, Instant.ofEpochMilli(3)),
+ Row.of("Bob", 4, Instant.ofEpochMilli(4)))
+ .build();
+
/** Corresponds to {@link TestProcessTableFunctionBase}. */
private static final String BASE_SINK_SCHEMA = "`out` STRING";
+ /** Corresponds to {@link TestProcessTableFunctionBase}. */
+ private static final String TIMED_BASE_SINK_SCHEMA = "`out` STRING, `rowtime` TIMESTAMP_LTZ(3)";
+
+ /** Corresponds to {@link TestProcessTableFunctionBase}. */
+ private static final String KEYED_TIMED_BASE_SINK_SCHEMA =
+ "`name` STRING, `out` STRING, `rowtime` TIMESTAMP_LTZ(3)";
+
/** Corresponds to {@link TestProcessTableFunctionBase}. */
private static final String KEYED_BASE_SINK_SCHEMA = "`name` STRING, `out` STRING";
@@ -359,4 +416,334 @@ public class ProcessTableFunctionTestPrograms {
.runSql(
"INSERT INTO sink SELECT * FROM f(columnList1 => NULL, columnList3 => DESCRIPTOR(a, b, c))")
.build();
+
+ public static final TableTestProgram PROCESS_TIME_CONVERSIONS =
+ TableTestProgram.of(
+ "process-time-conversions",
+ "test all support conversion classes for both time and watermarks")
+ .setupTemporarySystemFunction("f", TimeConversionsFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[{Time (Long: 0, Instant: 1970-01-01T00:00:00Z, LocalDateTime: 1970-01-01T00:00), "
+ + "Watermark (Long: null, Instant: null, LocalDateTime: null)}, 1970-01-01T00:00:00Z]",
+ "+I[{Time (Long: 1, Instant: 1970-01-01T00:00:00.001Z, LocalDateTime: 1970-01-01T00:00:00.001), "
+ + "Watermark (Long: -1, Instant: 1969-12-31T23:59:59.999Z, LocalDateTime: 1969-12-31T23:59:59.999)}, 1970-01-01T00:00:00.001Z]",
+ "+I[{Time (Long: 2, Instant: 1970-01-01T00:00:00.002Z, LocalDateTime: 1970-01-01T00:00:00.002), "
+ + "Watermark (Long: 0, Instant: 1970-01-01T00:00:00Z, LocalDateTime: 1970-01-01T00:00)}, 1970-01-01T00:00:00.002Z]",
+ "+I[{Time (Long: 3, Instant: 1970-01-01T00:00:00.003Z, LocalDateTime: 1970-01-01T00:00:00.003), "
+ + "Watermark (Long: 1, Instant: 1970-01-01T00:00:00.001Z, LocalDateTime: 1970-01-01T00:00:00.001)}, 1970-01-01T00:00:00.003Z]",
+ "+I[{Time (Long: 4, Instant: 1970-01-01T00:00:00.004Z, LocalDateTime: 1970-01-01T00:00:00.004), "
+ + "Watermark (Long: 2, Instant: 1970-01-01T00:00:00.002Z, LocalDateTime: 1970-01-01T00:00:00.002)}, 1970-01-01T00:00:00.004Z]",
+ "+I[{Time (Long: 5, Instant: 1970-01-01T00:00:00.005Z, LocalDateTime: 1970-01-01T00:00:00.005),"
+ + " Watermark (Long: 3, Instant: 1970-01-01T00:00:00.003Z, LocalDateTime: 1970-01-01T00:00:00.003)}, 1970-01-01T00:00:00.005Z]",
+ "+I[{Time (Long: 6, Instant: 1970-01-01T00:00:00.006Z, LocalDateTime: 1970-01-01T00:00:00.006), "
+ + "Watermark (Long: 4, Instant: 1970-01-01T00:00:00.004Z, LocalDateTime: 1970-01-01T00:00:00.004)}, 1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_REGULAR_TIMESTAMP =
+ TableTestProgram.of(
+ "process-regular-timestamp",
+ "tests regular TIMESTAMP type instead of TIMESTAMP_LTZ")
+ .setupTemporarySystemFunction("f", TimeConversionsFunction.class)
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING",
+ "score INT",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL '1' HOUR")
+ .producedValues(
+ Row.of("Bob", 1, LocalDateTime.of(2025, 1, 1, 1, 0)),
+ Row.of("Alice", 1, LocalDateTime.of(2025, 1, 1, 2, 0)),
+ Row.of("Bob", 2, LocalDateTime.of(2025, 1, 1, 3, 0)),
+ Row.of("Bob", 3, LocalDateTime.of(2025, 1, 1, 4, 0)),
+ Row.of("Bob", 4, LocalDateTime.of(2025, 1, 1, 5, 0)),
+ Row.of("Bob", 5, LocalDateTime.of(2025, 1, 1, 6, 0)),
+ Row.of("Bob", 6, LocalDateTime.of(2025, 1, 1, 7, 0)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("`out` STRING, `rowtime` TIMESTAMP(3)")
+ .consumedValues(
+ "+I[{Time (Long: 1735693200000, Instant: 2025-01-01T01:00:00Z, LocalDateTime: 2025-01-01T01:00), Watermark (Long: null, Instant: null, LocalDateTime: null)}, 2025-01-01T01:00]",
+ "+I[{Time (Long: 1735696800000, Instant: 2025-01-01T02:00:00Z, LocalDateTime: 2025-01-01T02:00), Watermark (Long: 1735689600000, Instant: 2025-01-01T00:00:00Z, LocalDateTime: 2025-01-01T00:00)}, 2025-01-01T02:00]",
+ "+I[{Time (Long: 1735700400000, Instant: 2025-01-01T03:00:00Z, LocalDateTime: 2025-01-01T03:00), Watermark (Long: 1735693200000, Instant: 2025-01-01T01:00:00Z, LocalDateTime: 2025-01-01T01:00)}, 2025-01-01T03:00]",
+ "+I[{Time (Long: 1735704000000, Instant: 2025-01-01T04:00:00Z, LocalDateTime: 2025-01-01T04:00), Watermark (Long: 1735696800000, Instant: 2025-01-01T02:00:00Z, LocalDateTime: 2025-01-01T02:00)}, 2025-01-01T04:00]",
+ "+I[{Time (Long: 1735707600000, Instant: 2025-01-01T05:00:00Z, LocalDateTime: 2025-01-01T05:00), Watermark (Long: 1735700400000, Instant: 2025-01-01T03:00:00Z, LocalDateTime: 2025-01-01T03:00)}, 2025-01-01T05:00]",
+ "+I[{Time (Long: 1735711200000, Instant: 2025-01-01T06:00:00Z, LocalDateTime: 2025-01-01T06:00), Watermark (Long: 1735704000000, Instant: 2025-01-01T04:00:00Z, LocalDateTime: 2025-01-01T04:00)}, 2025-01-01T06:00]",
+ "+I[{Time (Long: 1735714800000, Instant: 2025-01-01T07:00:00Z, LocalDateTime: 2025-01-01T07:00), Watermark (Long: 1735707600000, Instant: 2025-01-01T05:00:00Z, LocalDateTime: 2025-01-01T05:00)}, 2025-01-01T07:00]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_NAMED_TIMERS =
+ TableTestProgram.of(
+ "process-partitioned-named-timers",
+ "test create/fire/replace/clear/clear-all named timers")
+ .setupTemporarySystemFunction("f", NamedTimersFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer timeout1 for 1 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer timeout2 for 2 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer timeout3 for 3 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer timeout4 for 9223372036854775807 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer timeout5 for 9223372036854775807 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Timer timeout1 fired at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Registering timer timeout2 for 5 at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Clearing timer timeout3 at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Processing input row +I[Bob, 3, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Processing input row +I[Bob, 4, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Processing input row +I[Bob, 5, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Processing input row +I[Bob, 6, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z]",
+ "+I[Bob, {Timer timeout2 fired at time 5 watermark 5}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Clearing all timers at time 5 watermark 5}, 1970-01-01T00:00:00.005Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_UNNAMED_TIMERS =
+ TableTestProgram.of(
+ "process-partitioned-unnamed-timers",
+ "test create/fire/re-create/clear/clear-all unnamed timers")
+ .setupTemporarySystemFunction("f", UnnamedTimersFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer for 1 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer for 2 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer for 9223372036854775807 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Timer null fired at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Registering timer for 5 at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Clearing timer 2 at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Processing input row +I[Bob, 3, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Processing input row +I[Bob, 4, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Processing input row +I[Bob, 5, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Processing input row +I[Bob, 6, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z]",
+ "+I[Bob, {Timer null fired at time 5 watermark 5}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Clearing all timers at time 5 watermark 5}, 1970-01-01T00:00:00.005Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_LATE_EVENTS =
+ TableTestProgram.of(
+ "process-late-events",
+ "test that late events are dropped in both input and when registering timers")
+ .setupTemporarySystemFunction("f", LateTimersFunction.class)
+ .setupTableSource(TIMED_SOURCE_LATE_EVENTS)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer t for 0 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer for 0 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer t for 0 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer for 0 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Timer null fired at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Timer null fired at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Registering timer again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Timer t fired at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Timer t fired at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Registering timer again for 0 at time 0 watermark 0}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:01:39.999Z] at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]",
+ "+I[Bob, {Registering timer t for 0 at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]",
+ "+I[Bob, {Registering timer for 0 at time 99999 watermark 0}, 1970-01-01T00:01:39.999Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_SCALAR_ARGS_TIME =
+ TableTestProgram.of(
+ "process-scalar-args-time",
+ "test time is not available for PTFs without tables")
+ .setupTemporarySystemFunction("f", ScalarArgsTimeFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(BASE_SINK_SCHEMA)
+ .consumedValues("+I[{Time null and watermark null}]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f()")
+ .build();
+
+ public static final TableTestProgram PROCESS_OPTIONAL_PARTITION_BY_TIME =
+ TableTestProgram.of(
+ "process-optional-partition-by-time",
+ "test time and timers without PARTITION BY")
+ .setupTemporarySystemFunction("f", OptionalPartitionOnTimeFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[{Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[{Registering timer t for 1 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[{Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[{Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z]",
+ "+I[{Timer t fired at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[{Registering timer again for 2 at time 1 watermark 1}, 1970-01-01T00:00:00.001Z]",
+ "+I[{Processing input row +I[Bob, 3, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z]",
+ "+I[{Timer again fired at time 2 watermark 2}, 1970-01-01T00:00:00.002Z]",
+ "+I[{Processing input row +I[Bob, 4, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z]",
+ "+I[{Processing input row +I[Bob, 5, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z]",
+ "+I[{Processing input row +I[Bob, 6, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_OPTIONAL_ON_TIME =
+ TableTestProgram.of(
+ "process-optional-on-time",
+ "test optional time attribute, fire once for constant timer")
+ .setupTemporarySystemFunction("f", OptionalOnTimeFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time null watermark null}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark null}]",
+ "+I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time null watermark -1}]",
+ "+I[Alice, {Registering timer t for 2 at time null watermark -1}]",
+ "+I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time null watermark 0}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark 0}]",
+ "+I[Bob, {Processing input row +I[Bob, 3, 1970-01-01T00:00:00.003Z] at time null watermark 1}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark 1}]",
+ "+I[Alice, {Timer t fired at time 2 watermark 2}]",
+ "+I[Bob, {Timer t fired at time 2 watermark 2}]",
+ "+I[Bob, {Processing input row +I[Bob, 4, 1970-01-01T00:00:00.004Z] at time null watermark 2}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark 2}]",
+ "+I[Bob, {Processing input row +I[Bob, 5, 1970-01-01T00:00:00.005Z] at time null watermark 3}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark 3}]",
+ "+I[Bob, {Processing input row +I[Bob, 6, 1970-01-01T00:00:00.006Z] at time null watermark 4}]",
+ "+I[Bob, {Registering timer t for 2 at time null watermark 4}]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name)")
+ .build();
+
+ public static final TableTestProgram PROCESS_POJO_STATE_TIME =
+ TableTestProgram.of(
+ "process-pojo-state-time",
+ "test state access from both eval() and onTimer()")
+ .setupTemporarySystemFunction("f", PojoStateTimeFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Score(s='null', i=null), +I[Bob, 1, 1970-01-01T00:00:00Z]}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Registering timer t for 2 at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Score(s='null', i=null), +I[Alice, 1, 1970-01-01T00:00:00.001Z]}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Registering timer t for 3 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Score(s='null', i=1), +I[Bob, 2, 1970-01-01T00:00:00.002Z]}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Score(s='null', i=2), +I[Bob, 3, 1970-01-01T00:00:00.003Z]}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Timer t fired at time 2 watermark 2}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Score(s='null', i=30), +I[Bob, 4, 1970-01-01T00:00:00.004Z]}, 1970-01-01T00:00:00.004Z]",
+ "+I[Alice, {Timer t fired at time 3 watermark 3}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Score(s='null', i=31), +I[Bob, 5, 1970-01-01T00:00:00.005Z]}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Score(s='null', i=32), +I[Bob, 6, 1970-01-01T00:00:00.006Z]}, 1970-01-01T00:00:00.006Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
+ public static final TableTestProgram PROCESS_CHAINED_TIME =
+ TableTestProgram.of(
+ "process-chained-time",
+ "test two chained PTFs where the second (receiving) PTF wraps the events of the first (sending) PTF")
+ .setupTemporarySystemFunction("f1", ChainedSendingFunction.class)
+ .setupTemporarySystemFunction("f2", ChainedReceivingFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 1, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 1 at time 0 watermark null}, 1970-01-01T00:00:00Z] at time 0 watermark null}, 1970-01-01T00:00:00Z]",
+ "+I[Alice, {Processing input row +I[Alice, {Processing input row +I[Alice, 1, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Alice, {Processing input row +I[Alice, {Registering timer t for 2 at time 1 watermark -1}, 1970-01-01T00:00:00.001Z] at time 1 watermark -1}, 1970-01-01T00:00:00.001Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 2, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 3 at time 2 watermark 0}, 1970-01-01T00:00:00.002Z] at time 2 watermark 0}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 3, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 4 at time 3 watermark 1}, 1970-01-01T00:00:00.003Z] at time 3 watermark 1}, 1970-01-01T00:00:00.003Z]",
+ "+I[Alice, {Processing input row +I[Alice, {Timer t fired at time 2 watermark 2}, 1970-01-01T00:00:00.002Z] at time 2 watermark 1}, 1970-01-01T00:00:00.002Z]",
+ "+I[Alice, {Processing input row +I[Alice, {2}, 1970-01-01T00:00:00.002Z] at time 2 watermark 1}, 1970-01-01T00:00:00.002Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 4, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 5 at time 4 watermark 2}, 1970-01-01T00:00:00.004Z] at time 4 watermark 2}, 1970-01-01T00:00:00.004Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 5, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 6 at time 5 watermark 3}, 1970-01-01T00:00:00.005Z] at time 5 watermark 3}, 1970-01-01T00:00:00.005Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Processing input row +I[Bob, 6, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Registering timer t for 7 at time 6 watermark 4}, 1970-01-01T00:00:00.006Z] at time 6 watermark 4}, 1970-01-01T00:00:00.006Z]",
+ "+I[Bob, {Processing input row +I[Bob, {Timer t fired at time 7 watermark 9223372036854775807}, 1970-01-01T00:00:00.007Z] at time 7 watermark 5}, 1970-01-01T00:00:00.007Z]",
+ "+I[Bob, {Processing input row +I[Bob, {7}, 1970-01-01T00:00:00.007Z] at time 7 watermark 5}, 1970-01-01T00:00:00.007Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "WITH "
+ + "ptf1 AS (SELECT * FROM f1(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))), "
+ + "ptf2 AS (SELECT * FROM f2(r => TABLE ptf1 PARTITION BY name, on_time => DESCRIPTOR(rowtime))) "
+ + "SELECT * FROM ptf2")
+ .build();
+
+ public static final TableTestProgram PROCESS_INVALID_TABLE_AS_ROW_TIMERS =
+ TableTestProgram.of(
+ "process-invalid-table-as-row-timers",
+ "error if timers are registered for PTFs with row semantic tables")
+ .setupTemporarySystemFunction("f", InvalidTableAsRowTimersFunction.class)
+ .setupSql(BASIC_VALUES)
+ .runFailingSql(
+ "SELECT * FROM f(r => TABLE t)",
+ TableRuntimeException.class,
+ "Timers are not supported in the current PTF declaration.")
+ .build();
+
+ public static final TableTestProgram PROCESS_INVALID_PASS_THROUGH_TIMERS =
+ TableTestProgram.of(
+ "process-invalid-pass-through-timers",
+ "error if timers are registered for PTFs with pass-through columns")
+ .setupTemporarySystemFunction("f", InvalidPassThroughTimersFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .runFailingSql(
+ "SELECT * FROM f(r => TABLE t PARTITION BY name, on_time => DESCRIPTOR(ts))",
+ TableRuntimeException.class,
+ "Timers are not supported in the current PTF declaration.")
+ .build();
+
+ public static final TableTestProgram PROCESS_INVALID_UPDATING_TIMERS =
+ TableTestProgram.of(
+ "process-invalid-updating-timers",
+ "error if timers are registered for PTFs with updating inputs")
+ .setupTemporarySystemFunction("f", InvalidUpdatingTimersFunction.class)
+ .setupSql(UPDATING_VALUES)
+ .runFailingSql(
+ "SELECT * FROM f(r => TABLE t PARTITION BY name)",
+ TableRuntimeException.class,
+ "Timers are not supported in the current PTF declaration.")
+ .build();
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index c633f395fd0..b5457cc3d3d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -29,13 +29,18 @@ import org.apache.flink.table.runtime.operators.process.ProcessTableOperator;
import org.apache.flink.types.ColumnList;
import org.apache.flink.types.Row;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
import static org.apache.flink.table.annotation.ArgumentTrait.OPTIONAL_PARTITION_BY;
import static org.apache.flink.table.annotation.ArgumentTrait.PASS_COLUMNS_THROUGH;
+import static org.apache.flink.table.annotation.ArgumentTrait.REQUIRE_ON_TIME;
import static org.apache.flink.table.annotation.ArgumentTrait.SUPPORT_UPDATES;
import static org.apache.flink.table.annotation.ArgumentTrait.TABLE_AS_ROW;
import static org.apache.flink.table.annotation.ArgumentTrait.TABLE_AS_SET;
/** Testing functions for {@link ProcessTableFunction}. */
+@SuppressWarnings("unused")
public class ProcessTableFunctionTestUtils {
/** Testing function. */
@@ -47,13 +52,68 @@ public class ProcessTableFunctionTestUtils {
@DataTypeHint("ROW<`out` STRING>")
public abstract static class TestProcessTableFunctionBase extends ProcessTableFunction<Row> {
- public void collectObjects(Object... objects) {
+ protected void collectObjects(Object... objects) {
// Row.toString is useful because it can handle all common objects,
// but we use '{}' to indicate that this is a custom string output.
final String asString = Row.of(objects).toString();
final String objectsAsString = asString.substring(3, asString.length() - 1);
collect(Row.of("{" + objectsAsString + "}"));
}
+
+ protected void collectEvalEvent(TimeContext<Long> timeCtx, Row r) {
+ collectObjects(
+ String.format(
+ "Processing input row %s at time %s watermark %s",
+ r, timeCtx.time(), timeCtx.currentWatermark()));
+ }
+
+ protected void collectOnTimerEvent(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectObjects(
+ String.format(
+ "Timer %s fired at time %s watermark %s",
+ ctx.currentTimer(), timeCtx.time(), timeCtx.currentWatermark()));
+ }
+
+ protected void collectCreateTimer(TimeContext<Long> timeCtx, String name, long time) {
+ collectObjects(
+ String.format(
+ "Registering timer %s for %s at time %s watermark %s",
+ name, time, timeCtx.time(), timeCtx.currentWatermark()));
+ timeCtx.registerOnTime(name, time);
+ }
+
+ protected void collectCreateTimer(TimeContext<Long> timeCtx, long time) {
+ collectObjects(
+ String.format(
+ "Registering timer for %s at time %s watermark %s",
+ time, timeCtx.time(), timeCtx.currentWatermark()));
+ timeCtx.registerOnTime(time);
+ }
+
+ protected void collectClearTimer(TimeContext<Long> timeCtx, String name) {
+ collectObjects(
+ String.format(
+ "Clearing timer %s at time %s watermark %s",
+ name, timeCtx.time(), timeCtx.currentWatermark()));
+ timeCtx.clearTimer(name);
+ }
+
+ protected void collectClearTimer(TimeContext<Long> timeCtx, long time) {
+ collectObjects(
+ String.format(
+ "Clearing timer %s at time %s watermark %s",
+ time, timeCtx.time(), timeCtx.currentWatermark()));
+ timeCtx.clearTimer(time);
+ }
+
+ protected void collectClearAllTimers(TimeContext<Long> timeCtx) {
+ collectObjects(
+ String.format(
+ "Clearing all timers at time %s watermark %s",
+ timeCtx.time(), timeCtx.currentWatermark()));
+ timeCtx.clearAllTimers();
+ }
}
/** Testing function. */
@@ -208,7 +268,6 @@ public class ProcessTableFunctionTestUtils {
/** Testing function. */
public static class TimeToLiveStateFunction extends ProcessTableFunction<String> {
- @SuppressWarnings("unused")
public void eval(
Context ctx,
@StateHint(type = @DataTypeHint("ROW<emitted BOOLEAN>")) Row s0,
@@ -216,8 +275,8 @@ public class ProcessTableFunctionTestUtils {
@StateHint(ttl = "0") Score s2,
@StateHint Score s3,
@ArgumentHint({TABLE_AS_SET, OPTIONAL_PARTITION_BY}) Row r) {
- final ProcessTableOperator.ProcessFunctionContext internalContext =
- (ProcessTableOperator.ProcessFunctionContext) ctx;
+ final ProcessTableOperator.RunnerContext internalContext =
+ (ProcessTableOperator.RunnerContext) ctx;
if (s0.getFieldAs("emitted") == null) {
collect(
String.format(
@@ -240,6 +299,241 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class RequiredTimeFunction extends TestProcessTableFunctionBase {
+ public void eval(@ArgumentHint({ArgumentTrait.TABLE_AS_ROW, REQUIRE_ON_TIME}) Row r) {
+ collectObjects(r);
+ }
+ }
+
+ /** Testing function. */
+ public static class TimeConversionsFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint({TABLE_AS_ROW, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> asLong = ctx.timeContext(Long.class);
+ final TimeContext<Instant> asInstant = ctx.timeContext(Instant.class);
+ final TimeContext<LocalDateTime> asLocalDateTime = ctx.timeContext(LocalDateTime.class);
+ // read
+ collectObjects(
+ String.format(
+ "Time (Long: %s, Instant: %s, LocalDateTime: %s), "
+ + "Watermark (Long: %s, Instant: %s, LocalDateTime: %s)",
+ asLong.time(),
+ asInstant.time(),
+ asLocalDateTime.time(),
+ asLong.currentWatermark(),
+ asInstant.currentWatermark(),
+ asLocalDateTime.currentWatermark()));
+ }
+ }
+
+ /** Testing function. */
+ @SuppressWarnings("SameParameterValue")
+ public static class NamedTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint({TABLE_AS_SET, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+
+ if (timeCtx.time() == 0) {
+ // create
+ collectCreateTimer(timeCtx, "timeout1", timeCtx.time() + 1);
+ collectCreateTimer(timeCtx, "timeout2", timeCtx.time() + 2);
+ collectCreateTimer(timeCtx, "timeout3", timeCtx.time() + 3);
+ collectCreateTimer(timeCtx, "timeout4", Long.MAX_VALUE);
+ collectCreateTimer(timeCtx, "timeout5", Long.MAX_VALUE);
+ }
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ // read
+ collectOnTimerEvent(ctx);
+ if (ctx.currentTimer().equals("timeout1")) {
+ // replace
+ collectCreateTimer(timeCtx, "timeout2", timeCtx.time() + 4);
+ // clear
+ collectClearTimer(timeCtx, "timeout3");
+ } else if (ctx.currentTimer().equals("timeout2")) {
+ // clear all
+ collectClearAllTimers(timeCtx);
+ }
+ }
+ }
+
+ /** Testing function. */
+ @SuppressWarnings("SameParameterValue")
+ public static class UnnamedTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint({TABLE_AS_SET, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+
+ if (timeCtx.time() == 0) {
+ // create
+ collectCreateTimer(timeCtx, timeCtx.time() + 1);
+ collectCreateTimer(timeCtx, timeCtx.time() + 2);
+ collectCreateTimer(timeCtx, Long.MAX_VALUE);
+ }
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ // read
+ collectOnTimerEvent(ctx);
+ if (timeCtx.time() == 1) {
+ // re-create
+ collectCreateTimer(timeCtx, timeCtx.time() + 4);
+ // clear
+ collectClearTimer(timeCtx, 2);
+ } else if (timeCtx.time() == 5) {
+ // clear all
+ collectClearAllTimers(timeCtx);
+ }
+ }
+ }
+
+ /** Testing function. */
+ public static class LateTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint({TABLE_AS_SET, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+ // all timers should be executed once
+ if (timeCtx.time() == 99998) {
+ // will never be fired because it's late
+ collectCreateTimer(timeCtx, "late", 1L);
+ }
+ collectCreateTimer(timeCtx, "t", 0L);
+ collectCreateTimer(timeCtx, 0L);
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectOnTimerEvent(ctx);
+ // will never be fired because it's late
+ collectCreateTimer(timeCtx, "again", timeCtx.time());
+ }
+ }
+
+ /** Testing function. */
+ public static class ScalarArgsTimeFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectObjects(
+ String.format(
+ "Time %s and watermark %s",
+ timeCtx.time(), timeCtx.currentWatermark()));
+ }
+ }
+
+ /** Testing function. */
+ public static class OptionalPartitionOnTimeFunction extends TestProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @ArgumentHint({TABLE_AS_SET, OPTIONAL_PARTITION_BY, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+ if (timeCtx.time() == 0) {
+ collectCreateTimer(timeCtx, "t", timeCtx.time() + 1);
+ }
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectOnTimerEvent(ctx);
+ if (ctx.currentTimer().equals("t")) {
+ collectCreateTimer(timeCtx, "again", timeCtx.time() + 1);
+ }
+ }
+ }
+
+ /** Testing function. */
+ public static class OptionalOnTimeFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint(TABLE_AS_SET) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+ collectCreateTimer(timeCtx, "t", 2);
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectOnTimerEvent(ctx);
+ }
+ }
+
+ /** Testing function. */
+ public static class PojoStateTimeFunction extends TestProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint Score s,
+ @ArgumentHint({TABLE_AS_SET, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectObjects(s, r);
+ if (s.i == null) {
+ s.i = 1;
+ collectCreateTimer(timeCtx, "t", timeCtx.time() + 2);
+ } else {
+ s.i += 1;
+ }
+ }
+
+ public void onTimer(OnTimerContext ctx, Score s) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectOnTimerEvent(ctx);
+ s.i *= 10;
+ }
+ }
+
+ /** Testing function. */
+ public static class ChainedSendingFunction extends TestProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @ArgumentHint({TABLE_AS_SET, OPTIONAL_PARTITION_BY, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+ collectCreateTimer(timeCtx, "t", timeCtx.time() + 1);
+ }
+
+ public void onTimer(OnTimerContext ctx) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectOnTimerEvent(ctx);
+ collectObjects(timeCtx.time());
+ }
+ }
+
+ /** Testing function. */
+ public static class ChainedReceivingFunction extends TestProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @ArgumentHint({TABLE_AS_SET, OPTIONAL_PARTITION_BY, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ collectEvalEvent(timeCtx, r);
+ }
+ }
+
+ /** Testing function. */
+ public static class InvalidTableAsRowTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint(TABLE_AS_ROW) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ timeCtx.registerOnTime(42L);
+ }
+ }
+
+ /** Testing function. */
+ public static class InvalidPassThroughTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @ArgumentHint({TABLE_AS_SET, PASS_COLUMNS_THROUGH, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ timeCtx.registerOnTime(42L);
+ }
+ }
+
+ /** Testing function. */
+ public static class InvalidUpdatingTimersFunction extends TestProcessTableFunctionBase {
+ public void eval(Context ctx, @ArgumentHint({TABLE_AS_SET, SUPPORT_UPDATES}) Row r) {
+ final TimeContext<Long> timeCtx = ctx.timeContext(Long.class);
+ timeCtx.registerOnTime(42L);
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Helpers
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java
index 2d8f4546996..2564a6124bb 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.table.test.program.ConfigOptionTestStep;
+import org.apache.flink.table.test.program.FailingSqlTestStep;
import org.apache.flink.table.test.program.FunctionTestStep;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
@@ -65,7 +66,7 @@ public abstract class SemanticTestBase implements TableTestProgramRunner {
@Override
public EnumSet<TestKind> supportedRunSteps() {
- return EnumSet.of(TestKind.SQL);
+ return EnumSet.of(TestKind.SQL, TestKind.FAILING_SQL);
}
@AfterEach
@@ -119,7 +120,22 @@ public abstract class SemanticTestBase implements TableTestProgramRunner {
}
}
- program.getRunSqlTestStep().apply(env).await();
+ for (TestStep testStep : program.runSteps) {
+ switch (testStep.getKind()) {
+ case SQL:
+ {
+ final SqlTestStep sqlTestStep = (SqlTestStep) testStep;
+ sqlTestStep.apply(env).await();
+ }
+ break;
+ case FAILING_SQL:
+ {
+ final FailingSqlTestStep sqlTestStep = (FailingSqlTestStep) testStep;
+ sqlTestStep.apply(env);
+ }
+ break;
+ }
+ }
for (SinkTestStep sinkTestStep : program.getSetupSinkTestSteps()) {
List<String> actualResults = getActualResults(sinkTestStep, sinkTestStep.name);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
index cd7a295b45a..ba2ce3916ce 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.RequiredTimeFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ScalarArgsFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowFunction;
import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TableAsRowPassThroughFunction;
@@ -76,6 +77,10 @@ public class ProcessTableFunctionTest extends TableTestBase {
.executeSql("CREATE VIEW t_type_diff AS SELECT 'Bob' AS name, TRUE AS isValid");
util.tableEnv()
.executeSql("CREATE VIEW t_updating AS SELECT name, COUNT(*) FROM t GROUP BY name");
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE t_watermarked (name STRING, score INT, ts TIMESTAMP_LTZ(3), WATERMARK FOR ts AS ts) "
+ + "WITH ('connector' = 'datagen')");
util.tableEnv()
.executeSql("CREATE TABLE t_sink (`out` STRING) WITH ('connector' = 'blackhole')");
util.tableEnv()
@@ -145,6 +150,13 @@ public class ProcessTableFunctionTest extends TableTestBase {
+ "columnList3 => DESCRIPTOR())");
}
+ @Test
+ void testOnTime() {
+ util.addTemporarySystemFunction("f", RequiredTimeFunction.class);
+ util.verifyRelPlan(
+ "SELECT `out`, `rowtime` FROM f(r => TABLE t_watermarked, on_time => DESCRIPTOR(ts))");
+ }
+
@Test
void testMissingUid() {
// Function name contains special characters and can thus not be used as UID
@@ -232,7 +244,7 @@ public class ProcessTableFunctionTest extends TableTestBase {
// function expects <STRING name, INT score>
"SELECT * FROM f(u => TABLE t_type_diff, i => 1)",
"No match found for function signature "
- + "f(<RecordType(CHAR(3) name, BOOLEAN isValid)>, <NUMERIC>, <CHARACTER>)"),
+ + "f(<RecordType(CHAR(3) name, BOOLEAN isValid)>, <NUMERIC>, <COLUMN_LIST>, <CHARACTER>)"),
ErrorSpec.of(
"table as set with missing partition by",
TableAsSetFunction.class,
@@ -244,7 +256,7 @@ public class ProcessTableFunctionTest extends TableTestBase {
// function expects <STRING name, INT score>
"SELECT * FROM f(u => TABLE t_type_diff PARTITION BY name, i => 1)",
"No match found for function signature "
- + "f(<RecordType(CHAR(3) name, BOOLEAN isValid)>, <NUMERIC>, <CHARACTER>)"),
+ + "f(<RecordType(CHAR(3) name, BOOLEAN isValid)>, <NUMERIC>, <COLUMN_LIST>, <CHARACTER>)"),
ErrorSpec.of(
"table function instead of process table function",
NoProcessTableFunction.class,
@@ -255,7 +267,7 @@ public class ProcessTableFunctionTest extends TableTestBase {
"reserved args",
ReservedArgFunction.class,
"SELECT * FROM f(uid => 'my-ptf')",
- "Function signature must not declare system arguments. Reserved argument names are: [uid]"),
+ "Function signature must not declare system arguments. Reserved argument names are: [on_time, uid]"),
ErrorSpec.of(
"multiple table args",
MultiTableFunction.class,
@@ -308,7 +320,29 @@ public class ProcessTableFunctionTest extends TableTestBase {
"invalid descriptor",
DescriptorFunction.class,
"SELECT * FROM f(columnList1 => NULL, columnList3 => DESCRIPTOR(b.INVALID))",
- "column alias must be a simple identifier"));
+ "column alias must be a simple identifier"),
+ ErrorSpec.of(
+ "ambiguous on_time reference",
+ TableAsRowFunction.class,
+ "WITH duplicate_ts AS (SELECT ts AS ts1, ts AS ts2 FROM t_watermarked) "
+ + "SELECT * FROM f(r => TABLE duplicate_ts, i => 1, on_time => DESCRIPTOR(ts1, ts2))",
+ "Ambiguous time attribute found. The `on_time` argument must reference at "
+ + "most one column in a table argument. Currently, the columns in "
+ + "`on_time` point to both 'ts1' and 'ts2' in table argument 'r'."),
+ ErrorSpec.of(
+ "invalid on_time data type",
+ RequiredTimeFunction.class,
+ "SELECT * FROM f(r => TABLE t_watermarked)",
+ "Table argument 'r' requires a time attribute. "
+ + "Please provide one using the implicit `on_time` argument. "
+ + "For example: myFunction(..., on_time => DESCRIPTOR(`my_timestamp`)"),
+ ErrorSpec.of(
+ "invalid on_time column",
+ TableAsRowFunction.class,
+ "SELECT * FROM f(r => TABLE t_watermarked, i => 1, on_time => DESCRIPTOR(ts, INVALID))",
+ "Invalid time attribute declaration. Each column in the `on_time` argument must "
+ + "reference at least one column in one of the table arguments. "
+ + "Unknown references: [INVALID]"));
}
/** Testing function. */
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.xml
index 9a300e42e34..e71a198ba6e 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.xml
@@ -23,12 +23,12 @@ limitations under the License.
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(DESCRIPTOR(_UTF-16LE'a'), DESCRIPTOR(_UTF-16LE'b', _UTF-16LE'c'), DESCRIPTOR(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(DESCRIPTOR(_UTF-16LE'a'), DESCRIPTOR(_UTF-16LE'b', _UTF-16LE'c'), DESCRIPTOR(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(DESCRIPTOR(_UTF-16LE'a'), DESCRIPTOR(_UTF-16LE'b', _UTF-16LE'c'), DESCRIPTOR(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(DESCRIPTOR(_UTF-16LE'a'), DESCRIPTOR(_UTF-16LE'b', _UTF-16LE'c'), DESCRIPTOR(), DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Values(tuples=[[{ }]])
]]>
</Resource>
@@ -40,7 +40,7 @@ ProcessTableFunction(invocation=[f(DESCRIPTOR(_UTF-16LE'a'), DESCRIPTOR(_UTF-16L
<Resource name="ast">
<![CDATA[
LogicalProject(score=[$0], out=[$1])
-+- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($1), 1, DEFAULT())], rowType=[RecordType(INTEGER score, VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($1), 1, DEFAULT(), DEFAULT())], rowType=[RecordType(INTEGER score, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -48,7 +48,7 @@ LogicalProject(score=[$0], out=[$1])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($1), 1, DEFAULT())], uid=[f], select=[score,out], rowType=[RecordType(INTEGER score, VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($1), 1, DEFAULT(), DEFAULT())], uid=[f], select=[score,out], rowType=[RecordType(INTEGER score, VARCHAR(2147483647) out)])
+- Exchange(distribution=[hash[score]])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
]]>
@@ -61,13 +61,34 @@ ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($1), 1, DEFAULT())], u
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(_UTF-16LE'my-ptf')], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(DEFAULT(), _UTF-16LE'my-ptf')], rowType=[RecordType(VARCHAR(2147483647) out)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(_UTF-16LE'my-ptf')], uid=[my-ptf], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(DEFAULT(), _UTF-16LE'my-ptf')], uid=[my-ptf], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Values(tuples=[[{ }]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testOnTime">
+ <Resource name="sql">
+ <![CDATA[SELECT `out`, `rowtime` FROM f(r => TABLE t_watermarked, on_time => DESCRIPTOR(ts))]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(out=[$0], rowtime=[$1])
++- LogicalTableFunctionScan(invocation=[f(TABLE(#0), DESCRIPTOR(_UTF-16LE'ts'), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out, TIMESTAMP_LTZ(3) *ROWTIME* rowtime)])
+ +- LogicalProject(name=[$0], score=[$1], ts=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[ts], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, t_watermarked]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+ProcessTableFunction(invocation=[f(TABLE(#0), DESCRIPTOR(_UTF-16LE'ts'), DEFAULT())], uid=[null], select=[out,rowtime], rowType=[RecordType(VARCHAR(2147483647) out, TIMESTAMP_LTZ(3) *ROWTIME* rowtime)])
++- WatermarkAssigner(rowtime=[ts], watermark=[ts])
+ +- TableSourceScan(table=[[default_catalog, default_database, t_watermarked]], fields=[name, score, ts])
]]>
</Resource>
</TestCase>
@@ -78,12 +99,12 @@ ProcessTableFunction(invocation=[f(_UTF-16LE'my-ptf')], uid=[my-ptf], select=[ou
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(1, true, DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(1, true, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(1, true, DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(1, true, DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Values(tuples=[[{ }]])
]]>
</Resource>
@@ -93,14 +114,14 @@ ProcessTableFunction(invocation=[f(1, true, DEFAULT())], uid=[null], select=[out
<![CDATA[
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'a')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'a')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'b')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'b')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -112,11 +133,11 @@ Exchange(distribution=[hash[name]])(reuse_id=[1])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
Sink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
-+- ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'a')], uid=[a], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
++- ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'a')], uid=[a], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- Reused(reference_id=[1])
Sink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
-+- ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'b')], uid=[b], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
++- ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'b')], uid=[b], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -128,12 +149,12 @@ Sink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(1, true, _UTF-16LE'my-uid')], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(1, true, DEFAULT(), _UTF-16LE'my-uid')], rowType=[RecordType(VARCHAR(2147483647) out)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(1, true, _UTF-16LE'my-uid')], uid=[my-uid], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(1, true, DEFAULT(), _UTF-16LE'my-uid')], uid=[my-uid], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Values(tuples=[[{ }]])
]]>
</Resource>
@@ -143,14 +164,14 @@ ProcessTableFunction(invocation=[f(1, true, _UTF-16LE'my-uid')], uid=[my-uid], s
<![CDATA[
LogicalSink(table=[default_catalog.default_database.t_sink], fields=[out])
+- LogicalProject(out=[$0])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
LogicalSink(table=[default_catalog.default_database.t_sink], fields=[out])
+- LogicalProject(out=[$0])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 42, DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 42, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -161,11 +182,11 @@ LogicalSink(table=[default_catalog.default_database.t_sink], fields=[out])
Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])(reuse_id=[1])
Sink(table=[default_catalog.default_database.t_sink], fields=[out])
-+- ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
++- ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Reused(reference_id=[1])
Sink(table=[default_catalog.default_database.t_sink], fields=[out])
-+- ProcessTableFunction(invocation=[f(TABLE(#0), 42, DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
++- ProcessTableFunction(invocation=[f(TABLE(#0), 42, DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -177,7 +198,7 @@ Sink(table=[default_catalog.default_database.t_sink], fields=[out])
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], score=[$1], out=[$2])
-+- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT())], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -185,7 +206,7 @@ LogicalProject(name=[$0], score=[$1], out=[$2])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT())], uid=[null], select=[name,score,out], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], uid=[null], select=[name,score,out], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
]]>
</Resource>
@@ -197,7 +218,7 @@ ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT())], uid=[null], select
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], score=[$1], out=[$2])
-+- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT())], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -205,7 +226,7 @@ LogicalProject(name=[$0], score=[$1], out=[$2])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT())], uid=[f], select=[name,score,out], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), DEFAULT())], uid=[f], select=[name,score,out], rowType=[RecordType(VARCHAR(5) name, INTEGER score, VARCHAR(2147483647) out)])
+- Exchange(distribution=[hash[name]])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
]]>
@@ -216,14 +237,14 @@ ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT())], u
<![CDATA[
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -231,7 +252,7 @@ LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name,
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], uid=[same], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])(reuse_id=[1])
+ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], uid=[same], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])(reuse_id=[1])
+- Exchange(distribution=[hash[name]])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -249,7 +270,7 @@ Sink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
+- LogicalFilter(condition=[=($0, _UTF-16LE'Bob')])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -257,7 +278,7 @@ LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name,
LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
+- LogicalProject(name=[$0], out=[$1])
+- LogicalFilter(condition=[=($0, _UTF-16LE'Alice')])
- +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+ +- LogicalTableFunctionScan(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalProject(name=[$0], score=[$1])
+- LogicalValues(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -265,7 +286,7 @@ LogicalSink(table=[default_catalog.default_database.t_keyed_sink], fields=[name,
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, _UTF-16LE'same')], uid=[same], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])(reuse_id=[1])
+ProcessTableFunction(invocation=[f(TABLE(#0) PARTITION BY($0), 1, DEFAULT(), _UTF-16LE'same')], uid=[same], select=[name,out], rowType=[RecordType(VARCHAR(5) name, VARCHAR(2147483647) out)])(reuse_id=[1])
+- Exchange(distribution=[hash[name]])
+- Values(tuples=[[{ _UTF-16LE'Bob', 12 }, { _UTF-16LE'Alice', 42 }]])
@@ -286,12 +307,12 @@ Sink(table=[default_catalog.default_database.t_keyed_sink], fields=[name, out])
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(1, true, DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(1, true, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(1, true, DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(1, true, DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Values(tuples=[[{ }]])
]]>
</Resource>
@@ -303,7 +324,7 @@ ProcessTableFunction(invocation=[f(1, true, DEFAULT())], uid=[null], select=[out
<Resource name="ast">
<![CDATA[
LogicalProject(out=[$0])
-+- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
++- LogicalTableFunctionScan(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) out)])
+- LogicalProject(name=[$0], different=[$1])
+- LogicalProject(name=[_UTF-16LE'Bob'], different=[12])
+- LogicalValues(tuples=[[{ 0 }]])
@@ -311,7 +332,7 @@ LogicalProject(out=[$0])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+ProcessTableFunction(invocation=[f(TABLE(#0), 1, DEFAULT(), DEFAULT())], uid=[null], select=[out], rowType=[RecordType(VARCHAR(2147483647) out)])
+- Calc(select=['Bob' AS name, 12 AS different])
+- Values(tuples=[[{ 0 }]])
]]>
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
index 12c0f5aa530..3136cefbcad 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
@@ -18,36 +18,205 @@
package org.apache.flink.table.runtime.generated;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.runtime.operators.process.PassAllCollector;
+import org.apache.flink.table.runtime.operators.process.PassThroughCollectorBase;
import org.apache.flink.table.runtime.operators.process.ProcessTableOperator;
-import org.apache.flink.util.Collector;
+import org.apache.flink.table.runtime.operators.process.ProcessTableOperator.RunnerContext;
+import org.apache.flink.table.runtime.operators.process.ProcessTableOperator.RunnerOnTimerContext;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.function.RunnableWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
/**
* Abstraction of code-generated calls to {@link ProcessTableFunction} to be used within {@link
* ProcessTableOperator}.
*/
+@Internal
public abstract class ProcessTableRunner extends AbstractRichFunction {
- public Collector<RowData> runnerCollector;
- public ProcessTableFunction.Context runnerContext;
+ // Constant references after initialization
+ private ValueState<RowData>[] stateHandles;
+ private HashFunction[] stateHashCode;
+ private RecordEqualiser[] stateEquals;
+ private boolean emitRowtime;
+
+ // Contexts
+ protected RunnerContext runnerContext;
+ protected RunnerOnTimerContext runnerOnTimerContext;
+
+ // Collectors
+ protected PassThroughCollectorBase evalCollector;
+ protected PassAllCollector onTimerCollector;
+
+ // Current input table
+ protected int inputIndex = -1;
+ protected RowData inputRow;
+
+ // Current time
+ private long currentWatermark = Long.MIN_VALUE;
+ private @Nullable Long rowtime;
+ private @Nullable StringData timerName;
+
+ /** State entries to be converted into external data structure; null if state is empty. */
+ protected RowData[] stateToFunction;
/**
- * @param stateToFunction state entries to be converted into external data structure; null if
- * state is empty
- * @param stateCleared reference to whether the state has been cleared within the function; if
- * yes, a conversion from external to internal data structure is not necessary anymore
- * @param stateFromFunction state ready for persistence; null if {@param stateCleared} was true
- * during conversion
- * @param inputIndex index of the currently processed input table
- * @param row data of the currently processed input table
+ * Reference to whether the state has been cleared within the function; if yes, a conversion
+ * from external to internal data structure is not necessary anymore.
*/
- public abstract void processElement(
- RowData[] stateToFunction,
- boolean[] stateCleared,
- RowData[] stateFromFunction,
- int inputIndex,
- RowData row)
- throws Exception;
+ protected boolean[] stateCleared;
+
+ /** State ready for persistence; null if {@link #stateCleared} was true during conversion. */
+ protected RowData[] stateFromFunction;
+
+ public void initialize(
+ ValueState<RowData>[] stateHandles,
+ HashFunction[] stateHashCode,
+ RecordEqualiser[] stateEquals,
+ boolean emitRowtime,
+ RunnerContext runnerContext,
+ RunnerOnTimerContext runnerOnTimerContext,
+ PassThroughCollectorBase evalCollector,
+ PassAllCollector onTimerCollector) {
+ this.stateHandles = stateHandles;
+ this.stateHashCode = stateHashCode;
+ this.stateEquals = stateEquals;
+ this.emitRowtime = emitRowtime;
+
+ // Accessed by generated code
+ this.runnerContext = runnerContext;
+ this.runnerOnTimerContext = runnerOnTimerContext;
+ this.evalCollector = evalCollector;
+ this.onTimerCollector = onTimerCollector;
+ this.stateToFunction = new RowData[stateHandles.length];
+ this.stateCleared = new boolean[stateHandles.length];
+ this.stateFromFunction = new RowData[stateHandles.length];
+ }
+
+ public void ingestTableEvent(int pos, RowData row, int timeColumn) {
+ evalCollector.setPrefix(row);
+ if (timeColumn == -1) {
+ rowtime = null;
+ } else {
+ final long inputTime = row.getTimestamp(timeColumn, 3).getMillisecond();
+ if (emitRowtime) {
+ evalCollector.setRowtime(inputTime);
+ }
+ rowtime = inputTime;
+ }
+ inputIndex = pos;
+ inputRow = row;
+ }
+
+ public void ingestTimerEvent(RowData key, @Nullable StringData name, long timerTime) {
+ onTimerCollector.setPrefix(key);
+ if (emitRowtime) {
+ onTimerCollector.setRowtime(timerTime);
+ }
+ rowtime = timerTime;
+ timerName = name;
+ }
+
+ public void ingestWatermarkEvent(long watermarkTime) {
+ currentWatermark = watermarkTime;
+ }
+
+ public void clearAllState() {
+ Arrays.fill(stateCleared, true);
+ }
+
+ public void clearState(int statePos) {
+ stateCleared[statePos] = true;
+ }
+
+ public long getCurrentWatermark() {
+ return currentWatermark;
+ }
+
+ public @Nullable Long getTime() {
+ return rowtime;
+ }
+
+ public @Nullable StringData getTimerName() {
+ return timerName;
+ }
+
+ public void processEval() throws Exception {
+ // Drop late events
+ if (rowtime != null && rowtime <= currentWatermark) {
+ return;
+ }
+ processMethod(this::callEval);
+ }
+
+ public void processOnTimer() throws Exception {
+ processMethod(this::callOnTimer);
+ }
+
+ public abstract void callEval() throws Exception;
+
+ public abstract void callOnTimer() throws Exception;
+
+ private void processMethod(RunnableWithException method) throws Exception {
+ if (stateHandles.length > 0) {
+ // For each function call:
+ // - the state is read from Flink
+ // - converted into external data structure
+ // - evaluated
+ // - converted into internal data structure (if not cleared)
+ // - the state is written into Flink
+ moveStateToFunction();
+ method.run();
+ moveStateFromFunction();
+ } else {
+ method.run();
+ }
+ }
+
+ private void moveStateToFunction() throws IOException {
+ Arrays.fill(stateCleared, false);
+ for (int i = 0; i < stateHandles.length; i++) {
+ final RowData value = stateHandles[i].value();
+ stateToFunction[i] = value;
+ }
+ }
+
+ private void moveStateFromFunction() throws IOException {
+ for (int i = 0; i < stateHandles.length; i++) {
+ final RowData fromFunction = stateFromFunction[i];
+ if (fromFunction == null || isEmpty(fromFunction)) {
+ // Reduce state size
+ stateHandles[i].clear();
+ } else {
+ final HashFunction hashCode = stateHashCode[i];
+ final RecordEqualiser equals = stateEquals[i];
+ final RowData toFunction = stateToFunction[i];
+ // Reduce state updates by checking if something has changed
+ if (toFunction == null
+ || hashCode.hashCode(toFunction) != hashCode.hashCode(fromFunction)
+ || !equals.equals(toFunction, fromFunction)) {
+ stateHandles[i].update(fromFunction);
+ }
+ }
+ }
+ }
+
+ private static boolean isEmpty(RowData row) {
+ for (int i = 0; i < row.getArity(); i++) {
+ if (!row.isNullAt(i)) {
+ return false;
+ }
+ }
+ return row.getRowKind() == RowKind.INSERT;
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ExternalTimeContext.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ExternalTimeContext.java
new file mode 100644
index 00000000000..96ec7cbec23
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ExternalTimeContext.java
@@ -0,0 +1,91 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * User-facing time context that converts to/from the desired time class.
+ *
+ * @param <TimeConversion> a conversion supported by {@link TimeConverter}
+ */
+@Internal
+class ExternalTimeContext<TimeConversion>
+ implements ProcessTableFunction.TimeContext<TimeConversion> {
+
+ private final ReadableInternalTimeContext internalTimeContext;
+ private final TimeConverter<TimeConversion> timeConverter;
+
+ ExternalTimeContext(
+ ReadableInternalTimeContext internalTimeContext,
+ TimeConverter<TimeConversion> timeConverter) {
+ this.internalTimeContext = internalTimeContext;
+ this.timeConverter = timeConverter;
+ }
+
+ @Override
+ public TimeConversion time() {
+ final Long internal = internalTimeContext.time();
+ if (internal == null) {
+ return null;
+ }
+ return timeConverter.toExternal(internal);
+ }
+
+ @Override
+ public TimeConversion currentWatermark() {
+ final Long internal = internalTimeContext.currentWatermark();
+ if (internal == null) {
+ return null;
+ }
+ return timeConverter.toExternal(internal);
+ }
+
+ @Override
+ public void registerOnTime(String name, TimeConversion time) {
+ Preconditions.checkNotNull(name, "Name must not be null.");
+ Preconditions.checkNotNull(time, "Time must not be null.");
+ internalTimeContext.registerOnTime(name, timeConverter.toInternal(time));
+ }
+
+ @Override
+ public void registerOnTime(TimeConversion time) {
+ Preconditions.checkNotNull(time, "Time must not be null.");
+ internalTimeContext.registerOnTime(timeConverter.toInternal(time));
+ }
+
+ @Override
+ public void clearTimer(String name) {
+ Preconditions.checkNotNull(name, "Name must not be null.");
+ internalTimeContext.clearTimer(name);
+ }
+
+ @Override
+ public void clearTimer(TimeConversion time) {
+ Preconditions.checkNotNull(time, "Time must not be null.");
+ internalTimeContext.clearTimer(timeConverter.toInternal(time));
+ }
+
+ @Override
+ public void clearAllTimers() {
+ internalTimeContext.clearAllTimers();
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassAllCollector.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassAllCollector.java
index 64328fe2661..78aa5174264 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassAllCollector.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassAllCollector.java
@@ -18,30 +18,20 @@
package org.apache.flink.table.runtime.operators.process;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.utils.JoinedRowData;
-import org.apache.flink.table.types.inference.SystemTypeInference;
-/** Forwards all input columns in sync with the output strategy of {@link SystemTypeInference}. */
+/** Forwards all columns of the given row. */
+@Internal
public class PassAllCollector extends PassThroughCollectorBase {
- private final JoinedRowData joinedRowData;
- private RowData inputRow;
-
public PassAllCollector(Output<StreamRecord<RowData>> output) {
super(output);
- joinedRowData = new JoinedRowData();
- }
-
- void setInput(RowData input) {
- inputRow = input;
}
- @Override
- public void collect(RowData functionOutput) {
- joinedRowData.replace(inputRow, functionOutput);
- super.collect(joinedRowData);
+ public void setPrefix(RowData input) {
+ prefix = input;
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassPartitionKeysCollector.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassPartitionKeysCollector.java
index 776dc9e0056..a9df92ea15c 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassPartitionKeysCollector.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassPartitionKeysCollector.java
@@ -18,34 +18,22 @@
package org.apache.flink.table.runtime.operators.process;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.utils.JoinedRowData;
import org.apache.flink.table.data.utils.ProjectedRowData;
-import org.apache.flink.table.types.inference.SystemTypeInference;
-/**
- * Forwards input partition keys in sync with the output strategy of {@link SystemTypeInference}.
- */
+/** Forwards partition keys of the given row. */
+@Internal
public class PassPartitionKeysCollector extends PassThroughCollectorBase {
- private final ProjectedRowData projectedInput;
- private final JoinedRowData joinedRowData;
-
public PassPartitionKeysCollector(Output<StreamRecord<RowData>> output, int[] partitionKeys) {
super(output);
- projectedInput = ProjectedRowData.from(partitionKeys);
- joinedRowData = new JoinedRowData();
- }
-
- void setInput(RowData input) {
- projectedInput.replaceRow(input);
+ prefix = ProjectedRowData.from(partitionKeys);
}
- @Override
- public void collect(RowData functionOutput) {
- joinedRowData.replace(projectedInput, functionOutput);
- super.collect(joinedRowData);
+ public void setPrefix(RowData input) {
+ ((ProjectedRowData) prefix).replaceRow(input);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
index db39f73a042..c1fdae779e3 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/PassThroughCollectorBase.java
@@ -18,17 +18,44 @@
package org.apache.flink.table.runtime.operators.process;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.data.utils.JoinedRowData;
import org.apache.flink.table.runtime.util.StreamRecordCollector;
/** Base class for collectors that pass input columns. */
+@Internal
public abstract class PassThroughCollectorBase extends StreamRecordCollector<RowData> {
+ private final JoinedRowData withPrefix;
+ private final JoinedRowData withRowtime;
+
+ private RowData rowtime;
+ protected RowData prefix;
+
public PassThroughCollectorBase(Output<StreamRecord<RowData>> output) {
super(output);
+ // constructs a flattened row of [[prefix | function output] | rowtime]
+ withPrefix = new JoinedRowData();
+ withRowtime = new JoinedRowData();
+ prefix = GenericRowData.of();
+ rowtime = GenericRowData.of();
+ }
+
+ public abstract void setPrefix(RowData input);
+
+ public void setRowtime(Long time) {
+ rowtime = GenericRowData.of(TimestampData.fromEpochMillis(time));
}
- abstract void setInput(RowData input);
+ @Override
+ public void collect(RowData functionOutput) {
+ withPrefix.replace(prefix, functionOutput);
+ withRowtime.replace(withPrefix, rowtime);
+ super.collect(withRowtime);
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperator.java
index 9a952602dda..30ba25d82aa 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessTableOperator.java
@@ -23,29 +23,44 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.DefaultOpenContext;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.state.KeyedStateStore;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
import org.apache.flink.table.functions.TableSemantics;
import org.apache.flink.table.runtime.generated.HashFunction;
import org.apache.flink.table.runtime.generated.ProcessTableRunner;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.operators.process.TimeConverter.InstantTimeConverter;
+import org.apache.flink.table.runtime.operators.process.TimeConverter.LocalDateTimeConverter;
+import org.apache.flink.table.runtime.operators.process.TimeConverter.LongTimeConverter;
import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.runtime.typeutils.StringDataSerializer;
import org.apache.flink.table.runtime.util.StateConfigUtil;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.types.RowKind;
import javax.annotation.Nullable;
-import java.io.IOException;
-import java.util.Arrays;
+import java.time.Instant;
+import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -53,7 +68,7 @@ import java.util.Optional;
/** Operator for {@link ProcessTableFunction}. */
public class ProcessTableOperator extends AbstractStreamOperator<RowData>
- implements OneInputStreamOperator<RowData, RowData> {
+ implements OneInputStreamOperator<RowData, RowData>, Triggerable<RowData, Object> {
private final @Nullable RuntimeTableSemantics tableSemantics;
private final List<RuntimeStateInfo> stateInfos;
@@ -61,12 +76,15 @@ public class ProcessTableOperator extends AbstractStreamOperator<RowData>
private final HashFunction[] stateHashCode;
private final RecordEqualiser[] stateEquals;
- private transient PassThroughCollectorBase collector;
+ private transient ReadableInternalTimeContext internalTimeContext;
+ private transient PassThroughCollectorBase evalCollector;
+ private transient PassAllCollector onTimerCollector;
private transient ValueStateDescriptor<RowData>[] stateDescriptors;
private transient ValueState<RowData>[] stateHandles;
- private transient RowData[] stateToFunction;
- private transient boolean[] stateCleared;
- private transient RowData[] stateFromFunction;
+
+ private transient @Nullable MapState<StringData, Long> namedTimersMapState;
+ private transient @Nullable InternalTimerService<StringData> namedTimerService;
+ private transient @Nullable InternalTimerService<VoidNamespace> unnamedTimerService;
public ProcessTableOperator(
StreamOperatorParameters<RowData> parameters,
@@ -84,117 +102,108 @@ public class ProcessTableOperator extends AbstractStreamOperator<RowData>
}
@Override
- @SuppressWarnings({"rawtypes", "unchecked"})
public void open() throws Exception {
super.open();
- // Pass through columns
- if (tableSemantics == null || tableSemantics.passColumnsThrough()) {
- collector = new PassAllCollector(output);
- } else {
- collector = new PassPartitionKeysCollector(output, tableSemantics.partitionByColumns());
- }
- processTableRunner.runnerCollector = this.collector;
+ final RunnerContext runnerContext = new RunnerContext();
+ final RunnerOnTimerContext runnerOnTimerContext = new RunnerOnTimerContext();
- // State
- final KeyedStateStore keyedStateStore = getKeyedStateStore();
- stateDescriptors = new ValueStateDescriptor[stateInfos.size()];
- stateHandles = new ValueState[stateInfos.size()];
- stateToFunction = new RowData[stateInfos.size()];
- stateCleared = new boolean[stateInfos.size()];
- stateFromFunction = new RowData[stateInfos.size()];
- for (int i = 0; i < stateInfos.size(); i++) {
- final RuntimeStateInfo stateInfo = stateInfos.get(i);
- final LogicalType type = stateInfo.getType();
- final ValueStateDescriptor<RowData> descriptor =
- new ValueStateDescriptor(
- stateInfo.getStateName(), InternalSerializers.create(type));
- final StateTtlConfig ttlConfig =
- StateConfigUtil.createTtlConfig(stateInfo.getTimeToLive());
- if (ttlConfig.isEnabled()) {
- descriptor.enableTimeToLive(ttlConfig);
- }
- stateDescriptors[i] = descriptor;
- stateHandles[i] = keyedStateStore.getState(descriptor);
- }
+ setTimerServices();
+ setTimeContext();
+ setCollectors();
+ setStateDescriptors();
+ setStateHandles();
- // Context
- processTableRunner.runnerContext = new ProcessFunctionContext();
+ processTableRunner.initialize(
+ stateHandles,
+ stateHashCode,
+ stateEquals,
+ shouldEmitRowtime(),
+ runnerContext,
+ runnerOnTimerContext,
+ evalCollector,
+ onTimerCollector);
- // Prepare runner
+ // Open runner
FunctionUtils.setFunctionRuntimeContext(processTableRunner, getRuntimeContext());
FunctionUtils.openFunction(processTableRunner, DefaultOpenContext.INSTANCE);
}
@Override
public void processElement(StreamRecord<RowData> element) throws Exception {
- collector.setInput(element.getValue());
-
- // For each function call:
- // - the state is read from Flink
- // - converted into external data structure
- // - evaluated
- // - converted into internal data structure (if not cleared)
- // - the state is written into Flink
-
- moveStateToFunction();
- processTableRunner.processElement(
- stateToFunction, stateCleared, stateFromFunction, 0, element.getValue());
- moveStateFromFunction();
- }
-
- private void moveStateToFunction() throws IOException {
- Arrays.fill(stateCleared, false);
- for (int i = 0; i < stateHandles.length; i++) {
- final RowData value = stateHandles[i].value();
- stateToFunction[i] = value;
+ // Set table argument
+ if (tableSemantics != null) {
+ processTableRunner.ingestTableEvent(0, element.getValue(), tableSemantics.timeColumn());
}
+ processTableRunner.processEval();
}
- private void moveStateFromFunction() throws IOException {
- for (int i = 0; i < stateHandles.length; i++) {
- final RowData fromFunction = stateFromFunction[i];
- if (fromFunction == null || isEmpty(fromFunction)) {
- // Reduce state size
- stateHandles[i].clear();
- } else {
- final HashFunction hashCode = this.stateHashCode[i];
- final RecordEqualiser equals = this.stateEquals[i];
- final RowData toFunction = stateToFunction[i];
- // Reduce state updates by checking if something has changed
- if (toFunction == null
- || hashCode.hashCode(toFunction) != hashCode.hashCode(fromFunction)
- || !equals.equals(toFunction, fromFunction)) {
- stateHandles[i].update(fromFunction);
- }
- }
- }
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ processTableRunner.ingestWatermarkEvent(mark.getTimestamp());
}
- private static boolean isEmpty(RowData row) {
- for (int i = 0; i < row.getArity(); i++) {
- if (!row.isNullAt(i)) {
- return false;
- }
- }
- return row.getRowKind() == RowKind.INSERT;
+ @Override
+ public void onEventTime(InternalTimer<RowData, Object> timer) throws Exception {
+ final Object namedTimer = timer.getNamespace();
+ processTableRunner.ingestTimerEvent(
+ timer.getKey(),
+ namedTimer == VoidNamespace.INSTANCE ? null : (StringData) namedTimer,
+ timer.getTimestamp());
+ processTableRunner.processOnTimer();
}
+ @Override
+ public void onProcessingTime(InternalTimer<RowData, Object> timer) throws Exception {}
+
+ /** Implementation of {@link ProcessTableFunction.Context}. */
@Internal
- public class ProcessFunctionContext implements ProcessTableFunction.Context {
+ public class RunnerContext implements ProcessTableFunction.Context {
private final Map<String, RuntimeTableSemantics> tableSemanticsMap;
private final Map<String, Integer> stateNameToPosMap;
- ProcessFunctionContext() {
- this.tableSemanticsMap =
- Optional.ofNullable(tableSemantics)
- .map(s -> Map.of(tableSemantics.getArgName(), tableSemantics))
- .orElse(Map.of());
- this.stateNameToPosMap = new HashMap<>();
+ RunnerContext() {
+ this.tableSemanticsMap = createTableSemanticsMap();
+ this.stateNameToPosMap = createStateNameToPosMap();
+ }
+
+ private Map<String, RuntimeTableSemantics> createTableSemanticsMap() {
+ return Optional.ofNullable(tableSemantics)
+ .map(s -> Map.of(tableSemantics.getArgName(), tableSemantics))
+ .orElse(Map.of());
+ }
+
+ private Map<String, Integer> createStateNameToPosMap() {
+ final Map<String, Integer> stateNameToPosMap = new HashMap<>();
for (int i = 0; i < stateInfos.size(); i++) {
- this.stateNameToPosMap.put(stateInfos.get(i).getStateName(), i);
+ stateNameToPosMap.put(stateInfos.get(i).getStateName(), i);
+ }
+ return stateNameToPosMap;
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked"})
+ public <TimeType> TimeContext<TimeType> timeContext(Class<TimeType> conversionClass) {
+ final TimeConverter<?> timeConverter;
+ if (conversionClass == Instant.class) {
+ timeConverter = InstantTimeConverter.INSTANCE;
+ } else if (conversionClass == LocalDateTime.class) {
+ timeConverter = LocalDateTimeConverter.INSTANCE;
+ } else if (conversionClass == Long.class) {
+ timeConverter = LongTimeConverter.INSTANCE;
+ } else {
+ throw new TableRuntimeException(
+ "Unsupported conversion class for TimeContext: "
+ + conversionClass.getName());
}
+
+ internalTimeContext.setTime(
+ processTableRunner.getCurrentWatermark(), processTableRunner.getTime());
+
+ return (TimeContext<TimeType>)
+ new ExternalTimeContext<>(internalTimeContext, timeConverter);
}
@Override
@@ -212,12 +221,23 @@ public class ProcessTableOperator extends AbstractStreamOperator<RowData>
if (statePos == null) {
throw new TableRuntimeException("Unknown state entry: " + stateName);
}
- stateCleared[statePos] = true;
+ processTableRunner.clearState(statePos);
}
@Override
public void clearAllState() {
- Arrays.fill(stateCleared, true);
+ processTableRunner.clearAllState();
+ }
+
+ @Override
+ public void clearAllTimers() {
+ internalTimeContext.clearAllTimers();
+ }
+
+ @Override
+ public void clearAll() {
+ clearAllState();
+ clearAllTimers();
}
@VisibleForTesting
@@ -229,4 +249,106 @@ public class ProcessTableOperator extends AbstractStreamOperator<RowData>
return stateDescriptors[statePos];
}
}
+
+ /** Implementation of {@link ProcessTableFunction.OnTimerContext}. */
+ @Internal
+ public class RunnerOnTimerContext extends RunnerContext
+ implements ProcessTableFunction.OnTimerContext {
+
+ @Override
+ public @Nullable String currentTimer() {
+ final StringData timerName = processTableRunner.getTimerName();
+ return timerName == null ? null : timerName.toString();
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Instances from table semantics
+ // --------------------------------------------------------------------------------------------
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private void setTimerServices() {
+ if (shouldEnableTimers()) {
+ final KeyedStateStore keyedStateStore = getKeyedStateStore();
+ final MapStateDescriptor<StringData, Long> namedTimersDescriptor =
+ new MapStateDescriptor<>(
+ "internal-named-timers-map",
+ StringDataSerializer.INSTANCE,
+ LongSerializer.INSTANCE);
+ namedTimersMapState = keyedStateStore.getMapState(namedTimersDescriptor);
+ namedTimerService =
+ getInternalTimerService(
+ "user-named-timers", StringDataSerializer.INSTANCE, (Triggerable) this);
+ unnamedTimerService =
+ getInternalTimerService(
+ "user-unnamed-timers",
+ VoidNamespaceSerializer.INSTANCE,
+ (Triggerable) this);
+ } else {
+ namedTimersMapState = null;
+ namedTimerService = null;
+ unnamedTimerService = null;
+ }
+ }
+
+ private void setTimeContext() {
+ if (shouldEnableTimers()) {
+ internalTimeContext =
+ new WritableInternalTimeContext(
+ namedTimersMapState, namedTimerService, unnamedTimerService);
+ } else {
+ internalTimeContext = new ReadableInternalTimeContext();
+ }
+ }
+
+ private void setCollectors() {
+ if (tableSemantics == null || tableSemantics.passColumnsThrough()) {
+ evalCollector = new PassAllCollector(output);
+ } else {
+ evalCollector =
+ new PassPartitionKeysCollector(output, tableSemantics.partitionByColumns());
+ }
+ onTimerCollector = new PassAllCollector(output);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void setStateDescriptors() {
+ final ValueStateDescriptor<RowData>[] stateDescriptors =
+ new ValueStateDescriptor[stateInfos.size()];
+ for (int i = 0; i < stateInfos.size(); i++) {
+ final RuntimeStateInfo stateInfo = stateInfos.get(i);
+ final LogicalType type = stateInfo.getType();
+ final ValueStateDescriptor<RowData> stateDescriptor =
+ new ValueStateDescriptor<>(
+ stateInfo.getStateName(), InternalSerializers.create(type));
+ final StateTtlConfig ttlConfig =
+ StateConfigUtil.createTtlConfig(stateInfo.getTimeToLive());
+ if (ttlConfig.isEnabled()) {
+ stateDescriptor.enableTimeToLive(ttlConfig);
+ }
+ stateDescriptors[i] = stateDescriptor;
+ }
+ this.stateDescriptors = stateDescriptors;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void setStateHandles() {
+ final KeyedStateStore keyedStateStore = getKeyedStateStore();
+ final ValueState<RowData>[] stateHandles = new ValueState[stateDescriptors.length];
+ for (int i = 0; i < stateInfos.size(); i++) {
+ stateHandles[i] = keyedStateStore.getState(stateDescriptors[i]);
+ }
+ this.stateHandles = stateHandles;
+ }
+
+ private boolean shouldEmitRowtime() {
+ return tableSemantics != null && tableSemantics.timeColumn() != -1;
+ }
+
+ private boolean shouldEnableTimers() {
+ return tableSemantics != null
+ && tableSemantics.hasSetSemantics()
+ && !tableSemantics.passColumnsThrough()
+ && tableSemantics.getChangelogMode().containsOnly(RowKind.INSERT);
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ReadableInternalTimeContext.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ReadableInternalTimeContext.java
new file mode 100644
index 00000000000..bedba764ffd
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ReadableInternalTimeContext.java
@@ -0,0 +1,86 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.types.inference.StaticArgumentTrait;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
+
+/**
+ * Internal {@link TimeContext} for PTFs with row semantics, PTFs without table arguments, or traits
+ * {@link StaticArgumentTrait#PASS_COLUMNS_THROUGH} and {@link StaticArgumentTrait#SUPPORT_UPDATES}.
+ */
+@Internal
+class ReadableInternalTimeContext implements ProcessTableFunction.TimeContext<Long> {
+
+ protected long currentWatermark;
+ protected @Nullable Long time;
+
+ void setTime(long currentWatermark, @Nullable Long time) {
+ this.currentWatermark = currentWatermark;
+ this.time = time;
+ }
+
+ @Override
+ public Long time() {
+ if (time == null) {
+ return null;
+ }
+ return time;
+ }
+
+ @Override
+ public Long currentWatermark() {
+ if (currentWatermark == Long.MIN_VALUE) {
+ return null;
+ }
+ return currentWatermark;
+ }
+
+ @Override
+ public void registerOnTime(String name, Long time) {
+ throw readOnlyException();
+ }
+
+ @Override
+ public void registerOnTime(Long time) {
+ throw readOnlyException();
+ }
+
+ @Override
+ public void clearTimer(String name) {}
+
+ @Override
+ public void clearTimer(Long time) {}
+
+ @Override
+ public void clearAllTimers() {}
+
+ private static TableRuntimeException readOnlyException() {
+ return new TableRuntimeException(
+ "Timers are not supported in the current PTF declaration. "
+ + "Note that only PTFs that take set semantic tables support timers. "
+ + "Also timers are not available for advanced traits such as supporting pass-through columns or updates.");
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
index c7d27aff3f0..30e1957bce0 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/RuntimeTableSemantics.java
@@ -43,6 +43,7 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
private final byte[] expectedChanges;
private final boolean passColumnsThrough;
private final boolean hasSetSemantics;
+ private final int timeColumn;
private transient ChangelogMode changelogMode;
@@ -53,7 +54,8 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
int[] partitionByColumns,
byte[] expectedChanges,
boolean passColumnsThrough,
- boolean hasSetSemantics) {
+ boolean hasSetSemantics,
+ int timeColumn) {
this.argName = argName;
this.inputIndex = inputIndex;
this.dataType = dataType;
@@ -61,6 +63,7 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
this.expectedChanges = expectedChanges;
this.passColumnsThrough = passColumnsThrough;
this.hasSetSemantics = hasSetSemantics;
+ this.timeColumn = timeColumn;
}
public String getArgName() {
@@ -79,6 +82,17 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
return hasSetSemantics;
}
+ public ChangelogMode getChangelogMode() {
+ if (changelogMode == null) {
+ final ChangelogMode.Builder builder = ChangelogMode.newBuilder();
+ for (byte expectedChange : expectedChanges) {
+ builder.addContainedKind(RowKind.fromByteValue(expectedChange));
+ }
+ changelogMode = builder.build();
+ }
+ return changelogMode;
+ }
+
@Override
public DataType dataType() {
return dataType;
@@ -96,7 +110,7 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
@Override
public int timeColumn() {
- return -1;
+ return timeColumn;
}
@Override
@@ -106,13 +120,6 @@ public class RuntimeTableSemantics implements TableSemantics, Serializable {
@Override
public Optional<ChangelogMode> changelogMode() {
- if (changelogMode == null) {
- final ChangelogMode.Builder builder = ChangelogMode.newBuilder();
- for (byte expectedChange : expectedChanges) {
- builder.addContainedKind(RowKind.fromByteValue(expectedChange));
- }
- changelogMode = builder.build();
- }
- return Optional.of(changelogMode);
+ return Optional.of(getChangelogMode());
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/TimeConverter.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/TimeConverter.java
new file mode 100644
index 00000000000..e45749e55c5
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/TimeConverter.java
@@ -0,0 +1,87 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.annotation.Internal;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+/** Time converter between internal and external time conversion classes. */
+@Internal
+abstract class TimeConverter<TimeConversion> {
+
+ abstract TimeConversion toExternal(long timeInternal);
+
+ abstract long toInternal(TimeConversion timeExternal);
+
+ private TimeConverter() {}
+
+ static class InstantTimeConverter extends TimeConverter<Instant> {
+
+ static final TimeConverter<Instant> INSTANCE = new InstantTimeConverter();
+
+ private InstantTimeConverter() {}
+
+ @Override
+ Instant toExternal(long timeInternal) {
+ return Instant.ofEpochMilli(timeInternal);
+ }
+
+ @Override
+ long toInternal(Instant timeExternal) {
+ return timeExternal.toEpochMilli();
+ }
+ }
+
+ static class LocalDateTimeConverter extends TimeConverter<LocalDateTime> {
+
+ static final LocalDateTimeConverter INSTANCE = new LocalDateTimeConverter();
+
+ private LocalDateTimeConverter() {}
+
+ @Override
+ LocalDateTime toExternal(long timeInternal) {
+ return LocalDateTime.ofInstant(Instant.ofEpochMilli(timeInternal), ZoneOffset.UTC);
+ }
+
+ @Override
+ long toInternal(LocalDateTime timeExternal) {
+ return timeExternal.toInstant(ZoneOffset.UTC).toEpochMilli();
+ }
+ }
+
+ static class LongTimeConverter extends TimeConverter<Long> {
+
+ static final LongTimeConverter INSTANCE = new LongTimeConverter();
+
+ private LongTimeConverter() {}
+
+ @Override
+ Long toExternal(long timeInternal) {
+ return timeInternal;
+ }
+
+ @Override
+ long toInternal(Long timeExternal) {
+ return timeExternal;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
new file mode 100644
index 00000000000..5fd996fecaa
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java
@@ -0,0 +1,186 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.runtime.state.VoidNamespace;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Internal {@link TimeContext} for PTFs with set semantics. */
+@Internal
+class WritableInternalTimeContext extends ReadableInternalTimeContext {
+
+ private final MapState<StringData, Long> namedTimersMapState;
+ private final InternalTimerService<StringData> namedTimerService;
+ private final InternalTimerService<VoidNamespace> unnamedTimerService;
+
+ WritableInternalTimeContext(
+ MapState<StringData, Long> namedTimersMapState,
+ InternalTimerService<StringData> namedTimerService,
+ InternalTimerService<VoidNamespace> unnamedTimerService) {
+ this.namedTimersMapState = namedTimersMapState;
+ this.namedTimerService = namedTimerService;
+ this.unnamedTimerService = unnamedTimerService;
+ }
+
+ @Override
+ public Long currentWatermark() {
+ final long currentWatermark = unnamedTimerService.currentWatermark();
+ if (currentWatermark == Long.MIN_VALUE) {
+ return null;
+ }
+ return currentWatermark;
+ }
+
+ @Override
+ public void registerOnTime(String name, Long time) {
+ registerOnTimeInternal(name, time);
+ }
+
+ @Override
+ public void registerOnTime(Long time) {
+ registerOnTimeInternal(null, time);
+ }
+
+ @Override
+ public void clearTimer(String name) {
+ Preconditions.checkNotNull(name, "Timer name must not be null.");
+ deleteTimerInternal(name);
+ }
+
+ @Override
+ public void clearTimer(Long time) {
+ deleteTimerInternal(time);
+ }
+
+ @Override
+ public void clearAllTimers() {
+ clearUnnamedTimers();
+ clearNamedTimers();
+ }
+
+ private void registerOnTimeInternal(@Nullable String name, long newTime) {
+ if (newTime <= unnamedTimerService.currentWatermark()) {
+ // Do not register timers for late events.
+ // Otherwise, the next watermark would trigger an onTimer() that emits late events.
+ return;
+ }
+ if (name != null) {
+ replaceNamedTimer(StringData.fromString(name), newTime);
+ } else {
+ registerUnnamedTimer(newTime);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Unnamed timers
+ // --------------------------------------------------------------------------------------------
+
+ protected void deleteTimerInternal(long time) {
+ deleteUnnamedTimer(time);
+ }
+
+ private void registerUnnamedTimer(long time) {
+ unnamedTimerService.registerEventTimeTimer(VoidNamespace.INSTANCE, time);
+ }
+
+ private void deleteUnnamedTimer(long time) {
+ unnamedTimerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, time);
+ }
+
+ private void clearUnnamedTimers() {
+ final List<Long> unnamedTimers = new ArrayList<>();
+ try {
+ unnamedTimerService.forEachEventTimeTimer((namespace, time) -> unnamedTimers.add(time));
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ unnamedTimers.forEach(
+ time -> unnamedTimerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, time));
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Named timers
+ // --------------------------------------------------------------------------------------------
+
+ private void deleteTimerInternal(String name) {
+ deleteNamedTimer(StringData.fromString(name));
+ }
+
+ private void replaceNamedTimer(StringData internalName, long newTime) {
+ final Long existingTime;
+ try {
+ existingTime = namedTimersMapState.get(internalName);
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ if (existingTime != null) {
+ namedTimerService.deleteEventTimeTimer(internalName, existingTime);
+ }
+ registerNamedTimer(internalName, newTime);
+ }
+
+ private void registerNamedTimer(StringData internalName, long newTime) {
+ try {
+ namedTimersMapState.put(internalName, newTime);
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ namedTimerService.registerEventTimeTimer(internalName, newTime);
+ }
+
+ private void deleteNamedTimer(StringData internalName) {
+ final Long existingTime;
+ try {
+ existingTime = namedTimersMapState.get(internalName);
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ if (existingTime != null) {
+ namedTimerService.deleteEventTimeTimer(internalName, existingTime);
+ try {
+ namedTimersMapState.remove(internalName);
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ }
+ }
+
+ private void clearNamedTimers() {
+ final Iterable<Map.Entry<StringData, Long>> entries;
+ try {
+ entries = namedTimersMapState.entries();
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(e);
+ }
+ entries.forEach(e -> namedTimerService.deleteEventTimeTimer(e.getKey(), e.getValue()));
+ namedTimersMapState.clear();
+ }
+}