[FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime #26157

snuyanzin · 2025-02-13T20:45:50Z

What is the purpose of the change

Queries with wrongly defined descriptors should fail during planning, like

SELECT *
FROM TABLE(
 CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '0' MINUTE, INTERVAL '0' HOUR));

SELECT *
FROM TABLE(
 HOP(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '0' MINUTE, INTERVAL '0' MINUTE));

SELECT *
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '0' MINUTE));

SELECT *
FROM TABLE(SESSION(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '0' MINUTE));

Brief change log

WindowUtil

Verifying this change

WindowTableFunctionTest.scala

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): ( no)
The public API, i.e., is any changed class annotated with @Public(Evolving): ( no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): ( no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: ( no)

Documentation

Does this pull request introduce a new feature? ( no)
If yes, how is the feature documented? (not applicable)

…ng planning rather than runtime

snuyanzin · 2025-02-13T20:54:50Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

@@ -217,6 +217,10 @@ object WindowUtil {
          null
        }
        val interval = getOperandAsLong(windowCall.operands(2))
+        if (interval <= 0) {
+          throw new TableException(


since TableException is already used for getOperandAsLong

flink/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

Lines 390 to 394 in cd0cfe4

case _: RexLiteral =>

throw new TableException(

"Window aggregate only support SECOND, MINUTE, HOUR, DAY as the time unit. " +

"MONTH and YEAR time unit are not supported yet.")

case _ => throw new TableException("Only constant window descriptors are supported.")

snuyanzin · 2025-02-13T20:58:09Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

@@ -217,6 +217,10 @@ object WindowUtil {
          null
        }
        val interval = getOperandAsLong(windowCall.operands(2))
+        if (interval <= 0) {


otherwise will fail at runtime at

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/assigners/TumblingWindowAssigner.java

Lines 50 to 53 in cd0cfe4

if (size <= 0) {

throw new IllegalArgumentException(

"TumblingWindowAssigner parameters must satisfy size > 0");

}

snuyanzin · 2025-02-13T21:00:55Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

@@ -227,6 +231,10 @@ object WindowUtil {
        }
        val slide = getOperandAsLong(windowCall.operands(2))
        val size = getOperandAsLong(windowCall.operands(3))
+        if (slide <= 0 || size <= 0) {


Otherwise

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/assigners/SlidingWindowAssigner.java

Lines 58 to 61 in cd0cfe4

if (size <= 0 || slide <= 0) {

throw new IllegalArgumentException(

"SlidingWindowAssigner parameters must satisfy slide > 0 and size > 0");

}

snuyanzin · 2025-02-13T21:13:50Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

@@ -237,13 +245,25 @@ object WindowUtil {
        }
        val step = getOperandAsLong(windowCall.operands(2))
        val maxSize = getOperandAsLong(windowCall.operands(3))
+        if (step <= 0 || maxSize <= 0) {


Otherwise

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/assigners/CumulativeWindowAssigner.java

Lines 53 to 60 in cd0cfe4

if (maxSize <= 0 || step <= 0) {

throw new IllegalArgumentException(

"CumulativeWindowAssigner parameters must satisfy step > 0 and size > 0");

}

if (maxSize % step != 0) {

throw new IllegalArgumentException(

"CumulativeWindowAssigner requires size must be an integral multiple of step.");

}

snuyanzin · 2025-02-13T21:14:22Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

        new CumulativeWindowSpec(Duration.ofMillis(maxSize), Duration.ofMillis(step), offset)
      case FlinkSqlOperatorTable.SESSION =>
        val tableArgCall = windowCall.operands(0).asInstanceOf[RexTableArgCall]
        if (!tableArgCall.getOrderKeys.isEmpty) {
          throw new ValidationException("Session window TVF doesn't support order by clause.")
        }
        val gap = getOperandAsLong(windowCall.operands(2))
+        if (gap <= 0) {
+          throw new TableException(


Otherwise

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/assigners/SessionWindowAssigner.java

Lines 47 to 50 in cd0cfe4

if (sessionGap <= 0) {

throw new IllegalArgumentException(

"SessionWindowAssigner parameters must satisfy 0 < size");

}

flinkbot · 2025-02-13T21:24:54Z

CI report:

9637446 UNKNOWN
9f5f286 Azure: SUCCESS
79da1de Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

snuyanzin · 2025-02-13T22:51:26Z

.../src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala

@@ -1120,7 +1120,7 @@ class WindowAggregateTest(aggPhaseEnforcer: AggregatePhaseStrategy) extends Tabl
        |   b,
        |   count(distinct c) AS uv
        |FROM TABLE(
-        |   CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '25' MINUTE, INTERVAL '1' HOUR))
+        |   CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '20' MINUTE, INTERVAL '1' HOUR))


looks like was incorrect test which didn't fail because it tested only planning

davidradl · 2025-02-14T14:42:54Z

...link-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/WindowUtil.scala

@@ -227,6 +232,11 @@ object WindowUtil {
        }
        val slide = getOperandAsLong(windowCall.operands(2))
        val size = getOperandAsLong(windowCall.operands(3))
+        if (slide <= 0 || size <= 0) {
+          throw new TableException(
+            s"HOP table function based aggregate requires slide and size being positive," +


nit: being -> to be.
nit2: Just a thought we could have a separate message for each invalid value.

nit: being -> to be.

thanks, fixed

nit2: Just a thought we could have a separate message for each invalid value.

I have a different opinion about that

since it will lead to the fact that once one value is fixed there will be another issue.

i would prefer to have similar approach as in runtime to avoid some missed cases

snuyanzin changed the title ~~[FLINK37321][table] Make window functions with descriptors fail during planning rather than runtime~~ [FLINK-37321][table] Make window functions with descriptors fail during planning rather than runtime Feb 13, 2025

[FLINK-37321][table] Make window functions with descriptors fail duri…

9637446

…ng planning rather than runtime

snuyanzin force-pushed the flink37321 branch from 5193a9b to 9637446 Compare February 13, 2025 20:46

snuyanzin changed the title ~~[FLINK-37321][table] Make window functions with descriptors fail during planning rather than runtime~~ [FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime Feb 13, 2025

snuyanzin commented Feb 13, 2025

View reviewed changes

One more rule

a7ee98a

snuyanzin commented Feb 13, 2025

View reviewed changes

fixup

3d3b3f0

snuyanzin commented Feb 13, 2025

View reviewed changes

flinkbot added the component=TableSQL/Planner label Feb 13, 2025

davidradl reviewed Feb 14, 2025

View reviewed changes

Address feedback

9f5f286

snuyanzin requested a review from davidradl February 14, 2025 16:46

Address feedback

79da1de

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime #26157

[FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime #26157

snuyanzin commented Feb 13, 2025

snuyanzin Feb 13, 2025

snuyanzin Feb 13, 2025 •

edited

Loading

snuyanzin Feb 13, 2025

snuyanzin Feb 13, 2025

snuyanzin Feb 13, 2025

flinkbot commented Feb 13, 2025 •

edited

Loading

snuyanzin Feb 13, 2025

davidradl Feb 14, 2025

snuyanzin Feb 14, 2025 •

edited

Loading

	case _: RexLiteral =>
	throw new TableException(
	"Window aggregate only support SECOND, MINUTE, HOUR, DAY as the time unit. " +
	"MONTH and YEAR time unit are not supported yet.")
	case _ => throw new TableException("Only constant window descriptors are supported.")

	if (size <= 0) {
	throw new IllegalArgumentException(
	"TumblingWindowAssigner parameters must satisfy size > 0");
	}

	if (size <= 0 \|\| slide <= 0) {
	throw new IllegalArgumentException(
	"SlidingWindowAssigner parameters must satisfy slide > 0 and size > 0");
	}

	if (maxSize <= 0 \|\| step <= 0) {
	throw new IllegalArgumentException(
	"CumulativeWindowAssigner parameters must satisfy step > 0 and size > 0");
	}
	if (maxSize % step != 0) {
	throw new IllegalArgumentException(
	"CumulativeWindowAssigner requires size must be an integral multiple of step.");
	}

	if (sessionGap <= 0) {
	throw new IllegalArgumentException(
	"SessionWindowAssigner parameters must satisfy 0 < size");
	}

[FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime #26157

Are you sure you want to change the base?

[FLINK-37321][table] Make window functions with wrong descriptors fail during planning rather than runtime #26157

Conversation

snuyanzin commented Feb 13, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

snuyanzin Feb 13, 2025

Choose a reason for hiding this comment

snuyanzin Feb 13, 2025 • edited Loading

Choose a reason for hiding this comment

snuyanzin Feb 13, 2025

Choose a reason for hiding this comment

snuyanzin Feb 13, 2025

Choose a reason for hiding this comment

snuyanzin Feb 13, 2025

Choose a reason for hiding this comment

flinkbot commented Feb 13, 2025 • edited Loading

CI report:

snuyanzin Feb 13, 2025

Choose a reason for hiding this comment

davidradl Feb 14, 2025

Choose a reason for hiding this comment

snuyanzin Feb 14, 2025 • edited Loading

Choose a reason for hiding this comment

snuyanzin Feb 13, 2025 •

edited

Loading

flinkbot commented Feb 13, 2025 •

edited

Loading

snuyanzin Feb 14, 2025 •

edited

Loading