This is an automated email from the ASF dual-hosted git repository.
dianfu 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 dd8eeed0d02 [FLINK-40424][table][python] Support windowing TVFs in the
DataFrame API (#29070)
dd8eeed0d02 is described below
commit dd8eeed0d0216d79ee19218ef256e9869fa19d31
Author: Timo Theusner <[email protected]>
AuthorDate: Fri Sep 11 16:03:03 2026 +0200
[FLINK-40424][table][python] Support windowing TVFs in the DataFrame API
(#29070)
---
.../docs/reference/pyflink.dataframe/dataframe.rst | 13 +
flink-python/pyflink/dataframe/dataframe.py | 243 ++++++++++++++-
.../pyflink/dataframe/tests/test_dataframe.py | 293 +++++++++++++++++-
.../table/operations/QueryOperationVisitor.java | 2 +
.../WindowTableFunctionQueryOperation.java | 222 ++++++++++++++
.../operations/utils/OperationTreeBuilder.java | 70 +++++
.../utils/QueryOperationDefaultVisitor.java | 6 +
.../WindowTableFunctionQueryOperationTest.java | 193 ++++++++++++
.../test/program/FailingTableApiTestStep.java | 13 +
.../flink/table/test/program/TableApiTestStep.java | 19 ++
.../planner/plan/QueryOperationConverter.java | 74 +++++
.../table/api/QueryOperationSqlSemanticTest.java | 5 +
.../api/QueryOperationSqlSerializationTest.java | 5 +
.../table/api/QueryOperationTestPrograms.java | 333 +++++++++++++++++++++
.../plan/WindowTableFunctionRelNodeTest.java | 179 +++++++++++
.../plan/WindowTableFunctionRelNodeTest.xml | 101 +++++++
16 files changed, 1762 insertions(+), 9 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
index ca7cecc499f..c4f36fb8ae6 100644
--- a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -116,6 +116,19 @@ Results
DataFrame.to_table
DataFrame.to_pandas
+Windowing
+---------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataFrame.tumble
+ DataFrame.hop
+ DataFrame.cumulate
+ DataFrame.session
+
Expressions
-----------
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index ed2cf3dab1c..426e3a5640d 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -16,6 +16,7 @@
# limitations under the License.
################################################################################
+import datetime
from typing import (
TYPE_CHECKING,
Any,
@@ -609,7 +610,7 @@ class DataFrame:
if partition_by is None or (isinstance(partition_by, (list, tuple))
and not partition_by):
partition_keys = []
else:
- partition_keys = _normalize_subset(partition_by)
+ partition_keys = _normalize_subset(partition_by, "partition_by")
descending_flags = _normalize_descending(descending, len(order_keys))
nulls = _normalize_nulls_first(nulls_first, len(order_keys))
@@ -695,6 +696,160 @@ class DataFrame:
order_expressions.append(expression.desc if is_descending else
expression.asc)
return DataFrame(self._table.order_by(*order_expressions))
+ # ======================== Windowing ========================
+
+ @PublicEvolving()
+ def tumble(
+ self,
+ *,
+ on: Union[str, Expression],
+ size: Union["datetime.timedelta", Expression],
+ ) -> "DataFrame":
+ """
+ Assign rows to fixed-size, non-overlapping (tumbling) windows.
+
+ Appends ``window_start``, ``window_end`` and ``window_time`` and
returns an ordinary
+ DataFrame.
+
+ :param on: An existing event-time or processing-time column.
+ :param size: Window length.
+ :return: A new DataFrame with the window columns appended.
+ :raises TypeError: If ``on`` or ``size`` has an unsupported type.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> from datetime import timedelta
+ >>> windowed = df.tumble(on="event_time",
size=timedelta(minutes=10))
+
+ .. versionadded:: 2.4.0
+ """
+ time_col = _resolve_window_time_column(on)
+ return _window_dataframe(
+ self._table, "TUMBLE", time_col, _to_interval_expression(size)
+ )
+
+ @PublicEvolving()
+ def hop(
+ self,
+ *,
+ on: Union[str, Expression],
+ slide: Union["datetime.timedelta", Expression],
+ size: Union["datetime.timedelta", Expression],
+ ) -> "DataFrame":
+ """
+ Assign rows to overlapping fixed-size (hopping/sliding) windows of
length ``size`` starting
+ every ``slide``. Appends
``window_start``/``window_end``/``window_time``.
+
+ :param on: An existing event-time or processing-time column.
+ :param slide: Interval between successive window starts.
+ :param size: Window length.
+ :return: A new DataFrame with the window columns appended.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> from datetime import timedelta
+ >>> windowed = df.hop(
+ ... on="event_time",
+ ... slide=timedelta(minutes=5),
+ ... size=timedelta(minutes=10),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ time_col = _resolve_window_time_column(on)
+ return _window_dataframe(
+ self._table,
+ "HOP",
+ time_col,
+ _to_interval_expression(slide),
+ _to_interval_expression(size),
+ )
+
+ @PublicEvolving()
+ def cumulate(
+ self,
+ *,
+ on: Union[str, Expression],
+ step: Union["datetime.timedelta", Expression],
+ size: Union["datetime.timedelta", Expression],
+ ) -> "DataFrame":
+ """
+ Assign rows to cumulating windows that share a start and grow by
``step`` up to ``size``.
+ Appends ``window_start``/``window_end``/``window_time``.
+
+ :param on: An existing event-time or processing-time column.
+ :param step: Interval by which each window grows.
+ :param size: Maximum window length (a whole multiple of ``step``).
+ :return: A new DataFrame with the window columns appended.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> from datetime import timedelta
+ >>> windowed = df.cumulate(
+ ... on="event_time",
+ ... step=timedelta(minutes=5),
+ ... size=timedelta(minutes=10),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ time_col = _resolve_window_time_column(on)
+ return _window_dataframe(
+ self._table,
+ "CUMULATE",
+ time_col,
+ _to_interval_expression(step),
+ _to_interval_expression(size),
+ )
+
+ @PublicEvolving()
+ def session(
+ self,
+ *,
+ on: Union[str, Expression],
+ gap: Union["datetime.timedelta", Expression],
+ partition_by: Optional[
+ Union[str, Expression, List[Union[str, Expression]]]
+ ] = None,
+ ) -> "DataFrame":
+ """
+ Assign rows to activity-based (session) windows that close after
``gap`` of inactivity.
+ Appends ``window_start``/``window_end``/``window_time``.
+ When ``partition_by`` is given, sessions are computed independently
per key, so a gap in
+ one key's activity does not close another key's session.
+
+ :param on: An existing event-time or processing-time column.
+ :param gap: Inactivity gap that closes a session.
+ :param partition_by: Optional column name or expression or list of
column names or
+ expressions to compute per-key sessions. ``None`` means a global
session.
+ :return: A new DataFrame with the window columns appended.
+ :raises TypeError: If ``partition_by`` contains an unsupported element
type.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> from datetime import timedelta
+ >>> windowed = df.session(
+ ... on="event_time",
+ ... gap=timedelta(minutes=10),
+ ... partition_by="id",
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ time_col = _resolve_window_time_column(on)
+ partition_cols = _resolve_partition_columns(partition_by)
+ return _window_dataframe(
+ self._table,
+ "SESSION",
+ time_col,
+ _to_interval_expression(gap),
+ partition_cols=partition_cols,
+ )
+
# ======================== Slicing ========================
@PublicEvolving()
@@ -1148,19 +1303,23 @@ class GroupedDataFrame:
# ======================== Internal Helpers ========================
-def _normalize_subset(subset: Union[str, List[str], None]) ->
Optional[List[str]]:
+def _normalize_subset(
+ subset: Union[str, List[str], None], parameter_name: str = "subset"
+) -> Optional[List[str]]:
if subset is None:
return None
if isinstance(subset, str):
return [subset]
if isinstance(subset, (list, tuple)):
if not subset:
- raise ValueError("subset must not be empty")
+ raise ValueError("%s must not be empty" % parameter_name)
for name in subset:
if not isinstance(name, str):
- raise TypeError("subset must be a string or a list of strings")
+ raise TypeError(
+ "%s must be a string or a list of strings" % parameter_name
+ )
return list(subset)
- raise TypeError("subset must be a string or a list of strings")
+ raise TypeError("%s must be a string or a list of strings" %
parameter_name)
def _normalize_order_by(
@@ -1324,6 +1483,80 @@ def _quote_identifier(name: str) -> str:
return "`" + name.replace("`", "``") + "`"
+def _resolve_window_time_column(on: Union[str, Expression]) -> Expression:
+ if isinstance(on, str):
+ return table_col(on)
+ if isinstance(on, Expression):
+ return on
+ raise TypeError("on must be a column name or expression")
+
+
+def _resolve_partition_columns(
+ partition_by: Optional[Union[str, Expression, List[Union[str,
Expression]]]]
+) -> List[Expression]:
+ if partition_by is None:
+ return []
+ if isinstance(partition_by, (str, Expression)):
+ candidates: List[Union[str, Expression]] = [partition_by]
+ else:
+ candidates = partition_by
+ columns: List[Expression] = []
+ for candidate in candidates:
+ if isinstance(candidate, str):
+ columns.append(table_col(candidate))
+ elif isinstance(candidate, Expression):
+ columns.append(candidate)
+ else:
+ raise TypeError(
+ "partition_by must be a column name, expression, or a list of
them"
+ )
+ return columns
+
+
+def _window_dataframe(
+ table: Table,
+ window_kind: str,
+ time_col: Expression,
+ *interval_exprs: Any,
+ partition_cols: List[Expression] = [],
+) -> "DataFrame":
+ jvm = get_gateway().jvm
+ kind = getattr(
+
jvm.org.apache.flink.table.operations.WindowTableFunctionQueryOperation.WindowKind,
+ window_kind,
+ )
+ intervals = jvm.java.util.ArrayList()
+ for interval in interval_exprs:
+ intervals.add(interval)
+ partition_list = jvm.java.util.ArrayList()
+ for partition_col in partition_cols:
+ partition_list.add(partition_col._j_expr)
+ operation_tree_builder = table._t_env._j_tenv.getOperationTreeBuilder()
+ window_op = operation_tree_builder.windowTableFunction(
+ kind,
+ time_col._j_expr,
+ intervals,
+ partition_list,
+ table._j_table.getQueryOperation(),
+ )
+ j_table = table._t_env._j_tenv.createTable(window_op)
+ return DataFrame(Table(j_table, table._t_env))
+
+
+def _to_interval_expression(value: Union["datetime.timedelta", Expression]) ->
Any:
+ if isinstance(value, datetime.timedelta):
+ millis = value // datetime.timedelta(milliseconds=1)
+ return (
+ get_gateway()
+
.jvm.org.apache.flink.table.expressions.ApiExpressionUtils.intervalOfMillis(
+ millis
+ )
+ )
+ if isinstance(value, Expression):
+ return value._j_expr
+ raise TypeError("interval must be a datetime.timedelta or Expression")
+
+
def _normalize_aggregations(
aggs: Tuple[Expression, ...], named_aggs: Dict[str, Expression]
) -> List[Expression]:
diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py
b/flink-python/pyflink/dataframe/tests/test_dataframe.py
index d51d82c857d..f104db9242f 100644
--- a/flink-python/pyflink/dataframe/tests/test_dataframe.py
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -16,19 +16,25 @@
# limitations under the License.
################################################################################
-import inspect
import array
import decimal
+import inspect
+import os
+import pandas as pd
+import pyarrow as pa
import unittest
-from py4j.protocol import Py4JJavaError
from datetime import date, datetime, time, timedelta, timezone
+from py4j.protocol import Py4JJavaError
from typing import NamedTuple
from unittest.mock import Mock, patch
-import pandas as pd
-import pyarrow as pa
import pyflink.dataframe as pf
from pyflink.common import Row
+from pyflink.dataframe.dataframe import (
+ _resolve_window_time_column,
+ _to_interval_expression,
+)
+from pyflink.dataframe.datatype import DataType
from pyflink.table import (
DataTypes as TableDataTypes,
EnvironmentSettings,
@@ -1567,6 +1573,147 @@ class DataFrameUniqueTests(PyFlinkDataFrameUTTestCase):
)
+class DataFrameWindowUnitTests(PyFlinkDataFrameUTTestCase):
+ _WINDOW_SCHEMA_COLUMN_NAMES = [
+ "id",
+ "amount",
+ "ts",
+ "window_start",
+ "window_end",
+ "window_time",
+ ]
+
+ def _window_schema_column_data_types(self):
+ source_types = list(
+ self.dataframe._table.get_resolved_schema().get_column_data_types()
+ )
+ time_column_type = source_types[2]
+ return source_types + [
+ TableDataTypes.TIMESTAMP(3).not_null(),
+ TableDataTypes.TIMESTAMP(3).not_null(),
+ time_column_type.not_null(),
+ ]
+
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [{"id": 1, "amount": 10, "ts": datetime(2020, 1, 1)}],
+ watermark=("ts", "ts - INTERVAL '1' SECOND"),
+ )
+
+ def test_timedelta_becomes_java_value_literal(self):
+ serialized = _to_interval_expression(
+ timedelta(minutes=10)
+ ).asSerializableString()
+
+ self.assertIn("INTERVAL", serialized)
+ self.assertIn("00:10:00", serialized)
+
+ def test_sub_millisecond_precision_is_truncated_not_rounded(self):
+ truncated = _to_interval_expression(
+ timedelta(milliseconds=1, microseconds=500)
+ ).asSerializableString()
+ whole = _to_interval_expression(
+ timedelta(milliseconds=1)
+ ).asSerializableString()
+
+ self.assertEqual(truncated, whole)
+
+ def test_sub_millisecond_timedelta_truncates_to_zero(self):
+ serialized = _to_interval_expression(
+ timedelta(microseconds=500)
+ ).asSerializableString()
+
+ zero = _to_interval_expression(timedelta(0)).asSerializableString()
+
+ self.assertEqual(serialized, zero)
+
+ def test_expression_is_unwrapped_to_j_expr(self):
+ expr = pf.lit(10).minutes
+
+ self.assertIs(_to_interval_expression(expr), expr._j_expr)
+
+ def test_rejects_unsupported_type(self):
+ with self.assertRaisesRegex(TypeError, "must be a datetime.timedelta
or Expression"):
+ _to_interval_expression("INTERVAL '10' MINUTE")
+
+ def test_accepts_string_column(self):
+ resolved = _resolve_window_time_column("ts")
+
+ self.assertIsInstance(resolved, Expression)
+ self.assertEqual(str(resolved), "ts")
+
+ def test_accepts_column_expression(self):
+ column = pf.col("ts")
+
+ self.assertIs(_resolve_window_time_column(column), column)
+
+ def test_rejects_wrong_type(self):
+ with self.assertRaisesRegex(TypeError, "on must be a column name or
expression"):
+ _resolve_window_time_column(10)
+
+ def test_tumble_appends_window_columns_timedelta(self):
+ self.assert_dataframe_schema(
+ self.dataframe.tumble(on="ts", size=timedelta(minutes=10)),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ expected_column_data_types=self._window_schema_column_data_types(),
+ )
+
+ def test_tumble_appends_window_columns_expression(self):
+ self.assert_dataframe_schema(
+ self.dataframe.tumble(on="ts", size=pf.lit(10).minutes),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ )
+
+ def test_hop_appends_window_columns(self):
+ self.assert_dataframe_schema(
+ self.dataframe.hop(
+ on="ts",
+ slide=timedelta(minutes=5),
+ size=timedelta(minutes=10),
+ ),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ expected_column_data_types=self._window_schema_column_data_types(),
+ )
+
+ def test_cumulate_appends_window_columns(self):
+ self.assert_dataframe_schema(
+ self.dataframe.cumulate(
+ on="ts",
+ step=timedelta(minutes=2),
+ size=timedelta(minutes=10),
+ ),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ expected_column_data_types=self._window_schema_column_data_types(),
+ )
+
+ def test_session_appends_window_columns(self):
+ self.assert_dataframe_schema(
+ self.dataframe.session(on="ts", gap=timedelta(minutes=10)),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ expected_column_data_types=self._window_schema_column_data_types(),
+ )
+
+ def test_session_with_partition_by_appends_window_columns(self):
+ self.assert_dataframe_schema(
+ self.dataframe.session(
+ on="ts",
+ gap=timedelta(minutes=10),
+ partition_by=["id", pf.col("amount")],
+ ),
+ self._WINDOW_SCHEMA_COLUMN_NAMES,
+ )
+
+ def test_session_partition_by_rejects_unsupported_type(self):
+ with self.assertRaisesRegex(
+ TypeError,
+ "^partition_by must be a column name, expression, or a list of
them$",
+ ):
+ self.dataframe.session(
+ on="ts", gap=timedelta(minutes=10), partition_by=[10]
+ )
+
+
class DataFrameTopNTests(PyFlinkDataFrameUTTestCase):
def setUp(self):
super().setUp()
@@ -1598,6 +1745,13 @@ class DataFrameTopNTests(PyFlinkDataFrameUTTestCase):
with self.assertRaisesRegex(ValueError, "partition_by column 'nope'
does not exist"):
self.df.top_n(2, partition_by="nope", order_by="amount")
+ def test_rejects_non_string_partition_by(self):
+ with self.assertRaises(TypeError) as error:
+ self.df.top_n(2, partition_by=[1], order_by="amount")
+ self.assertEqual(
+ str(error.exception), "partition_by must be a string or a list of
strings"
+ )
+
def test_sql_global_has_no_partition_by(self):
df = pf.from_records([{"id": 1, "name": "a", "score": 10}])
@@ -2176,5 +2330,136 @@ class DataFrameBatchITTests(PyFlinkITTestCase):
)
+class DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.t_env.get_config().set("table.exec.resource.default-parallelism",
"1")
+
+ def _rowtime_source(self):
+ return self._rowtime_source_from_rows(
+ "events.csv", ["1,10,0", "1,20,60000", "1,40,600000"]
+ )
+
+ def _multi_key_rowtime_source(self):
+ return self._rowtime_source_from_rows(
+ "multi_key_events.csv", ["1,10,0", "2,20,60000", "1,40,600000"]
+ )
+
+ def _rowtime_source_from_rows(self, filename, rows):
+ input_path = os.path.join(self.tempdir, filename)
+ with open(input_path, "w", encoding="utf-8") as events:
+ events.write("\n".join(rows) + "\n")
+ return pf.read_generic(
+ "filesystem",
+ schema={
+ "id": DataType.int64(),
+ "amount": DataType.int64(),
+ "ts_millis": DataType.int64(),
+ },
+ options={"path": input_path, "format": "csv"},
+ computed_columns={"event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("event_time", "event_time - INTERVAL '1' SECOND"),
+ )
+
+ def _proctime_source(self):
+ input_path = os.path.join(self.tempdir, "proctime_events.csv")
+ with open(input_path, "w", encoding="utf-8") as events:
+ events.write("1,10\n")
+ events.write("1,20\n")
+ events.write("1,40\n")
+ return pf.read_generic(
+ "filesystem",
+ schema={
+ "id": DataType.int64(),
+ "amount": DataType.int64(),
+ },
+ options={"path": input_path, "format": "csv"},
+ computed_columns={"proc_time": "PROCTIME()"},
+ )
+
+ def test_tumble_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .tumble(on="event_time", size=timedelta(minutes=10))
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
40])
+
+ def test_hop_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .hop(
+ on="event_time",
+ slide=timedelta(minutes=5),
+ size=timedelta(minutes=10),
+ )
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
30, 40, 40])
+
+ def test_cumulate_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .cumulate(
+ on="event_time",
+ step=timedelta(minutes=5),
+ size=timedelta(minutes=10),
+ )
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
30, 40, 40])
+
+ def test_session_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .session(on="event_time", gap=timedelta(minutes=5))
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+ rows = self._materialize(windowed, key=["window_start", "window_end",
"id"])
+
+ self.assertEqual(sorted(row[-1] for row in rows), [30, 40])
+
+ def test_session_partition_by_computes_per_key_sessions(self):
+ partitioned = (
+ self._multi_key_rowtime_source()
+ .session(
+ on="event_time",
+ gap=timedelta(seconds=90),
+ partition_by="id",
+ )
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ partitioned_rows = self._materialize(
+ partitioned, key=["window_start", "window_end", "id"]
+ )
+
+ self.assertEqual(sorted(row[-1] for row in partitioned_rows), [10, 20,
40])
+
+ def test_tumble_processing_time_assigns_aligned_windows(self):
+ rows = (
+ self._proctime_source()
+ .tumble(on="proc_time", size=timedelta(minutes=10))
+ .select("window_start", "window_end")
+ .collect()
+ )
+
+ self.assertEqual(len(rows), 3)
+ for start, end in rows:
+ self.assertEqual(end - start, timedelta(minutes=10))
+ self.assertEqual(
+ (start.minute % 10, start.second, start.microsecond), (0, 0, 0)
+ )
+
+
if __name__ == "__main__":
unittest.main()
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperationVisitor.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperationVisitor.java
index 01fd08cf362..02eb183b63b 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperationVisitor.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperationVisitor.java
@@ -33,6 +33,8 @@ public interface QueryOperationVisitor<T> {
T visit(WindowAggregateQueryOperation windowAggregate);
+ T visit(WindowTableFunctionQueryOperation windowTableFunction);
+
T visit(JoinQueryOperation join);
T visit(SetQueryOperation setOperation);
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperation.java
new file mode 100644
index 00000000000..ae736e996b8
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperation.java
@@ -0,0 +1,222 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.expressions.ApiExpressionUtils;
+import org.apache.flink.table.expressions.SqlFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.EncodingUtils;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.canBeTimeAttributeType;
+
+/**
+ * Relational operation that assigns rows to windows using a windowing
table-valued function
+ * (TUMBLE/HOP/CUMULATE/SESSION). Appends {@code window_start}, {@code
window_end} and {@code
+ * window_time} to the input and returns the enriched relation.
+ */
+@Internal
+public class WindowTableFunctionQueryOperation implements QueryOperation {
+
+ private static final String INPUT_ALIAS = "$$T_WIN";
+
+ /** Window kind; the name maps to the SQL window operator. */
+ @Internal
+ public enum WindowKind {
+ TUMBLE(1),
+ HOP(2),
+ CUMULATE(2),
+ SESSION(1);
+
+ private final int expectedIntervalCount;
+
+ WindowKind(int expectedIntervalCount) {
+ this.expectedIntervalCount = expectedIntervalCount;
+ }
+
+ /** Number of interval operands this window kind requires. */
+ public int expectedIntervalCount() {
+ return expectedIntervalCount;
+ }
+ }
+
+ private final WindowKind windowKind;
+ private final String timeColumn;
+ private final List<Duration> intervals;
+ private final List<String> partitionKeys;
+ private final QueryOperation child;
+ private final ResolvedSchema resolvedSchema;
+
+ public WindowTableFunctionQueryOperation(
+ WindowKind windowKind,
+ String timeColumn,
+ List<Duration> intervals,
+ QueryOperation child) {
+ this(windowKind, timeColumn, intervals, List.of(), child);
+ }
+
+ public WindowTableFunctionQueryOperation(
+ WindowKind windowKind,
+ String timeColumn,
+ List<Duration> intervals,
+ List<String> partitionKeys,
+ QueryOperation child) {
+ if (intervals.size() != windowKind.expectedIntervalCount()) {
+ throw new ValidationException(
+ String.format(
+ "Window kind %s requires %d interval(s), but got
%d.",
+ windowKind, windowKind.expectedIntervalCount(),
intervals.size()));
+ }
+ final ResolvedSchema inputSchema = child.getResolvedSchema();
+ final int timeIndex = inputSchema.getColumnNames().indexOf(timeColumn);
+ if (timeIndex < 0) {
+ throw new ValidationException(
+ String.format(
+ "Window time column '%s' does not exist. Available
columns: %s",
+ timeColumn, inputSchema.getColumnNames()));
+ }
+ for (String partitionKey : partitionKeys) {
+ if (inputSchema.getColumnNames().indexOf(partitionKey) < 0) {
+ throw new ValidationException(
+ String.format(
+ "Window partition column '%s' does not exist. "
+ + "Available columns: %s",
+ partitionKey, inputSchema.getColumnNames()));
+ }
+ }
+ final DataType timeType =
inputSchema.getColumnDataTypes().get(timeIndex);
+ if (!canBeTimeAttributeType(timeType.getLogicalType())) {
+ throw new ValidationException(
+ String.format(
+ "Window time column '%s' must be a TIMESTAMP or
TIMESTAMP_LTZ column, "
+ + "but was %s.",
+ timeColumn, timeType.getLogicalType()));
+ }
+ this.windowKind = windowKind;
+ this.timeColumn = timeColumn;
+ this.intervals = intervals;
+ this.partitionKeys = partitionKeys;
+ this.child = child;
+ this.resolvedSchema = deriveResolvedSchema(inputSchema, timeType);
+ }
+
+ private static ResolvedSchema deriveResolvedSchema(
+ ResolvedSchema inputSchema, DataType timeColumnType) {
+ final List<Column> columns = new ArrayList<>(inputSchema.getColumns());
+ columns.add(Column.physical("window_start",
DataTypes.TIMESTAMP(3).notNull()));
+ columns.add(Column.physical("window_end",
DataTypes.TIMESTAMP(3).notNull()));
+ columns.add(Column.physical("window_time", timeColumnType.notNull()));
+ return ResolvedSchema.of(columns);
+ }
+
+ public WindowKind getWindowKind() {
+ return windowKind;
+ }
+
+ public String getTimeColumn() {
+ return timeColumn;
+ }
+
+ public List<Duration> getIntervals() {
+ return intervals;
+ }
+
+ public List<String> getPartitionKeys() {
+ return partitionKeys;
+ }
+
+ public QueryOperation getChild() {
+ return child;
+ }
+
+ @Override
+ public ResolvedSchema getResolvedSchema() {
+ return resolvedSchema;
+ }
+
+ @Override
+ public String asSummaryString() {
+ final Map<String, Object> args = new LinkedHashMap<>();
+ args.put("window", windowKind);
+ args.put("timeCol", timeColumn);
+ args.put("intervals", intervals);
+ if (!partitionKeys.isEmpty()) {
+ args.put("partitionKeys", partitionKeys);
+ }
+ return OperationUtils.formatWithChildren(
+ "WindowTableFunction", args, getChildren(),
Operation::asSummaryString);
+ }
+
+ @Override
+ public String asSerializableString(SqlFactory sqlFactory) {
+ return String.format(
+ "SELECT %s FROM TABLE(%s\n) %s",
+ OperationUtils.formatSelectColumns(getResolvedSchema(),
INPUT_ALIAS),
+ OperationUtils.indent(serializeWindowCall(sqlFactory)),
+ INPUT_ALIAS);
+ }
+
+ private String serializeWindowCall(SqlFactory sqlFactory) {
+ final String childSql =
OperationUtils.indent(child.asSerializableString(sqlFactory));
+ final String descriptor = "DESCRIPTOR(" +
EncodingUtils.escapeIdentifier(timeColumn) + ")";
+ final String serializedIntervals =
+ intervals.stream()
+ .map(
+ d ->
+
ApiExpressionUtils.intervalOfMillis(d.toMillis())
+
.asSerializableString(sqlFactory))
+ .collect(Collectors.joining(", "));
+ final String partitionClause = serializePartitionClause();
+ return String.format(
+ "%s((%s\n)%s, %s, %s)",
+ windowKind, childSql, partitionClause, descriptor,
serializedIntervals);
+ }
+
+ private String serializePartitionClause() {
+ if (partitionKeys.isEmpty()) {
+ return "";
+ }
+ final String keys =
+ partitionKeys.stream()
+ .map(EncodingUtils::escapeIdentifier)
+ .collect(Collectors.joining(", "));
+ return " PARTITION BY (" + keys + ")";
+ }
+
+ @Override
+ public List<QueryOperation> getChildren() {
+ return List.of(child);
+ }
+
+ @Override
+ public <T> T accept(QueryOperationVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
index 6eb66edfb28..290999be8e7 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
@@ -32,8 +32,11 @@ import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.expressions.ApiExpressionUtils;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ExpressionUtils;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.UnresolvedCallExpression;
+import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.expressions.resolver.ExpressionResolver;
import org.apache.flink.table.expressions.resolver.LookupCallResolver;
import org.apache.flink.table.expressions.resolver.SqlExpressionResolver;
@@ -50,7 +53,9 @@ import
org.apache.flink.table.operations.PartitionQueryOperation;
import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.operations.ValuesQueryOperation;
import
org.apache.flink.table.operations.WindowAggregateQueryOperation.ResolvedGroupWindow;
+import org.apache.flink.table.operations.WindowTableFunctionQueryOperation;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.utils.DataTypeUtils;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.table.typeutils.FieldInfoUtils;
@@ -58,6 +63,7 @@ import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -363,6 +369,70 @@ public final class OperationTreeBuilder {
flattenedProjection, aggregateWithAlias.aliases,
groupingExpressions.size());
}
+ public QueryOperation windowTableFunction(
+ WindowTableFunctionQueryOperation.WindowKind windowKind,
+ Expression timeColumn,
+ List<Expression> intervals,
+ QueryOperation child) {
+ return windowTableFunction(windowKind, timeColumn, intervals,
List.of(), child);
+ }
+
+ public QueryOperation windowTableFunction(
+ WindowTableFunctionQueryOperation.WindowKind windowKind,
+ Expression timeColumn,
+ List<Expression> intervals,
+ List<Expression> partitionColumns,
+ QueryOperation child) {
+ final String timeColumnName = extractColumnName(timeColumn);
+ final List<Duration> durations = new ArrayList<>(intervals.size());
+ for (Expression interval : intervals) {
+ durations.add(toDayTimeDuration(interval));
+ }
+ final List<String> partitionKeys = new
ArrayList<>(partitionColumns.size());
+ for (Expression partitionColumn : partitionColumns) {
+ partitionKeys.add(extractColumnName(partitionColumn));
+ }
+ return new WindowTableFunctionQueryOperation(
+ windowKind, timeColumnName, durations, partitionKeys, child);
+ }
+
+ private static String extractColumnName(Expression column) {
+ final Expression unwrapped = ApiExpressionUtils.unwrapFromApi(column);
+ if (unwrapped instanceof UnresolvedReferenceExpression) {
+ return ((UnresolvedReferenceExpression) unwrapped).getName();
+ }
+ if (unwrapped instanceof FieldReferenceExpression) {
+ return ((FieldReferenceExpression) unwrapped).getName();
+ }
+ throw new ValidationException(
+ "Window column must be a column reference, but was: " +
column.asSummaryString());
+ }
+
+ private static Duration toDayTimeDuration(Expression interval) {
+ final Expression unwrapped =
ApiExpressionUtils.unwrapFromApi(interval);
+ if (!(unwrapped instanceof ValueLiteralExpression)) {
+ throw new ValidationException(
+ "Window interval must be an interval literal, e.g.
lit(10).minutes().");
+ }
+ final ValueLiteralExpression literal = (ValueLiteralExpression)
unwrapped;
+ final LogicalTypeRoot root =
literal.getOutputDataType().getLogicalType().getTypeRoot();
+ if (root != LogicalTypeRoot.INTERVAL_DAY_TIME) {
+ throw new ValidationException(
+ "Window interval must be a day-time interval, but was " +
root + ".");
+ }
+ final Duration duration =
+ literal.getValueAs(Duration.class)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "Could not read interval value
from " + literal));
+ if (duration.isZero() || duration.isNegative()) {
+ throw new ValidationException(
+ "Window interval must be positive, but was " +
duration.toMillis() + " ms.");
+ }
+ return duration;
+ }
+
public QueryOperation join(
QueryOperation left,
QueryOperation right,
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/QueryOperationDefaultVisitor.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/QueryOperationDefaultVisitor.java
index 08029f22f2a..a2782230f0a 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/QueryOperationDefaultVisitor.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/QueryOperationDefaultVisitor.java
@@ -35,6 +35,7 @@ import org.apache.flink.table.operations.SourceQueryOperation;
import org.apache.flink.table.operations.TableSourceQueryOperation;
import org.apache.flink.table.operations.ValuesQueryOperation;
import org.apache.flink.table.operations.WindowAggregateQueryOperation;
+import org.apache.flink.table.operations.WindowTableFunctionQueryOperation;
/**
* A utility {@link QueryOperationVisitor} that calls {@link
@@ -59,6 +60,11 @@ public abstract class QueryOperationDefaultVisitor<T>
implements QueryOperationV
return defaultMethod(windowAggregate);
}
+ @Override
+ public T visit(WindowTableFunctionQueryOperation windowTableFunction) {
+ return defaultMethod(windowTableFunction);
+ }
+
@Override
public T visit(JoinQueryOperation join) {
return defaultMethod(join);
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperationTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperationTest.java
new file mode 100644
index 00000000000..bc91b1c70df
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/WindowTableFunctionQueryOperationTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import
org.apache.flink.table.operations.WindowTableFunctionQueryOperation.WindowKind;
+import org.apache.flink.table.types.AtomicDataType;
+import org.apache.flink.table.types.logical.TimestampKind;
+import org.apache.flink.table.types.logical.TimestampType;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for describing {@link WindowTableFunctionQueryOperation}. */
+class WindowTableFunctionQueryOperationTest {
+
+ private static QueryOperation source() {
+ final ResolvedSchema schema =
+ ResolvedSchema.of(
+ Column.physical("id", DataTypes.INT()),
+ Column.physical("amount", DataTypes.INT()),
+ Column.physical(
+ "ts",
+ new AtomicDataType(
+ new TimestampType(false,
TimestampKind.ROWTIME, 3))));
+ return new SourceQueryOperation(
+ ContextResolvedTable.temporary(
+ ObjectIdentifier.of("cat1", "db1", "src"),
+ new ResolvedCatalogTable(
+ CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().build())
+ .build(),
+ schema)));
+ }
+
+ @Test
+ void testConstructorAppendsWindowColumns() {
+ assertThat(
+ new WindowTableFunctionQueryOperation(
+ WindowKind.TUMBLE,
+ "ts",
+ List.of(Duration.ofMinutes(10)),
+ source())
+ .getResolvedSchema()
+ .getColumnNames())
+ .containsExactly("id", "amount", "ts", "window_start",
"window_end", "window_time");
+ }
+
+ @Test
+ void testWindowColumnsMatchPlannerNullability() {
+ final ResolvedSchema schema =
+ new WindowTableFunctionQueryOperation(
+ WindowKind.TUMBLE, "ts",
List.of(Duration.ofMinutes(10)), source())
+ .getResolvedSchema();
+
+
assertThat(schema.getColumn("window_start").orElseThrow(AssertionError::new).getDataType())
+ .isEqualTo(DataTypes.TIMESTAMP(3).notNull());
+
assertThat(schema.getColumn("window_end").orElseThrow(AssertionError::new).getDataType())
+ .isEqualTo(DataTypes.TIMESTAMP(3).notNull());
+
assertThat(schema.getColumn("window_time").orElseThrow(AssertionError::new).getDataType())
+ .isEqualTo(DataTypes.TIMESTAMP(3).notNull());
+ }
+
+ @Test
+ void testWrongIntervalCountRejected() {
+ assertThatThrownBy(
+ () ->
+ new WindowTableFunctionQueryOperation(
+ WindowKind.HOP,
+ "ts",
+ List.of(Duration.ofMinutes(10)),
+ source()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("HOP")
+ .hasMessageContaining("requires 2 interval(s), but got 1");
+ }
+
+ @Test
+ void testIntervalsArePreserved() {
+ assertThat(
+ new WindowTableFunctionQueryOperation(
+ WindowKind.HOP,
+ "ts",
+ List.of(Duration.ofMinutes(5),
Duration.ofMinutes(10)),
+ source())
+ .getIntervals())
+ .containsExactly(Duration.ofMinutes(5),
Duration.ofMinutes(10));
+ }
+
+ @Test
+ void testUnknownTimeColumnRejected() {
+ assertThatThrownBy(
+ () ->
+ new WindowTableFunctionQueryOperation(
+ WindowKind.TUMBLE,
+ "nope",
+ List.of(Duration.ofMinutes(10)),
+ source()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("nope");
+ }
+
+ @Test
+ void testNonTimestampTimeColumnRejected() {
+ assertThatThrownBy(
+ () ->
+ new WindowTableFunctionQueryOperation(
+ WindowKind.TUMBLE,
+ "amount",
+ List.of(Duration.ofMinutes(10)),
+ source()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("TIMESTAMP or TIMESTAMP_LTZ");
+ }
+
+ @Test
+ void testAsSerializableStringEmitsWindowTvf() {
+ String sql =
+ new WindowTableFunctionQueryOperation(
+ WindowKind.TUMBLE, "ts",
List.of(Duration.ofMinutes(10)), source())
+ .asSerializableString();
+
assertThat(sql).contains("TUMBLE((").contains("DESCRIPTOR(`ts`)").contains("INTERVAL");
+ }
+
+ @Test
+ void testPartitionKeysArePreserved() {
+ assertThat(
+ new WindowTableFunctionQueryOperation(
+ WindowKind.SESSION,
+ "ts",
+ List.of(Duration.ofMinutes(10)),
+ List.of("id"),
+ source())
+ .getPartitionKeys())
+ .containsExactly("id");
+ }
+
+ @Test
+ void testAsSerializableStringEmitsPartitionBy() {
+ String sql =
+ new WindowTableFunctionQueryOperation(
+ WindowKind.SESSION,
+ "ts",
+ List.of(Duration.ofMinutes(10)),
+ List.of("id"),
+ source())
+ .asSerializableString();
+ assertThat(sql).contains("SESSION((").contains(") PARTITION BY
(`id`)");
+ }
+
+ @Test
+ void testUnknownPartitionColumnRejected() {
+ assertThatThrownBy(
+ () ->
+ new WindowTableFunctionQueryOperation(
+ WindowKind.SESSION,
+ "ts",
+ List.of(Duration.ofMinutes(10)),
+ List.of("nope"),
+ source()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("nope");
+ }
+}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingTableApiTestStep.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingTableApiTestStep.java
index ab91f955025..2277b0b37c2 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingTableApiTestStep.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingTableApiTestStep.java
@@ -24,8 +24,11 @@ import org.apache.flink.table.api.Table;
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.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.expressions.DefaultSqlFactory;
import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.test.program.TableApiTestStep.TableEnvAccessor;
import org.apache.flink.table.types.AbstractDataType;
import org.apache.flink.util.Preconditions;
@@ -104,6 +107,16 @@ public final class FailingTableApiTestStep implements
TestStep {
return env.sqlQuery(query);
}
+ @Override
+ public Table createTable(QueryOperation operation) {
+ return ((TableEnvironmentImpl)
env).createTable(operation);
+ }
+
+ @Override
+ public OperationTreeBuilder getOperationTreeBuilder() {
+ return ((TableEnvironmentImpl)
env).getOperationTreeBuilder();
+ }
+
@Override
public Model fromModel(String modelPath) {
return env.fromModel(modelPath);
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableApiTestStep.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableApiTestStep.java
index 176e067addb..9df033a4836 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableApiTestStep.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableApiTestStep.java
@@ -24,8 +24,11 @@ import org.apache.flink.table.api.ModelDescriptor;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.expressions.DefaultSqlFactory;
import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.utils.OperationTreeBuilder;
import org.apache.flink.table.types.AbstractDataType;
import java.util.function.Function;
@@ -84,6 +87,16 @@ public class TableApiTestStep implements TestStep {
return env.sqlQuery(query);
}
+ @Override
+ public Table createTable(QueryOperation operation) {
+ return ((TableEnvironmentImpl)
env).createTable(operation);
+ }
+
+ @Override
+ public OperationTreeBuilder getOperationTreeBuilder() {
+ return ((TableEnvironmentImpl)
env).getOperationTreeBuilder();
+ }
+
@Override
public Model fromModel(String modelPath) {
return env.fromModel(modelPath);
@@ -137,6 +150,12 @@ public class TableApiTestStep implements TestStep {
/** See {@link TableEnvironment#sqlQuery(String)}. */
Table sqlQuery(String query);
+ /** See {@link TableEnvironmentImpl#createTable(QueryOperation)}. */
+ Table createTable(QueryOperation operation);
+
+ /** See {@link TableEnvironmentImpl#getOperationTreeBuilder()}. */
+ OperationTreeBuilder getOperationTreeBuilder();
+
/** See {@link TableEnvironment#fromModel(String)}. */
Model fromModel(String modelPath);
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
index 41b5ebe632a..468430dbc7b 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
@@ -65,6 +65,7 @@ import
org.apache.flink.table.operations.TableSourceQueryOperation;
import org.apache.flink.table.operations.ValuesQueryOperation;
import org.apache.flink.table.operations.WindowAggregateQueryOperation;
import
org.apache.flink.table.operations.WindowAggregateQueryOperation.ResolvedGroupWindow;
+import org.apache.flink.table.operations.WindowTableFunctionQueryOperation;
import org.apache.flink.table.operations.utils.QueryOperationDefaultVisitor;
import org.apache.flink.table.planner.calcite.FlinkContext;
import org.apache.flink.table.planner.calcite.FlinkRelBuilder;
@@ -76,6 +77,8 @@ import
org.apache.flink.table.planner.expressions.RexNodeExpression;
import org.apache.flink.table.planner.expressions.SqlAggFunctionVisitor;
import
org.apache.flink.table.planner.expressions.converter.ExpressionConverter;
import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable;
+import org.apache.flink.table.planner.functions.sql.SqlWindowTableFunction;
import org.apache.flink.table.planner.functions.utils.TableSqlFunction;
import
org.apache.flink.table.planner.operations.InternalDataStreamQueryOperation;
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
@@ -104,6 +107,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.DataTypeUtils;
import org.apache.flink.util.Preconditions;
+import org.apache.calcite.avatica.util.TimeUnit;
import org.apache.calcite.plan.ViewExpanders;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.CorrelationId;
@@ -113,13 +117,19 @@ import
org.apache.calcite.rel.logical.LogicalTableFunctionScan;
import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.rel.logical.LogicalValues;
import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RelBuilder.AggCall;
import org.apache.calcite.tools.RelBuilder.GroupKey;
+import java.math.BigDecimal;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -227,6 +237,11 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
.build();
}
+ @Override
+ public RelNode visit(WindowTableFunctionQueryOperation
windowTableFunction) {
+ return convertWindowTableFunction(windowTableFunction);
+ }
+
private NamedWindowProperty convertToWindowProperty(
Expression expression, WindowReference windowReference) {
Preconditions.checkArgument(
@@ -593,6 +608,65 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
throw new TableException("Unknown table operation: " + other);
}
+ private RelNode
convertWindowTableFunction(WindowTableFunctionQueryOperation windowOp) {
+ final RexBuilder rexBuilder = relBuilder.getRexBuilder();
+
+ final RelNode input = relBuilder.build();
+ final RelDataType inputRowType = input.getRowType();
+
+ final String timeColumn = windowOp.getTimeColumn();
+ final RelDataType timeAttributeType =
+ inputRowType.getField(timeColumn, true, false).getType();
+ final RelDataType outputRowType =
+ SqlWindowTableFunction.inferRowType(
+ relBuilder.getTypeFactory(), inputRowType,
timeAttributeType);
+
+ final int[] partitionIndices =
+ windowOp.getPartitionKeys().stream()
+ .mapToInt(name -> inputRowType.getField(name,
true, false).getIndex())
+ .toArray();
+
+ final SqlFunction operator =
windowOperator(windowOp.getWindowKind());
+ final List<RexNode> operands = new ArrayList<>();
+ operands.add(
+ new RexTableArgCall(
+ inputRowType, 0, partitionIndices, new int[0], new
SortOrder[0]));
+ operands.add(
+ rexBuilder.makeCall(
+ FlinkSqlOperatorTable.DESCRIPTOR,
rexBuilder.makeLiteral(timeColumn)));
+ for (Duration interval : windowOp.getIntervals()) {
+ operands.add(
+ rexBuilder.makeIntervalLiteral(
+ BigDecimal.valueOf(interval.toMillis()),
+ new SqlIntervalQualifier(
+ TimeUnit.MILLISECOND, null,
SqlParserPos.ZERO)));
+ }
+
+ final RexNode call = rexBuilder.makeCall(outputRowType, operator,
operands);
+ return LogicalTableFunctionScan.create(
+ relBuilder.getCluster(),
+ new ArrayList<>(Collections.singletonList(input)),
+ call,
+ null,
+ outputRowType,
+ Set.of());
+ }
+
+ private SqlFunction
windowOperator(WindowTableFunctionQueryOperation.WindowKind kind) {
+ switch (kind) {
+ case TUMBLE:
+ return FlinkSqlOperatorTable.TUMBLE;
+ case HOP:
+ return FlinkSqlOperatorTable.HOP;
+ case CUMULATE:
+ return FlinkSqlOperatorTable.CUMULATE;
+ case SESSION:
+ return FlinkSqlOperatorTable.SESSION;
+ default:
+ throw new TableException("Unsupported window kind: " +
kind);
+ }
+ }
+
@Override
public <U> RelNode visit(TableSourceQueryOperation<U>
tableSourceOperation) {
TableSource<?> tableSource = tableSourceOperation.getTableSource();
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
index 2862284602d..46c7122d116 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
@@ -50,6 +50,11 @@ public class QueryOperationSqlSemanticTest extends
SemanticTestBase {
QueryOperationTestPrograms.LIMIT_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_TUMBLE_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_HOP_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_CUMULATE_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_SESSION_QUERY_OPERATION,
+
QueryOperationTestPrograms.WINDOW_TVF_SESSION_PARTITION_QUERY_OPERATION,
QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION,
QueryOperationTestPrograms.LATERAL_JOIN_QUERY_OPERATION,
QueryOperationTestPrograms.GROUP_HOP_WINDOW_EVENT_TIME,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
index d3175a773bb..7b5795fe9a7 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
@@ -71,6 +71,11 @@ public class QueryOperationSqlSerializationTest implements
TableTestProgramRunne
QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_PROCTIME_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_TUMBLE_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_HOP_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_CUMULATE_QUERY_OPERATION,
+ QueryOperationTestPrograms.WINDOW_TVF_SESSION_QUERY_OPERATION,
+
QueryOperationTestPrograms.WINDOW_TVF_SESSION_PARTITION_QUERY_OPERATION,
QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION,
QueryOperationTestPrograms.LATERAL_JOIN_QUERY_OPERATION,
QueryOperationTestPrograms.SQL_QUERY_OPERATION,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
index 0035c9154df..6c11367955f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
@@ -23,6 +23,7 @@ import
org.apache.flink.table.api.config.ExecutionConfigOptions;
import
org.apache.flink.table.api.config.ExecutionConfigOptions.AsyncOutputMode;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.WindowTableFunctionQueryOperation;
import org.apache.flink.table.planner.factories.TestValuesModelFactory;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedSendingFunction;
@@ -44,6 +45,7 @@ import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import static org.apache.flink.table.api.Expressions.$;
@@ -393,10 +395,341 @@ public class QueryOperationTestPrograms {
+ ") $$T_PROJECT")
.build();
+ static final TableTestProgram WINDOW_TVF_TUMBLE_QUERY_OPERATION =
+ TableTestProgram.of(
+ "window-tvf-tumble-query-operation",
+ "verifies sql serialization of a programmatic
window TVF operation")
+ .setupTableSource(
+ SourceTestStep.newBuilder("s")
+ .addSchema(
+ "a bigint",
+ "b string",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedValues(
+ Row.of(2L, "apple",
localOfSeconds(0)),
+ Row.of(3L, "apple",
localOfSeconds(4)),
+ Row.of(1L, "apple",
localOfSeconds(7)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "w_start TIMESTAMP(3)",
+ "w_end TIMESTAMP(3)",
+ "a_sum bigint")
+ .consumedValues(
+ Row.of(localOfSeconds(0),
localOfSeconds(5), 5L),
+ Row.of(localOfSeconds(5),
localOfSeconds(10), 1L))
+ .build())
+ .runTableApi(
+ env ->
+ env.createTable(
+
env.getOperationTreeBuilder()
+
.windowTableFunction(
+
WindowTableFunctionQueryOperation
+
.WindowKind.TUMBLE,
+ $("ts"),
+
List.of(lit(5).seconds()),
+
env.from("s")
+
.getQueryOperation()))
+ .groupBy($("window_start"),
$("window_end"))
+ .select(
+ $("window_start"),
+ $("window_end"),
+ $("a").sum()),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`window_start`,
`$$T_PROJECT`.`window_end`, "
+ + "`$$T_PROJECT`.`EXPR$0` FROM (\n"
+ + " SELECT `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`, "
+ + "(SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM
(\n"
+ + " SELECT `$$T_WIN`.`a`,
`$$T_WIN`.`b`, `$$T_WIN`.`ts`, "
+ + "`$$T_WIN`.`window_start`,
`$$T_WIN`.`window_end`, "
+ + "`$$T_WIN`.`window_time` FROM TABLE(\n"
+ + " TUMBLE((\n"
+ + " SELECT
`$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, "
+ + "`$$T_SOURCE`.`ts` FROM
`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ + " ), DESCRIPTOR(`ts`),
INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3))\n"
+ + " ) $$T_WIN\n"
+ + " ) $$T_AGG\n"
+ + " GROUP BY `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`\n"
+ + ") $$T_PROJECT")
+ .build();
+
+ static final TableTestProgram WINDOW_TVF_HOP_QUERY_OPERATION =
+ TableTestProgram.of(
+ "window-tvf-hop-query-operation",
+ "verifies sql serialization of a programmatic HOP
window TVF operation")
+ .setupTableSource(
+ SourceTestStep.newBuilder("s")
+ .addSchema(
+ "a bigint",
+ "b string",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedValues(
+ Row.of(2L, "apple",
localOfSeconds(0)),
+ Row.of(3L, "apple",
localOfSeconds(4)),
+ Row.of(1L, "apple",
localOfSeconds(7)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "w_start TIMESTAMP(3)",
+ "w_end TIMESTAMP(3)",
+ "a_sum bigint")
+ .consumedValues(
+ Row.of(
+ LocalDateTime.of(2023, 12,
31, 23, 59, 55),
+ localOfSeconds(5),
+ 5L),
+ Row.of(localOfSeconds(0),
localOfSeconds(10), 6L),
+ Row.of(localOfSeconds(5),
localOfSeconds(15), 1L))
+ .build())
+ .runTableApi(
+ env ->
+ env.createTable(
+
env.getOperationTreeBuilder()
+
.windowTableFunction(
+
WindowTableFunctionQueryOperation
+
.WindowKind.HOP,
+ $("ts"),
+ List.of(
+
lit(5).seconds(),
+
lit(10).seconds()),
+
env.from("s")
+
.getQueryOperation()))
+ .groupBy($("window_start"),
$("window_end"))
+ .select(
+ $("window_start"),
+ $("window_end"),
+ $("a").sum()),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`window_start`,
`$$T_PROJECT`.`window_end`, "
+ + "`$$T_PROJECT`.`EXPR$0` FROM (\n"
+ + " SELECT `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`, "
+ + "(SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM
(\n"
+ + " SELECT `$$T_WIN`.`a`,
`$$T_WIN`.`b`, `$$T_WIN`.`ts`, "
+ + "`$$T_WIN`.`window_start`,
`$$T_WIN`.`window_end`, "
+ + "`$$T_WIN`.`window_time` FROM TABLE(\n"
+ + " HOP((\n"
+ + " SELECT
`$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, "
+ + "`$$T_SOURCE`.`ts` FROM
`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ + " ), DESCRIPTOR(`ts`),
INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3), "
+ + "INTERVAL '0 00:00:10.000' DAY(2) TO
SECOND(3))\n"
+ + " ) $$T_WIN\n"
+ + " ) $$T_AGG\n"
+ + " GROUP BY `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`\n"
+ + ") $$T_PROJECT")
+ .build();
+
+ static final TableTestProgram WINDOW_TVF_CUMULATE_QUERY_OPERATION =
+ TableTestProgram.of(
+ "window-tvf-cumulate-query-operation",
+ "verifies sql serialization of a programmatic
CUMULATE window TVF operation")
+ .setupTableSource(
+ SourceTestStep.newBuilder("s")
+ .addSchema(
+ "a bigint",
+ "b string",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedValues(
+ Row.of(2L, "apple",
localOfSeconds(0)),
+ Row.of(3L, "apple",
localOfSeconds(4)),
+ Row.of(1L, "apple",
localOfSeconds(7)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "w_start TIMESTAMP(3)",
+ "w_end TIMESTAMP(3)",
+ "a_sum bigint")
+ .consumedValues(
+ Row.of(localOfSeconds(0),
localOfSeconds(5), 5L),
+ Row.of(localOfSeconds(0),
localOfSeconds(10), 6L))
+ .build())
+ .runTableApi(
+ env ->
+ env.createTable(
+
env.getOperationTreeBuilder()
+
.windowTableFunction(
+
WindowTableFunctionQueryOperation
+
.WindowKind.CUMULATE,
+ $("ts"),
+ List.of(
+
lit(5).seconds(),
+
lit(10).seconds()),
+
env.from("s")
+
.getQueryOperation()))
+ .groupBy($("window_start"),
$("window_end"))
+ .select(
+ $("window_start"),
+ $("window_end"),
+ $("a").sum()),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`window_start`,
`$$T_PROJECT`.`window_end`, "
+ + "`$$T_PROJECT`.`EXPR$0` FROM (\n"
+ + " SELECT `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`, "
+ + "(SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM
(\n"
+ + " SELECT `$$T_WIN`.`a`,
`$$T_WIN`.`b`, `$$T_WIN`.`ts`, "
+ + "`$$T_WIN`.`window_start`,
`$$T_WIN`.`window_end`, "
+ + "`$$T_WIN`.`window_time` FROM TABLE(\n"
+ + " CUMULATE((\n"
+ + " SELECT
`$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, "
+ + "`$$T_SOURCE`.`ts` FROM
`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ + " ), DESCRIPTOR(`ts`),
INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3), "
+ + "INTERVAL '0 00:00:10.000' DAY(2) TO
SECOND(3))\n"
+ + " ) $$T_WIN\n"
+ + " ) $$T_AGG\n"
+ + " GROUP BY `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`\n"
+ + ") $$T_PROJECT")
+ .build();
+
+ static final TableTestProgram WINDOW_TVF_SESSION_QUERY_OPERATION =
+ TableTestProgram.of(
+ "window-tvf-session-query-operation",
+ "verifies sql serialization of a programmatic
SESSION window TVF operation")
+ .setupTableSource(
+ SourceTestStep.newBuilder("s")
+ .addSchema(
+ "a bigint",
+ "b string",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedValues(
+ Row.of(2L, "apple",
localOfSeconds(0)),
+ Row.of(3L, "apple",
localOfSeconds(4)),
+ Row.of(1L, "apple",
localOfSeconds(7)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "w_start TIMESTAMP(3)",
+ "w_end TIMESTAMP(3)",
+ "a_sum bigint")
+ .consumedValues(
+ Row.of(localOfSeconds(0),
localOfSeconds(12), 6L))
+ .build())
+ .runTableApi(
+ env ->
+ env.createTable(
+
env.getOperationTreeBuilder()
+
.windowTableFunction(
+
WindowTableFunctionQueryOperation
+
.WindowKind.SESSION,
+ $("ts"),
+
List.of(lit(5).seconds()),
+
env.from("s")
+
.getQueryOperation()))
+ .groupBy($("window_start"),
$("window_end"))
+ .select(
+ $("window_start"),
+ $("window_end"),
+ $("a").sum()),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`window_start`,
`$$T_PROJECT`.`window_end`, "
+ + "`$$T_PROJECT`.`EXPR$0` FROM (\n"
+ + " SELECT `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`, "
+ + "(SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM
(\n"
+ + " SELECT `$$T_WIN`.`a`,
`$$T_WIN`.`b`, `$$T_WIN`.`ts`, "
+ + "`$$T_WIN`.`window_start`,
`$$T_WIN`.`window_end`, "
+ + "`$$T_WIN`.`window_time` FROM TABLE(\n"
+ + " SESSION((\n"
+ + " SELECT
`$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, "
+ + "`$$T_SOURCE`.`ts` FROM
`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ + " ), DESCRIPTOR(`ts`),
INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3))\n"
+ + " ) $$T_WIN\n"
+ + " ) $$T_AGG\n"
+ + " GROUP BY `$$T_AGG`.`window_start`,
`$$T_AGG`.`window_end`\n"
+ + ") $$T_PROJECT")
+ .build();
+
+ static final TableTestProgram WINDOW_TVF_SESSION_PARTITION_QUERY_OPERATION
=
+ TableTestProgram.of(
+ "window-tvf-session-partition-query-operation",
+ "verifies sql serialization of a programmatic
per-key SESSION window TVF operation")
+ .setupTableSource(
+ SourceTestStep.newBuilder("s")
+ .addSchema(
+ "a bigint",
+ "b string",
+ "ts TIMESTAMP(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'1' SECOND")
+ .producedValues(
+ Row.of(2L, "apple",
localOfSeconds(0)),
+ Row.of(3L, "apple",
localOfSeconds(4)),
+ Row.of(1L, "banana",
localOfSeconds(7)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "b string",
+ "w_start TIMESTAMP(3)",
+ "w_end TIMESTAMP(3)",
+ "a_sum bigint")
+ .consumedValues(
+ Row.of(
+ "apple",
+ localOfSeconds(0),
+ localOfSeconds(9),
+ 5L),
+ Row.of(
+ "banana",
+ localOfSeconds(7),
+ localOfSeconds(12),
+ 1L))
+ .build())
+ .runTableApi(
+ env ->
+ env.createTable(
+
env.getOperationTreeBuilder()
+
.windowTableFunction(
+
WindowTableFunctionQueryOperation
+
.WindowKind.SESSION,
+ $("ts"),
+
List.of(lit(5).seconds()),
+
List.of($("b")),
+
env.from("s")
+
.getQueryOperation()))
+ .groupBy($("b"),
$("window_start"), $("window_end"))
+ .select(
+ $("b"),
+ $("window_start"),
+ $("window_end"),
+ $("a").sum()),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`b`,
`$$T_PROJECT`.`window_start`, "
+ + "`$$T_PROJECT`.`window_end`,
`$$T_PROJECT`.`EXPR$0` FROM (\n"
+ + " SELECT `$$T_AGG`.`b`,
`$$T_AGG`.`window_start`, "
+ + "`$$T_AGG`.`window_end`,
(SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM (\n"
+ + " SELECT `$$T_WIN`.`a`,
`$$T_WIN`.`b`, `$$T_WIN`.`ts`, "
+ + "`$$T_WIN`.`window_start`,
`$$T_WIN`.`window_end`, "
+ + "`$$T_WIN`.`window_time` FROM TABLE(\n"
+ + " SESSION((\n"
+ + " SELECT
`$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, "
+ + "`$$T_SOURCE`.`ts` FROM
`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ + " ) PARTITION BY (`b`),
DESCRIPTOR(`ts`), "
+ + "INTERVAL '0 00:00:05.000' DAY(2) TO
SECOND(3))\n"
+ + " ) $$T_WIN\n"
+ + " ) $$T_AGG\n"
+ + " GROUP BY `$$T_AGG`.`b`,
`$$T_AGG`.`window_start`, "
+ + "`$$T_AGG`.`window_end`\n"
+ + ") $$T_PROJECT")
+ .build();
+
private static Instant dayOfSeconds(int second) {
return LocalDateTime.of(2024, 1, 1, 0, 0,
second).atZone(ZoneId.of("UTC")).toInstant();
}
+ private static LocalDateTime localOfSeconds(int second) {
+ return LocalDateTime.of(2024, 1, 1, 0, 0, second);
+ }
+
static final TableTestProgram JOIN_QUERY_OPERATION =
TableTestProgram.of("join-query-operation", "verifies sql
serialization")
.setupTableSource(
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.java
new file mode 100644
index 00000000000..9a30c2b4257
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.java
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.table.operations.WindowTableFunctionQueryOperation;
+import
org.apache.flink.table.operations.WindowTableFunctionQueryOperation.WindowKind;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Verifies that a programmatically built {@link
WindowTableFunctionQueryOperation} is lowered by
+ * {@code QueryOperationConverter} to the same window-TVF plan as the
equivalent {@code
+ * TABLE(TUMBLE(...))} / {@code HOP} / {@code CUMULATE} / {@code SESSION} SQL.
+ */
+class WindowTableFunctionRelNodeTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ util.tableEnv()
+ .executeSql(
+ "CREATE TEMPORARY TABLE src (\n"
+ + " id INT, amount INT, ts TIMESTAMP(3),\n"
+ + " WATERMARK FOR ts AS ts - INTERVAL '1'
SECOND\n"
+ + ") WITH ('connector' = 'datagen')");
+ }
+
+ @Test
+ void testTumbleRelPlan() {
+ final Table table =
+ tEnv().createTable(
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.TUMBLE,
+ $("ts"),
+ List.of(lit(10).minutes()),
+
tEnv().from("src").getQueryOperation()));
+ util.verifyRelPlan(table);
+ }
+
+ @Test
+ void testHopRelPlan() {
+ final Table table =
+ tEnv().createTable(
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.HOP,
+ $("ts"),
+ List.of(lit(5).minutes(),
lit(10).minutes()),
+
tEnv().from("src").getQueryOperation()));
+ util.verifyRelPlan(table);
+ }
+
+ @Test
+ void testCumulateRelPlan() {
+ final Table table =
+ tEnv().createTable(
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.CUMULATE,
+ $("ts"),
+ List.of(lit(5).minutes(),
lit(10).minutes()),
+
tEnv().from("src").getQueryOperation()));
+ util.verifyRelPlan(table);
+ }
+
+ @Test
+ void testSessionRelPlan() {
+ final Table table =
+ tEnv().createTable(
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.SESSION,
+ $("ts"),
+ List.of(lit(10).minutes()),
+
tEnv().from("src").getQueryOperation()));
+ util.verifyRelPlan(table);
+ }
+
+ @Test
+ void testSessionWithPartitionRelPlan() {
+ final Table table =
+ tEnv().createTable(
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.SESSION,
+ $("ts"),
+ List.of(lit(10).minutes()),
+ List.of($("id")),
+
tEnv().from("src").getQueryOperation()));
+ util.verifyRelPlan(table);
+ }
+
+ @Test
+ void testYearMonthIntervalRejected() {
+ assertThatThrownBy(
+ () ->
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.TUMBLE,
+ $("ts"),
+ List.of(lit(1).month()),
+
tEnv().from("src").getQueryOperation()))
+ .hasMessageContaining("day-time");
+ }
+
+ @Test
+ void testZeroIntervalRejected() {
+ assertThatThrownBy(
+ () ->
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.TUMBLE,
+ $("ts"),
+ List.of(lit(0).seconds()),
+
tEnv().from("src").getQueryOperation()))
+ .hasMessageContaining("positive");
+ }
+
+ @Test
+ void testNegativeIntervalRejected() {
+ assertThatThrownBy(
+ () ->
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.TUMBLE,
+ $("ts"),
+ List.of(lit(-5).seconds()),
+
tEnv().from("src").getQueryOperation()))
+ .hasMessageContaining("positive");
+ }
+
+ @Test
+ void testNonLiteralIntervalRejected() {
+ assertThatThrownBy(
+ () ->
+ tEnv().getOperationTreeBuilder()
+ .windowTableFunction(
+ WindowKind.TUMBLE,
+ $("ts"),
+ List.of($("ts")),
+
tEnv().from("src").getQueryOperation()))
+ .hasMessageContaining("interval literal");
+ }
+
+ private TableEnvironmentImpl tEnv() {
+ return (TableEnvironmentImpl) util.tableEnv();
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.xml
new file mode 100644
index 00000000000..ad458449423
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/WindowTableFunctionRelNodeTest.xml
@@ -0,0 +1,101 @@
+<?xml version="1.0" ?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to you under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<Root>
+ <TestCase name="testCumulateRelPlan">
+ <Resource name="ast">
+ <![CDATA[
+LogicalTableFunctionScan(invocation=[CUMULATE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'ts'), 300000:INTERVAL MILLISECOND, 600000:INTERVAL
MILLISECOND)], rowType=[RecordType(INTEGER id, INTEGER amount, TIMESTAMP(3)
*ROWTIME* ts, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3)
*ROWTIME* window_time)])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, src]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+WindowTableFunction(window=[CUMULATE(time_col=[ts], max_size=[10 min], step=[5
min])])
++- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, src]],
fields=[id, amount, ts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testHopRelPlan">
+ <Resource name="ast">
+ <![CDATA[
+LogicalTableFunctionScan(invocation=[HOP(TABLE(#0), DESCRIPTOR(_UTF-16LE'ts'),
300000:INTERVAL MILLISECOND, 600000:INTERVAL MILLISECOND)],
rowType=[RecordType(INTEGER id, INTEGER amount, TIMESTAMP(3) *ROWTIME* ts,
TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME*
window_time)])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, src]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+WindowTableFunction(window=[HOP(time_col=[ts], size=[10 min], slide=[5 min])])
++- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, src]],
fields=[id, amount, ts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSessionRelPlan">
+ <Resource name="ast">
+ <![CDATA[
+LogicalTableFunctionScan(invocation=[SESSION(TABLE(#0),
DESCRIPTOR(_UTF-16LE'ts'), 600000:INTERVAL MILLISECOND)],
rowType=[RecordType(INTEGER id, INTEGER amount, TIMESTAMP(3) *ROWTIME* ts,
TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME*
window_time)])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, src]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+WindowTableFunction(window=[SESSION(time_col=[ts], gap=[10 min])])
++- Exchange(distribution=[single])
+ +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, src]],
fields=[id, amount, ts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSessionWithPartitionRelPlan">
+ <Resource name="ast">
+ <![CDATA[
+LogicalTableFunctionScan(invocation=[SESSION(TABLE(#0) PARTITION BY($0),
DESCRIPTOR(_UTF-16LE'ts'), 600000:INTERVAL MILLISECOND)],
rowType=[RecordType(INTEGER id, INTEGER amount, TIMESTAMP(3) *ROWTIME* ts,
TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME*
window_time)])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, src]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+WindowTableFunction(window=[SESSION(time_col=[ts], gap=[10 min], partition
keys=[id])])
++- Exchange(distribution=[hash[id]])
+ +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, src]],
fields=[id, amount, ts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testTumbleRelPlan">
+ <Resource name="ast">
+ <![CDATA[
+LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'ts'), 600000:INTERVAL MILLISECOND)],
rowType=[RecordType(INTEGER id, INTEGER amount, TIMESTAMP(3) *ROWTIME* ts,
TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME*
window_time)])
++- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, src]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+WindowTableFunction(window=[TUMBLE(time_col=[ts], size=[10 min])])
++- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, src]],
fields=[id, amount, ts])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
