snuyanzin commented on code in PR #26157:
URL: https://github.com/apache/flink/pull/26157#discussion_r1955222196
##########
flink-table/flink-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) {
+ throw new TableException(
+ "Only positive step and size constant for CUMULATE window
descriptors are supported.")
+ }
+ if (maxSize % step != 0) {
+ throw new TableException(
+ "CUMULATE window requires size being an integral multiple of
step.")
+ }
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(
Review Comment:
Otherwise
https://github.com/apache/flink/blob/cd0cfe42c15042a3c27a0c341416aa32c0d594b6/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/groupwindow/assigners/SessionWindowAssigner.java#L47-L50
--
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]
