Timm0 commented on code in PR #29070:
URL: https://github.com/apache/flink/pull/29070#discussion_r3988190112
##########
flink-python/pyflink/dataframe/tests/test_dataframe.py:
##########
@@ -2176,5 +2330,141 @@ def
test_grouped_aggregation_with_batch_table_environment(self):
)
+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",
+
proc_time=pf.col("proc_time").cast(TableDataTypes.TIMESTAMP(3)),
Review Comment:
Good catch! I tested your proposed materialization (`window_time ==
window_end - 1ms`), but it's also not deterministic. On a `proctime` window
`window_time` is itself a `proctime` attribute, so materializing it re-samples
the wall clock at read time. So both paths hit the same problem.
Therefore I dropped the per-row time assertion. I was taking a look at other
existing tests and noticed that this is consistent with existing `proctime`
execution tests. E.g. `GroupWindowITCase.testProctimeCascadeWindowAgg` and
`WindowAggregateITCase.testRelaxFormProctimeCascadeWindowAgg` both even have
inline comments mentioning the missing assertions due to determinism.
Exact-value assertions only exist at the operator level with a manual clock
(e.g. `AlignedWindowTableFunctionOperatorTest`), which isn't feasible in our
case.
Please let me know if this is fine for you or if there is anything else I
should consider or test here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
