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 12b0ecfef57 [FLINK-40425][python] Support OVER windows in DataFrame API
12b0ecfef57 is described below
commit 12b0ecfef57524ccfef8398f03bfef4110b98ce8
Author: bladedragon <[email protected]>
AuthorDate: Sat Sep 26 22:56:57 2026 +0800
[FLINK-40425][python] Support OVER windows in DataFrame API
This closes #29303.
---
.../docs/reference/pyflink.dataframe/dataframe.rst | 46 +++
flink-python/pyflink/dataframe/__init__.py | 3 +
flink-python/pyflink/dataframe/over_window.py | 132 +++++++
.../pyflink/dataframe/tests/test_over_window.py | 415 +++++++++++++++++++++
flink-python/pyflink/table/_over_window.py | 218 +++++++++++
flink-python/pyflink/table/expression.py | 107 +++++-
.../resolver/rules/OverWindowResolverRule.java | 50 ++-
.../utils/OperationExpressionsUtils.java | 2 +
.../operations/utils/OperationTreeBuilder.java | 3 +
.../resolver/OverWindowExpressionResolverTest.java | 237 ++++++++++++
.../utils/OperationExpressionsUtilsTest.java | 64 ++++
11 files changed, 1256 insertions(+), 21 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
index 4a443034676..d03d62b27e6 100644
--- a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -203,3 +203,49 @@ Functions for constructing column references and literal
expressions.
col
lit
+
+OVER windows
+------------
+
+An aggregate expression can define an inline OVER window. The ordering column
may be a column
+name or expression. Use the keyword ``order_by=`` to define an inline window;
a single positional
+argument such as ``over(col("w"))`` or ``over(alias=col("w"))`` refers to a
Table API named window.
+If neither ``rows`` nor ``range`` is provided, the default frame is unbounded
RANGE through the
+current range. A scalar ``rows`` bound includes that many preceding rows and
the current row, so
+``rows=10`` covers up to 11 rows. RANGE frames include all peers with the same
ordering value.
+
+``rows`` uses integer row offsets. ``range`` accepts ``datetime.timedelta`` or
a Flink duration
+string such as ``"1 hour"`` or ``"1 h"``; this is not SQL ``INTERVAL '1'
HOUR`` syntax. Duration
+values are converted to milliseconds, truncating sub-millisecond precision.
Use Table API frame
+constants from ``pyflink.table.expressions`` when matched to the frame type:
``*_ROW`` with
+``rows`` and ``*_RANGE`` with ``range``.
+
+The current expression layer requires a single time attribute in ``order_by``.
Streaming OVER
+windows additionally require ascending order, and all OVER aggregates in the
same projection must
+use the same frame. The upper bound must be the current row/range; FOLLOWING
bounds are not
+supported in streaming. Without ``partition_by``, streaming OVER uses a
singleton distribution and
+therefore runs with parallelism one, which can limit throughput. Batch plans
support multiple
+different OVER frames in the same projection.
+
+.. code-block:: python
+
+ import pyflink.dataframe as pf
+
+ running = pf.col("amount").sum.over(
+ order_by="event_time",
+ partition_by="user_id",
+ rows=10,
+ )
+ explicit = pf.col("amount").sum.over(
+ order_by="event_time",
+ rows=(pf.preceding(2), pf.following(1)),
+ )
+
+The public frame-bound descriptor and helpers are:
+
+.. autosummary::
+ :toctree: api/
+
+ over_window.FrameBoundary
+ preceding
+ following
diff --git a/flink-python/pyflink/dataframe/__init__.py
b/flink-python/pyflink/dataframe/__init__.py
index e463bb19d0c..174a3e4df79 100644
--- a/flink-python/pyflink/dataframe/__init__.py
+++ b/flink-python/pyflink/dataframe/__init__.py
@@ -73,6 +73,7 @@ from pyflink.dataframe.io import (
read_parquet,
)
from pyflink.dataframe.iteration import CloseableIterator
+from pyflink.dataframe.over_window import following, preceding
from pyflink.dataframe.sql import sql
from pyflink.dataframe.udf import udf
from pyflink.dataframe.udtf import udtf
@@ -107,6 +108,8 @@ __all__ = [
"list_databases",
"sql",
"config",
+ "preceding",
+ "following",
"set_table_environment",
"get_table_environment",
"get_or_create_table_environment",
diff --git a/flink-python/pyflink/dataframe/over_window.py
b/flink-python/pyflink/dataframe/over_window.py
new file mode 100644
index 00000000000..6565d7d749f
--- /dev/null
+++ b/flink-python/pyflink/dataframe/over_window.py
@@ -0,0 +1,132 @@
+################################################################################
+# 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.
+################################################################################
+
+"""Utilities for constructing DataFrame OVER windows."""
+
+import datetime
+from dataclasses import dataclass
+from typing import Any, Union
+
+from pyflink.table._over_window import _BoundaryDirection, _FrameBoundary
+from pyflink.util.api_stability_decorators import PublicEvolving
+
+__all__ = ["FrameBoundary", "preceding", "following"]
+
+
+@PublicEvolving()
+@dataclass(frozen=True, init=False)
+class FrameBoundary(_FrameBoundary):
+ """
+ An opaque bound for an inline OVER window frame.
+
+ Create instances with :func:`preceding` or :func:`following` instead of
constructing this
+ class directly.
+
+ .. versionadded:: 2.4.0
+ """
+
+ _direction: _BoundaryDirection
+ _value: Any
+
+ def __init__(self) -> None:
+ raise TypeError("FrameBoundary instances must be created with
preceding() or following()")
+
+ @classmethod
+ def _create(cls, direction: _BoundaryDirection, value: Any) ->
"FrameBoundary":
+ boundary = object.__new__(cls)
+ object.__setattr__(boundary, "_direction", direction)
+ object.__setattr__(boundary, "_value", value)
+ return boundary
+
+ def __repr__(self) -> str:
+ return "%s(%r)" % (self._direction.value, self._value)
+
+
+@PublicEvolving()
+def preceding(value: Union[int, str, datetime.timedelta]) -> FrameBoundary:
+ """
+ Create a preceding bound for an inline OVER window frame.
+
+ When used in ``rows``, ``value`` must be a positive integer row count.
When used in
+ ``range``, it must be a non-negative :class:`datetime.timedelta` or a
Flink duration string,
+ such as ``"1 hour"``. The value is validated when the boundary is passed to
+ :meth:`~pyflink.table.expression.Expression.over`.
+
+ :param value: Row count or time interval before the current row or range.
+ :return: A frame-bound descriptor for the lower bound of an explicit OVER
frame.
+
+ Example::
+
+ >>> from datetime import timedelta
+ >>> import pyflink.dataframe as pf
+ >>> from pyflink.table.expressions import CURRENT_RANGE, CURRENT_ROW
+ >>> running = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... rows=(pf.preceding(2), CURRENT_ROW),
+ ... )
+ >>> last_hour = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... range=(pf.preceding("1 hour"), CURRENT_RANGE),
+ ... )
+ >>> last_five_minutes = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... range=(pf.preceding(timedelta(minutes=5)), CURRENT_RANGE),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+
+ return FrameBoundary._create(_BoundaryDirection.PRECEDING, value)
+
+
+@PublicEvolving()
+def following(value: Union[int, str, datetime.timedelta]) -> FrameBoundary:
+ """
+ Create a following bound for an inline OVER window frame.
+
+ When used in ``rows``, ``value`` must be a positive integer row count.
When used in
+ ``range``, it must be a non-negative :class:`datetime.timedelta` or a
Flink duration string,
+ such as ``"1 hour"``. The value is validated when the boundary is passed to
+ :meth:`~pyflink.table.expression.Expression.over`. Following upper bounds
are supported only
+ in batch mode.
+
+ :param value: Row count or time interval after the current row or range.
+ :return: A frame-bound descriptor for the upper bound of an explicit OVER
frame.
+
+ Example::
+
+ >>> from datetime import timedelta
+ >>> import pyflink.dataframe as pf
+ >>> from pyflink.table.expressions import CURRENT_RANGE
+ >>> centered = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... rows=(pf.preceding(2), pf.following(1)),
+ ... )
+ >>> next_hour = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... range=(CURRENT_RANGE, pf.following("1 hour")),
+ ... )
+ >>> next_five_minutes = pf.col("amount").sum.over(
+ ... order_by="event_time",
+ ... range=(CURRENT_RANGE, pf.following(timedelta(minutes=5))),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+
+ return FrameBoundary._create(_BoundaryDirection.FOLLOWING, value)
diff --git a/flink-python/pyflink/dataframe/tests/test_over_window.py
b/flink-python/pyflink/dataframe/tests/test_over_window.py
new file mode 100644
index 00000000000..16826bf7b44
--- /dev/null
+++ b/flink-python/pyflink/dataframe/tests/test_over_window.py
@@ -0,0 +1,415 @@
+################################################################################
+# 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.
+################################################################################
+
+import os
+from datetime import datetime, timedelta
+
+import pyflink.dataframe as pf
+from pyflink.common import Row
+from pyflink.datastream import RuntimeExecutionMode, StreamExecutionEnvironment
+from pyflink.java_gateway import get_gateway
+from pyflink.table import StreamTableEnvironment
+from pyflink.table import expressions as table_expressions
+from pyflink.table.expressions import lit, row
+from pyflink.table.table import Table
+from pyflink.table.utils import to_expression_jarray
+from pyflink.testing.test_case_utils import PyFlinkTestCase
+from pyflink.util.exceptions import TableException
+from pyflink.util.java_utils import to_jarray
+
+
+def _register_rowtime_events(table_environment, input_path, rows):
+ with open(input_path, "w", encoding="utf-8") as events:
+ events.write("\n".join(rows) + "\n")
+ table_environment.execute_sql("DROP TEMPORARY TABLE IF EXISTS events")
+ table_environment.execute_sql(
+ "CREATE TEMPORARY TABLE events ("
+ "amount BIGINT, event_time TIMESTAMP(3), "
+ "WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND"
+ ") WITH ("
+ "'connector' = 'filesystem', "
+ "'path' = 'file://%s', "
+ "'format' = 'csv'"
+ ")" % input_path
+ )
+ return pf.from_table(table_environment.from_path("events"))
+
+
+class OverWindowTests(PyFlinkTestCase):
+
+ def test_dataframe_does_not_export_table_frame_constants(self):
+ for name in (
+ "CURRENT_ROW",
+ "UNBOUNDED",
+ "UNBOUNDED_PRECEDING",
+ "UNBOUNDED_FOLLOWING",
+ ):
+ self.assertFalse(hasattr(pf, name))
+ self.assertTrue(callable(pf.preceding))
+ self.assertTrue(callable(pf.following))
+
+ def test_legacy_over_call_is_preserved(self):
+ self.assertEqual("over(sum(amount), w)",
str(pf.col("amount").sum.over(pf.col("w"))))
+ self.assertEqual(
+ "over(sum(amount), w)",
+ str(pf.col("amount").sum.over(alias=pf.col("w"))),
+ )
+
+ def test_dataframe_over_supports_scalar_and_explicit_frames(self):
+ scalar = pf.col("amount").sum.over(order_by="rowtime", rows=2)
+ explicit = pf.col("amount").sum.over(
+ order_by="rowtime",
+ partition_by="category",
+ rows=(pf.preceding(2), pf.following(1)),
+ )
+
+ self.assertIn("over(sum(amount), rowtime, 2, CURRENT_ROW)",
str(scalar))
+ self.assertIn("over(sum(amount), rowtime, 2, 1, category)",
str(explicit))
+
+ def test_dataframe_over_expression_remains_composable(self):
+ expression = (
+ pf.col("amount").sum.over(order_by="rowtime", rows=2) + 1
+ ).alias("adjusted")
+ self.assertIn("as(plus(over(sum(amount)", str(expression))
+
+ def test_over_rejects_invalid_dispatch(self):
+ with self.assertRaisesRegex(
+ TypeError, "cannot specify both alias and order_by"
+ ):
+ pf.col("amount").sum.over(pf.col("rowtime"),
order_by=pf.col("other"))
+ with self.assertRaisesRegex(TypeError, "cannot combine alias"):
+ pf.col("amount").sum.over(pf.col("rowtime"), rows=2)
+ with self.assertRaisesRegex(TypeError, "unexpected keyword"):
+ pf.col("amount").sum.over(order_by="rowtime", unknown=True)
+
+ def test_frame_validation_through_public_over_api(self):
+ with self.assertRaisesRegex(ValueError, "mutually exclusive"):
+ pf.col("amount").sum.over(order_by="rowtime", rows=1, range="1
second")
+ with self.assertRaisesRegex(ValueError, "larger than 0"):
+ pf.col("amount").sum.over(order_by="rowtime", rows=0)
+ with self.assertRaisesRegex(TypeError, "two-element"):
+ pf.col("amount").sum.over(order_by="rowtime",
rows=(pf.preceding(1),))
+ with self.assertRaisesRegex(ValueError, "lower bound"):
+ pf.col("amount").sum.over(
+ order_by="rowtime",
+ rows=(pf.following(1), table_expressions.CURRENT_ROW),
+ )
+ with self.assertRaisesRegex(TypeError, "bool"):
+ pf.col("amount").sum.over(order_by="rowtime", rows=True)
+
+ def test_partition_validation_through_public_over_api(self):
+ expression = pf.col("amount").sum.over(
+ order_by="rowtime", partition_by=["category", pf.col("user")]
+ )
+ self.assertIn("category", str(expression))
+ self.assertIn("user", str(expression))
+ with self.assertRaisesRegex(TypeError, "must not be empty"):
+ pf.col("amount").sum.over(order_by="rowtime", partition_by=[])
+ with self.assertRaisesRegex(TypeError, "column name or Expression"):
+ pf.col("amount").sum.over(order_by="rowtime", partition_by=[42])
+
+ def
test_flink_duration_strings_are_accepted_and_sql_interval_is_rejected(self):
+ hour_word = pf.col("amount").sum.over(order_by="rowtime", range="1
hour")
+ hour_abbreviation = pf.col("amount").sum.over(
+ order_by="rowtime", range="1 h"
+ )
+
+ self.assertEqual(str(hour_word), str(hour_abbreviation))
+ with self.assertRaisesRegex(ValueError, "duration string"):
+ pf.col("amount").sum.over(
+ order_by="rowtime", range="INTERVAL '1' HOUR"
+ )
+
+ def test_sub_millisecond_range_is_truncated_to_milliseconds(self):
+ truncated = pf.col("amount").sum.over(
+ order_by="rowtime", range=timedelta(milliseconds=1,
microseconds=500)
+ )
+ whole_millisecond = pf.col("amount").sum.over(
+ order_by="rowtime", range=timedelta(milliseconds=1)
+ )
+
+ self.assertEqual(str(truncated), str(whole_millisecond))
+
+ def test_default_and_unbounded_rows_frames_are_constructed(self):
+ default_frame = pf.col("amount").sum.over(order_by="rowtime")
+ unbounded_rows = pf.col("amount").sum.over(
+ order_by="rowtime", rows=table_expressions.UNBOUNDED_ROW
+ )
+
+ self.assertIn("null, null", str(default_frame))
+ self.assertIn("UNBOUNDED_ROW", str(unbounded_rows))
+ self.assertIn("CURRENT_ROW", str(unbounded_rows))
+
+ def test_lead_and_lag_without_explicit_frames_use_omitted_bounds(self):
+ lag = table_expressions.lag(pf.col("amount")).over(order_by="rowtime")
+ lead =
table_expressions.lead(pf.col("amount")).over(order_by="rowtime")
+
+ self.assertIn("over(lag(amount, 1), rowtime, null, null)", str(lag))
+ self.assertIn("over(lead(amount, 1), rowtime, null, null)", str(lead))
+
+ def test_matching_table_frame_constants_are_accepted(self):
+ cases = (
+ (
+ "rows",
+ (table_expressions.UNBOUNDED_ROW,
table_expressions.CURRENT_ROW),
+ "UNBOUNDED_ROW, CURRENT_ROW",
+ ),
+ (
+ "range",
+ (table_expressions.UNBOUNDED_RANGE,
table_expressions.CURRENT_RANGE),
+ "UNBOUNDED_RANGE, CURRENT_RANGE",
+ ),
+ ("rows", table_expressions.CURRENT_ROW, "CURRENT_ROW,
CURRENT_ROW"),
+ ("range", table_expressions.CURRENT_RANGE, "CURRENT_RANGE,
CURRENT_RANGE"),
+ )
+ for kind, frame, expected in cases:
+ with self.subTest(kind=kind, frame=frame):
+ expression = pf.col("amount").sum.over(
+ order_by="rowtime", **{kind: frame}
+ )
+ self.assertIn("rowtime, " + expected, str(expression))
+
+ def test_mismatched_table_frame_constants_are_rejected(self):
+ cases = (
+ ("rows", table_expressions.CURRENT_RANGE, "CURRENT_RANGE"),
+ ("range", table_expressions.CURRENT_ROW, "CURRENT_ROW"),
+ ("rows", table_expressions.UNBOUNDED_RANGE, "UNBOUNDED_RANGE"),
+ ("range", table_expressions.UNBOUNDED_ROW, "UNBOUNDED_ROW"),
+ )
+ for kind, frame, constant in cases:
+ with self.subTest(kind=kind, constant=constant):
+ with self.assertRaisesRegex(TypeError, "%s.*%s" % (constant,
kind)):
+ pf.col("amount").sum.over(order_by="rowtime", **{kind:
frame})
+
+ def test_over_window_module_introspection_does_not_access_gateway(self):
+ import unittest
+ from unittest.mock import patch
+
+ from pyflink.dataframe import over_window
+ from pyflink.table import _over_window
+
+ with patch(
+ "pyflink.table.expression.get_gateway",
+ side_effect=AssertionError("module introspection accessed the
gateway"),
+ ):
+ for module in (over_window, _over_window):
+ for value in vars(module).values():
+ try:
+ issubclass(value, unittest.TestCase)
+ except TypeError:
+ pass
+
+
+class OverWindowBatchITTests(PyFlinkTestCase):
+ def setUp(self):
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+ self.env = StreamExecutionEnvironment.get_execution_environment()
+ self.env.set_runtime_mode(RuntimeExecutionMode.BATCH)
+ self.t_env = StreamTableEnvironment.create(self.env)
+ pf.set_table_environment(self.t_env)
+
+ def _rowtime_dataframe(self, rows):
+ # Build a bounded source whose schema carries the legacy ROWTIME
indicator.
+ jvm = get_gateway().jvm
+ event_time_type = jvm.org.apache.flink.table.types.AtomicDataType(
+ jvm.org.apache.flink.table.types.logical.TimestampType(
+ False,
jvm.org.apache.flink.table.types.logical.TimestampKind.ROWTIME, 3
+ )
+ ).bridgedTo(jvm.java.sql.Timestamp(0).getClass())
+ row_type = jvm.org.apache.flink.table.api.DataTypes.ROW(
+ to_jarray(
+ jvm.org.apache.flink.table.api.DataTypes.Field,
+ [
+ jvm.org.apache.flink.table.api.DataTypes.FIELD(
+ "amount",
jvm.org.apache.flink.table.api.DataTypes.BIGINT()
+ ),
+ jvm.org.apache.flink.table.api.DataTypes.FIELD(
+ "event_time", event_time_type
+ ),
+ ],
+ )
+ )
+ elements = [
+ row(
+ lit(int(value)),
+ lit(datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S.%f")),
+ )
+ for value, timestamp in (line.split(",") for line in rows)
+ ]
+ j_table = self.t_env._j_tenv.fromValues(
+ row_type, to_expression_jarray(elements)
+ )
+ return pf.from_table(Table(j_table, self.t_env))
+
+ def test_rows_frames_and_global_cumulative_result(self):
+ dataframe = self._rowtime_dataframe(
+ [
+ "10,1970-01-01 00:00:00.000",
+ "20,1970-01-01 00:00:01.000",
+ "5,1970-01-01 00:00:02.000",
+ ],
+ )
+
+ result = dataframe.select(
+ "amount",
+ rows_1=pf.col("amount").sum.over(
+ order_by="event_time", rows=1
+ ),
+ rows_2=pf.col("amount").sum.over(
+ order_by="event_time", rows=2
+ ),
+ global_total=pf.col("amount").sum.over(
+ order_by="event_time", rows=table_expressions.UNBOUNDED_ROW
+ ),
+ )
+
+ self.assertCountEqual(
+ result.collect(),
+ [
+ Row(10, 10, 10, 10),
+ Row(20, 30, 30, 30),
+ Row(5, 25, 35, 35),
+ ],
+ )
+
+ def test_rows_and_range_current_row_handle_peers_differently(self):
+ dataframe = self._rowtime_dataframe(
+ [
+ "10,1970-01-01 00:00:00.000",
+ "20,1970-01-01 00:00:00.000",
+ "5,1970-01-01 00:00:01.000",
+ ],
+ )
+
+ result = dataframe.select(
+ "amount",
+ rows_sum=pf.col("amount").sum.over(
+ order_by="event_time",
+ rows=(table_expressions.CURRENT_ROW,
table_expressions.CURRENT_ROW),
+ ),
+ range_sum=pf.col("amount").sum.over(
+ order_by="event_time",
+ range=(
+ table_expressions.CURRENT_RANGE,
+ table_expressions.CURRENT_RANGE,
+ ),
+ ),
+ )
+
+ self.assertCountEqual(
+ result.collect(),
+ [Row(10, 10, 30), Row(20, 20, 30), Row(5, 5, 5)],
+ )
+
+ def test_range_interval_uses_event_time_distance(self):
+ dataframe = self._rowtime_dataframe(
+ [
+ "10,1970-01-01 00:00:00.000",
+ "20,1970-01-01 00:00:01.000",
+ "5,1970-01-01 00:00:02.000",
+ ],
+ )
+
+ result = dataframe.select(
+ "amount",
+ range_sum=pf.col("amount").sum.over(
+ order_by="event_time", range=timedelta(seconds=1)
+ ),
+ )
+
+ self.assertCountEqual(
+ result.collect(),
+ [Row(10, 10), Row(20, 30), Row(5, 25)],
+ )
+
+
+class OverWindowStreamITTests(PyFlinkTestCase):
+ def setUp(self):
+ previous_environment = pf.get_table_environment()
+ self.addCleanup(pf.set_table_environment, previous_environment)
+ self.env = StreamExecutionEnvironment.get_execution_environment()
+ self.t_env = StreamTableEnvironment.create(self.env)
+ pf.set_table_environment(self.t_env)
+ self.input_path = os.path.join(self.tempdir, "over_events.csv")
+ _register_rowtime_events(
+ self.t_env,
+ self.input_path,
+ [
+ "10,1970-01-01 00:00:00.000",
+ "20,1970-01-01 00:00:01.000",
+ ],
+ )
+
+ def _rowtime_source(self):
+ return pf.from_table(self.t_env.from_path("events"))
+
+ def test_unbounded_rows_over_emits_streaming_results(self):
+ result = self._rowtime_source().select(
+ "amount",
+ running_total=pf.col("amount").sum.over(
+ order_by="event_time", rows=table_expressions.UNBOUNDED_ROW
+ ),
+ )
+
+ self.assertCountEqual(
+ result.collect(),
+ [Row(10, 10), Row(20, 30)],
+ )
+
+ def
test_different_frames_in_one_projection_keep_streaming_restriction(self):
+ dataframe = self._rowtime_source().select(
+ first=pf.col("amount").sum.over(order_by="event_time", rows=1),
+ second=pf.col("amount").sum.over(order_by="event_time", rows=2),
+ )
+
+ with self.assertRaisesRegex(
+ TableException, "All aggregates must be computed on the same
window"
+ ):
+ dataframe.collect()
+
+ def test_following_upper_bound_keeps_streaming_restriction(self):
+ dataframe = self._rowtime_source().select(
+ total=pf.col("amount").sum.over(
+ order_by="event_time",
+ rows=(pf.preceding(1), pf.following(1)),
+ )
+ )
+
+ with self.assertRaisesRegex(
+ TableException, "OVER RANGE FOLLOWING windows are not supported
yet"
+ ):
+ dataframe.collect()
+
+ def test_unpartitioned_over_plan_uses_singleton_distribution(self):
+ dataframe = self._rowtime_source().select(
+ total=pf.col("amount").sum.over(
+ order_by="event_time", rows=table_expressions.UNBOUNDED_ROW
+ )
+ )
+
+ self.assertRegex(
+ dataframe._table.explain().lower(),
+ r"exchange\(distribution=\[single\]\)",
+ )
+
+
+if __name__ == "__main__":
+ import unittest
+
+ unittest.main()
diff --git a/flink-python/pyflink/table/_over_window.py
b/flink-python/pyflink/table/_over_window.py
new file mode 100644
index 00000000000..b30094aaab3
--- /dev/null
+++ b/flink-python/pyflink/table/_over_window.py
@@ -0,0 +1,218 @@
+################################################################################
+# 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.
+################################################################################
+
+"""Internal helpers for constructing expression-level OVER windows."""
+
+import datetime
+from enum import Enum
+from typing import Any, List, Tuple, Union
+
+from pyflink.java_gateway import get_gateway
+from pyflink.table.expression import Expression
+
+
+class _BoundaryDirection(Enum):
+ PRECEDING = "preceding"
+ FOLLOWING = "following"
+
+
+class _FrameBoundary:
+ _direction: _BoundaryDirection
+ _value: Any
+
+
+def _to_column(value: Union[str, Expression], parameter_name: str) ->
Expression:
+ if isinstance(value, str):
+ from pyflink.table.expressions import col
+
+ return col(value)
+ if isinstance(value, Expression):
+ return value
+ raise TypeError("%s must be a column name or Expression" % parameter_name)
+
+
+def _normalize_partition_by(value: Any) -> List[Expression]:
+ if value is None:
+ return []
+ if isinstance(value, (str, Expression)):
+ values = [value]
+ elif isinstance(value, (list, tuple)):
+ values = list(value)
+ else:
+ raise TypeError("partition_by must be a column name or Expression
sequence")
+ if not values:
+ raise TypeError("partition_by must not be empty")
+ return [_to_column(item, "partition_by") for item in values]
+
+
+def _interval_expression(value: Union[str, datetime.timedelta]) -> Expression:
+ if isinstance(value, datetime.timedelta):
+ if value < datetime.timedelta(0):
+ raise ValueError("time interval must not be negative")
+ millis = value // datetime.timedelta(milliseconds=1)
+ elif isinstance(value, str):
+ try:
+ duration =
get_gateway().jvm.org.apache.flink.util.TimeUtils.parseDuration(value)
+ millis = duration.toMillis()
+ except Exception as exc:
+ raise ValueError("range must be a valid duration string") from exc
+ else:
+ raise TypeError("range boundaries must be a datetime.timedelta or
duration string")
+
+ if millis < 0:
+ raise ValueError("time interval must not be negative")
+ j_expr = (
+ get_gateway()
+
.jvm.org.apache.flink.table.expressions.ApiExpressionUtils.intervalOfMillis(millis)
+ )
+ return Expression(j_expr)
+
+
+def _row_expression(value: Any) -> Expression:
+ if isinstance(value, bool):
+ raise TypeError("row interval must not be bool")
+ if not isinstance(value, int):
+ raise TypeError("row boundaries must be integers")
+ if value <= 0:
+ raise ValueError("row interval must be larger than 0")
+ from pyflink.table.expressions import row_interval
+
+ return row_interval(value)
+
+
+def _current_expression(kind: str) -> Expression:
+ # Keep Expression instances out of module globals: introspection can
initialize the gateway.
+ from pyflink.table.expressions import CURRENT_RANGE, CURRENT_ROW
+
+ return CURRENT_ROW if kind == "rows" else CURRENT_RANGE
+
+
+def _value_expression(value: Any, kind: str) -> Expression:
+ return _row_expression(value) if kind == "rows" else
_interval_expression(value)
+
+
+def _normalize_bound(bound: Any, kind: str, position: str) -> Expression:
+ from pyflink.table.expressions import (
+ CURRENT_RANGE,
+ CURRENT_ROW as TABLE_CURRENT_ROW,
+ UNBOUNDED_RANGE,
+ UNBOUNDED_ROW,
+ )
+
+ table_bound_names = (
+ (TABLE_CURRENT_ROW, "CURRENT_ROW", "rows"),
+ (CURRENT_RANGE, "CURRENT_RANGE", "range"),
+ (UNBOUNDED_ROW, "UNBOUNDED_ROW", "rows"),
+ (UNBOUNDED_RANGE, "UNBOUNDED_RANGE", "range"),
+ )
+ for table_bound, table_name, table_kind in table_bound_names:
+ if bound is table_bound:
+ if kind != table_kind:
+ raise TypeError("%s cannot be used with %s frames" %
(table_name, kind))
+ return table_bound
+ if not isinstance(bound, _FrameBoundary):
+ raise TypeError(
+ "%s bound must be preceding(), following(), or a matching Table
API frame constant"
+ % position
+ )
+
+ expected_direction = (
+ _BoundaryDirection.PRECEDING
+ if position == "lower"
+ else _BoundaryDirection.FOLLOWING
+ )
+ if bound._direction is not expected_direction:
+ raise ValueError("invalid %s bound direction" % position)
+ return _value_expression(bound._value, kind)
+
+
+def _validate_explicit_bound_kinds(bounds: Tuple[Any, Any], kind: str) -> None:
+ for bound in bounds:
+ if not isinstance(bound, _FrameBoundary):
+ continue
+ value = bound._value
+ if kind == "rows" and (
+ isinstance(value, bool)
+ or not isinstance(value, int)
+ ):
+ raise ValueError("rows bounds must use the same frame kind")
+ if kind == "range" and isinstance(value, (bool, int)):
+ raise ValueError("range bounds must use the same frame kind")
+
+
+def _normalize_scalar_frame(value: Any, kind: str) -> Tuple[Expression,
Expression]:
+ if isinstance(value, Expression):
+ return _normalize_bound(value, kind, "lower"),
_current_expression(kind)
+ return _value_expression(value, kind), _current_expression(kind)
+
+
+def _normalize_frame(
+ rows: Any, range_: Any
+) -> Tuple[Expression, Expression]:
+ if rows is not None and range_ is not None:
+ raise ValueError("rows and range are mutually exclusive")
+
+ if rows is None and range_ is None:
+ from pyflink.table.expressions import lit
+ from pyflink.table.types import DataTypes
+
+ omitted_bound = lit(None, DataTypes.NULL())
+ return omitted_bound, omitted_bound
+
+ kind = "rows" if rows is not None else "range"
+ value = rows if rows is not None else range_
+ if isinstance(value, tuple):
+ if len(value) != 2:
+ raise TypeError("frame tuple must contain exactly two-element
bounds")
+ lower_bound, upper_bound = value
+ _validate_explicit_bound_kinds((lower_bound, upper_bound), kind)
+ return (
+ _normalize_bound(lower_bound, kind, "lower"),
+ _normalize_bound(upper_bound, kind, "upper"),
+ )
+ return _normalize_scalar_frame(value, kind)
+
+
+def _build_over_expression(
+ aggregate: Expression,
+ order_by: Union[str, Expression],
+ partition_by: Any = None,
+ rows: Any = None,
+ range_: Any = None,
+) -> Expression:
+ if not isinstance(aggregate, Expression):
+ raise TypeError("aggregate must be an Expression")
+ order_expression = _to_column(order_by, "order_by")
+ partitions = _normalize_partition_by(partition_by)
+ preceding_expression, following_expression = _normalize_frame(rows, range_)
+
+ gateway = get_gateway()
+ java_args = [
+ aggregate._j_expr,
+ order_expression._j_expr,
+ preceding_expression._j_expr,
+ following_expression._j_expr,
+ ] + [partition._j_expr for partition in partitions]
+ java_list = gateway.jvm.java.util.ArrayList()
+ for java_arg in java_args:
+ java_list.add(java_arg)
+ unresolved =
gateway.jvm.org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall(
+
gateway.jvm.org.apache.flink.table.functions.BuiltInFunctionDefinitions.OVER,
+ java_list,
+ )
+ return
Expression(gateway.jvm.org.apache.flink.table.api.ApiExpression(unresolved))
diff --git a/flink-python/pyflink/table/expression.py
b/flink-python/pyflink/table/expression.py
index 2b2ff232467..2f45ab0408d 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1597,22 +1597,97 @@ class Expression(Generic[T]):
"""
return _binary_op("repeat")(self, n)
- def over(self, alias) -> 'Expression':
- """
- Defines an aggregation to be used for a previously specified over
window.
-
- Example:
- ::
-
- >>> tab.window(Over
- >>> .partition_by(col('c'))
- >>> .order_by(col('rowtime'))
- >>> .preceding(row_interval(2))
- >>> .following(CURRENT_ROW)
- >>> .alias("w")) \\
- >>> .select(col('c'), col('a'), col('a').count.over(col('w')))
- """
- return _binary_op("over")(self, alias)
+ def over(
+ self,
+ alias=None,
+ *,
+ order_by=None,
+ partition_by=None,
+ rows=None,
+ range=None,
+ ) -> 'Expression':
+ """
+ Defines an aggregation over a named Table API window or an inline
DataFrame window.
+
+ A positional ``alias`` or ``alias=`` refers to a named Table API
window. Inline DataFrame
+ windows require ``order_by=`` and can use ``rows`` or ``range``.
Matching Table API frame
+ constants are also accepted (``*_ROW`` with ``rows``, ``*_RANGE`` with
``range``).
+
+ A scalar ``rows`` value is a positive integer describing the number of
preceding rows. A
+ scalar ``range`` value is a non-negative :class:`datetime.timedelta`
or Flink duration
+ string describing the preceding time interval. Both frames can instead
be specified as a
+ two-element ``(lower, upper)`` tuple. Explicit bounds are created with
+ :func:`~pyflink.dataframe.over_window.preceding` and
+ :func:`~pyflink.dataframe.over_window.following`, or with matching
constants from
+ :mod:`pyflink.table.expressions`.
+
+ :param alias: Expression identifying a previously declared Table API
OVER window. It
+ cannot be combined with inline window arguments.
+ :type alias: Optional[Expression]
+ :param order_by: Column name or expression identifying the single
ordering time attribute
+ of an inline window.
+ :type order_by: Optional[Union[str, Expression]]
+ :param partition_by: Optional column name, expression, or non-empty
list or tuple of column
+ names and expressions used as partition keys.
+ :type partition_by: Optional[Union[str, Expression, List[Union[str,
Expression]],
+ Tuple[Union[str, Expression], ...]]]
+ :param rows: Optional positive row count or two-element ROW-frame
bound tuple.
+ :type rows: Optional[Union[int, Expression, Tuple[Union[FrameBoundary,
Expression],
+ Union[FrameBoundary, Expression]]]]
+ :param range: Optional Flink duration string,
:class:`datetime.timedelta`, or two-element
+ RANGE-frame bound tuple.
+ :type range: Optional[Union[str, datetime.timedelta, Expression,
+ Tuple[Union[FrameBoundary, Expression],
+ Union[FrameBoundary, Expression]]]]
+ :return: An OVER aggregation expression.
+ :raises TypeError: If named- and inline-window arguments are combined,
a required
+ ``order_by`` is missing, or an argument has an
unsupported type.
+ :raises ValueError: If ``rows`` and ``range`` are both provided, or a
frame bound has an
+ invalid value, direction, or frame kind.
+
+ Given a DataFrame ``df`` containing ``key``, ``amount``, and
``rowtime``::
+
+ >>> import pyflink.dataframe as pf
+ >>> from pyflink.table.expressions import CURRENT_ROW,
UNBOUNDED_ROW
+ >>> result = df.select(
+ ... "key",
+ ... running=pf.col("amount").sum.over(
+ ... order_by="rowtime", partition_by="key",
+ ... rows=(pf.preceding(2), CURRENT_ROW)))
+
+ Table API frame constants can also define the bounds::
+
+ >>> cumulative = pf.col("amount").sum.over(
+ ... order_by="rowtime", rows=(UNBOUNDED_ROW, CURRENT_ROW))
+
+ For an existing named Table API window ``w``::
+
+ >>> pf.col("amount").sum.over(alias=pf.col("w"))
+
+ .. versionchanged:: 2.4.0
+ Added inline DataFrame window arguments while preserving named
Table API windows.
+ """
+ if alias is not None and order_by is not None:
+ raise TypeError("over() cannot specify both alias and order_by")
+ if alias is not None and (
+ partition_by is not None or rows is not None or range is not None
+ ):
+ raise TypeError("over() cannot combine alias with inline window
arguments")
+ if alias is not None:
+ return _binary_op("over")(self, alias)
+
+ if order_by is None:
+ raise TypeError("over() missing required argument: order_by")
+
+ from pyflink.table._over_window import _build_over_expression
+
+ return _build_over_expression(
+ self,
+ order_by,
+ partition_by=partition_by,
+ rows=rows,
+ range_=range,
+ )
@property
def reverse(self) -> 'Expression[str]':
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
index ea7fe0dfe66..5f78a0ec52f 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/OverWindowResolverRule.java
@@ -66,6 +66,34 @@ final class OverWindowResolverRule implements ResolverRule {
if (unresolvedCall.getFunctionDefinition() ==
BuiltInFunctionDefinitions.OVER) {
List<Expression> children = unresolvedCall.getChildren();
+
+ if (children.size() >= 4) {
+ UnresolvedCallExpression agg = (UnresolvedCallExpression)
children.get(0);
+ List<Expression> expandedChildren = new
ArrayList<>(children);
+ Expression preceding = children.get(2);
+ Expression following = children.get(3);
+
+ if (isLeadOrLag(agg)) {
+ if (!isNullLiteral(preceding) ||
!isNullLiteral(following)) {
+ throw leadLagWithBoundsException();
+ }
+ } else if (isNullLiteral(preceding) &&
isNullLiteral(following)) {
+ expandedChildren.set(2,
valueLiteral(OverWindowRange.UNBOUNDED_RANGE));
+ expandedChildren.set(3,
valueLiteral(OverWindowRange.CURRENT_RANGE));
+ }
+
+ return unresolvedCall.replaceArgs(
+ expandedChildren.stream()
+ .map(expr -> expr.accept(this))
+ .collect(Collectors.toList()));
+ }
+
+ if (children.size() != 2) {
+ throw new ValidationException(
+ "An OVER call expects a window alias or an inline
window "
+ + "specification.");
+ }
+
Expression alias = children.get(1);
LocalOverWindow referenceWindow =
@@ -81,13 +109,10 @@ final class OverWindowResolverRule implements ResolverRule
{
final List<Expression> newArgs = new ArrayList<>();
newArgs.add(agg);
newArgs.add(referenceWindow.getOrderBy());
- if (agg.getFunctionDefinition() ==
BuiltInFunctionDefinitions.LAG
- || agg.getFunctionDefinition() ==
BuiltInFunctionDefinitions.LEAD) {
+ if (isLeadOrLag(agg)) {
if (referenceWindow.getPreceding().isPresent()
|| referenceWindow.getFollowing().isPresent()) {
- throw new ValidationException(
- "LEAD/LAG functions do not support "
- + "providing RANGE/ROW bounds.");
+ throw leadLagWithBoundsException();
}
newArgs.add(valueLiteral(null, DataTypes.NULL()));
newArgs.add(valueLiteral(null, DataTypes.NULL()));
@@ -111,6 +136,21 @@ final class OverWindowResolverRule implements ResolverRule
{
}
}
+ private static boolean isLeadOrLag(UnresolvedCallExpression aggregate)
{
+ return aggregate.getFunctionDefinition() ==
BuiltInFunctionDefinitions.LAG
+ || aggregate.getFunctionDefinition() ==
BuiltInFunctionDefinitions.LEAD;
+ }
+
+ private static boolean isNullLiteral(Expression expression) {
+ return expression instanceof ValueLiteralExpression
+ && ((ValueLiteralExpression) expression).isNull();
+ }
+
+ private static ValidationException leadLagWithBoundsException() {
+ return new ValidationException(
+ "LEAD/LAG functions do not support providing RANGE/ROW
bounds.");
+ }
+
private Expression calculateOverWindowFollowing(
LocalOverWindow referenceWindow, Expression preceding) {
return referenceWindow
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationExpressionsUtils.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationExpressionsUtils.java
index 81b4daf380c..c71205b68f6 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationExpressionsUtils.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationExpressionsUtils.java
@@ -167,6 +167,8 @@ public class OperationExpressionsUtils {
aggregates.computeIfAbsent(unresolvedCall, expr -> "EXPR$" +
uniqueId++);
} else if (WINDOW_PROPERTIES.contains(functionDefinition)) {
properties.computeIfAbsent(unresolvedCall, expr -> "EXPR$" +
uniqueId++);
+ } else if (functionDefinition == BuiltInFunctionDefinitions.OVER) {
+ // OVER aggregates are evaluated by window processing, not as
regular aggregates.
} else {
unresolvedCall.getChildren().forEach(c -> c.accept(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 290999be8e7..28978469b6e 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
@@ -1018,6 +1018,9 @@ public final class OperationTreeBuilder {
@Override
public Void visit(UnresolvedCallExpression call) {
+ if (call.getFunctionDefinition() ==
BuiltInFunctionDefinitions.OVER) {
+ return null;
+ }
if (isFunctionOfKind(call, FunctionKind.AGGREGATE)) {
throw new ValidationException(exceptionMessage);
}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/OverWindowExpressionResolverTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/OverWindowExpressionResolverTest.java
new file mode 100644
index 00000000000..7be0da28cd7
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/resolver/OverWindowExpressionResolverTest.java
@@ -0,0 +1,237 @@
+/*
+ * 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.expressions.resolver;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Over;
+import org.apache.flink.table.api.OverWindow;
+import org.apache.flink.table.api.OverWindowRange;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.legacy.api.TableSchema;
+import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.SourceQueryOperation;
+import org.apache.flink.table.types.AtomicDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.TimestampKind;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
+import org.apache.flink.table.utils.FunctionLookupMock;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.lit;
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class OverWindowExpressionResolverTest {
+
+ @Test
+ void resolvesFieldsInExpandedOverExpression() {
+ Expression expandedOver =
+ unresolvedCall(
+ BuiltInFunctionDefinitions.OVER,
+ Arrays.asList(
+ unresolvedCall(
+ BuiltInFunctionDefinitions.SUM,
unresolvedRef("amount")),
+ unresolvedRef("rowtime"),
+ valueLiteral(2L),
+ valueLiteral(OverWindowRange.CURRENT_ROW),
+ unresolvedRef("tenant")));
+
+ CallExpression resolved =
+ (CallExpression)
+
resolver(Collections.emptyList()).resolve(List.of(expandedOver)).get(0);
+ List<Expression> children = resolved.getChildren();
+
+ assertThat(children).hasSize(5);
+ assertThat(((CallExpression) children.get(0)).getChildren().get(0))
+ .isInstanceOf(FieldReferenceExpression.class);
+ assertField(children.get(1), "rowtime");
+ assertThat(children.get(2).asSummaryString()).isEqualTo("2");
+ assertThat(children.get(3).asSummaryString()).contains("CURRENT_ROW");
+ assertField(children.get(4), "tenant");
+ }
+
+ @Test
+ void appliesDefaultRangeFrameToExpandedAggregate() {
+ Expression expandedOver =
+ expandedOver(
+ BuiltInFunctionDefinitions.SUM,
+ valueLiteral(null, DataTypes.NULL()),
+ valueLiteral(null, DataTypes.NULL()));
+
+ CallExpression resolved =
+ (CallExpression)
+
resolver(Collections.emptyList()).resolve(List.of(expandedOver)).get(0);
+ List<Expression> children = resolved.getChildren();
+
+
assertThat(children.get(2).asSummaryString()).contains("UNBOUNDED_RANGE");
+
assertThat(children.get(3).asSummaryString()).contains("CURRENT_RANGE");
+ }
+
+ @Test
+ void preservesOmittedBoundsForExpandedLeadAndLag() {
+ assertOmittedBounds(BuiltInFunctionDefinitions.LAG);
+ assertOmittedBounds(BuiltInFunctionDefinitions.LEAD);
+ }
+
+ @Test
+ void rejectsExplicitBoundsForExpandedLeadAndLag() {
+ assertExplicitBoundsRejected(BuiltInFunctionDefinitions.LAG);
+ assertExplicitBoundsRejected(BuiltInFunctionDefinitions.LEAD);
+ }
+
+ @Test
+ void expandsLegacyNamedOverWindowAlias() {
+ OverWindow window =
+
Over.partitionBy($("tenant")).orderBy($("rowtime")).preceding(lit(2L)).as("w");
+ Expression overAlias =
+ unresolvedCall(
+ BuiltInFunctionDefinitions.OVER,
+ unresolvedCall(BuiltInFunctionDefinitions.SUM,
unresolvedRef("amount")),
+ unresolvedRef("w"));
+
+ CallExpression resolved =
+ (CallExpression)
resolver(List.of(window)).resolve(List.of(overAlias)).get(0);
+ List<Expression> children = resolved.getChildren();
+
+ assertThat(children).hasSize(5);
+ assertThat(((CallExpression) children.get(0)).getChildren().get(0))
+ .isInstanceOf(FieldReferenceExpression.class);
+ assertField(children.get(1), "rowtime");
+ assertThat(children.get(2).asSummaryString()).isEqualTo("2");
+ assertThat(children.get(3).asSummaryString()).contains("CURRENT_ROW");
+ assertField(children.get(4), "tenant");
+ }
+
+ @Test
+ void rejectsMalformedOverExpression() {
+ Expression malformedOver =
+ unresolvedCall(
+ BuiltInFunctionDefinitions.OVER,
+ Arrays.asList(
+ unresolvedCall(
+ BuiltInFunctionDefinitions.SUM,
unresolvedRef("amount")),
+ unresolvedRef("rowtime"),
+ valueLiteral(2L)));
+
+ assertThatThrownBy(() ->
resolver(Collections.emptyList()).resolve(List.of(malformedOver)))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ private static Expression expandedOver(
+ FunctionDefinition aggregate, Expression preceding, Expression
following) {
+ return unresolvedCall(
+ BuiltInFunctionDefinitions.OVER,
+ Arrays.asList(
+ unresolvedCall(aggregate, unresolvedRef("amount")),
+ unresolvedRef("rowtime"),
+ preceding,
+ following));
+ }
+
+ private static void assertOmittedBounds(FunctionDefinition aggregate) {
+ CallExpression resolved =
+ (CallExpression)
+ resolver(Collections.emptyList())
+ .resolve(
+ List.of(
+ expandedOver(
+ aggregate,
+ valueLiteral(null,
DataTypes.NULL()),
+ valueLiteral(null,
DataTypes.NULL()))))
+ .get(0);
+
+ assertThat(resolved.getChildren().subList(2, 4))
+ .allSatisfy(
+ bound -> {
+
assertThat(bound).isInstanceOf(ValueLiteralExpression.class);
+ assertThat(((ValueLiteralExpression)
bound).isNull()).isTrue();
+ });
+ }
+
+ private static void assertExplicitBoundsRejected(FunctionDefinition
aggregate) {
+ Expression expandedOver =
+ expandedOver(
+ aggregate,
+ valueLiteral(OverWindowRange.UNBOUNDED_RANGE),
+ valueLiteral(OverWindowRange.CURRENT_RANGE));
+
+ assertThatThrownBy(() ->
resolver(Collections.emptyList()).resolve(List.of(expandedOver)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("LEAD/LAG functions do not support");
+ }
+
+ private static void assertField(Expression expression, String name) {
+ assertThat(expression).isInstanceOf(FieldReferenceExpression.class);
+ assertThat(((FieldReferenceExpression)
expression).getName()).isEqualTo(name);
+ }
+
+ private static ExpressionResolver resolver(List<OverWindow> windows) {
+ DataType rowtime =
+ new AtomicDataType(new LocalZonedTimestampType(false,
TimestampKind.ROWTIME, 3));
+ TableSchema schema =
+ TableSchema.builder()
+ .field("amount", DataTypes.BIGINT())
+ .field("rowtime", rowtime)
+ .field("tenant", DataTypes.STRING())
+ .build();
+ QueryOperation input =
+ new SourceQueryOperation(
+ ContextResolvedTable.anonymous(
+ new ResolvedCatalogTable(
+
CatalogTable.newBuilder().schema(schema.toSchema()).build(),
+ ResolvedSchema.physical(
+ schema.getFieldNames(),
+ schema.getFieldDataTypes()))));
+
+ return ExpressionResolver.resolverFor(
+ TableConfig.getDefault(),
+ Thread.currentThread().getContextClassLoader(),
+ name -> Optional.empty(),
+ new FunctionLookupMock(Collections.emptyMap()),
+ new DataTypeFactoryMock(),
+ (sqlExpression, inputRowType, outputType) -> {
+ throw new UnsupportedOperationException();
+ },
+ input)
+ .withOverWindows(windows)
+ .build();
+ }
+}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/utils/OperationExpressionsUtilsTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/utils/OperationExpressionsUtilsTest.java
new file mode 100644
index 00000000000..4ddeefdf9f3
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/utils/OperationExpressionsUtilsTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.table.api.OverWindowRange;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.UnresolvedCallExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
+import static
org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class OperationExpressionsUtilsTest {
+
+ @Test
+ void keepsAggregatesInsideExpandedOverExpressionsInProjection() {
+ Expression overExpression =
+ unresolvedCall(
+ BuiltInFunctionDefinitions.OVER,
+ Arrays.asList(
+ unresolvedCall(
+ BuiltInFunctionDefinitions.SUM,
unresolvedRef("amount")),
+ unresolvedRef("event_time"),
+ valueLiteral(1L),
+ valueLiteral(OverWindowRange.CURRENT_ROW)));
+
+ OperationExpressionsUtils.CategorizedExpressions categorized =
+
OperationExpressionsUtils.extractAggregationsAndProperties(List.of(overExpression));
+ UnresolvedCallExpression projectedOver =
+ (UnresolvedCallExpression) categorized.getProjections().get(0);
+
+ assertThat(categorized.getAggregations()).isEmpty();
+ assertThat(projectedOver.getFunctionDefinition())
+ .isEqualTo(BuiltInFunctionDefinitions.OVER);
+
assertThat(projectedOver.getChildren().get(0)).isInstanceOf(UnresolvedCallExpression.class);
+ assertThat(
+ ((UnresolvedCallExpression)
projectedOver.getChildren().get(0))
+ .getFunctionDefinition())
+ .isEqualTo(BuiltInFunctionDefinitions.SUM);
+ }
+}
