Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3574#discussion_r106955390
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
---
@@ -106,6 +112,10 @@ class DataStreamOverAggregate(
if (overWindow.lowerBound.isUnbounded &&
overWindow.upperBound.isCurrentRow) {
createUnboundedAndCurrentRowProcessingTimeOverWindow(inputDS)
+ } // lowerbound is a RowType and upperbound is PRECEEDING or
CURRENT ROW
+ else if (overWindow.isRows &&
+ (overWindow.upperBound.isPreceding ||
overWindow.upperBound.isCurrentRow)) {
--- End diff --
I think the condition should be `overWindow.lowerBound.isPreceding &&
!overWindow.lowerBound.isUnbounded && overWindow.upperBound.isCurrentRow`.
This will check that the lower bound in bounded and preceding and that the
upper bound is the current row. The window implementation addresses exactly
this case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
