[FLINK]Support processing time tumble window for nexmark q12

velox/exec/StreamingAggregation.cpp

@@ -210,10 +210,9 @@ void StreamingAggregation::assignGroups() {
   auto numInput = input_->size();
 
   inputGroups_.resize(numInput);
-
   // Look for the end of the last group.
   vector_size_t index = 0;
-  if (prevInput_) {
+  if (prevInput_ && numGroups_ > 0) {
     auto prevIndex = prevInput_->size() - 1;
     auto* prevGroup = groups_[numGroups_ - 1];
     for (; index < numInput; ++index) {
@@ -334,10 +333,9 @@ RowVectorPtr StreamingAggregation::getOutput() {
   evaluateAggregates();
 
   RowVectorPtr output;
-
-  if (numGroups_ > minOutputBatchSize_) {
+  if (numGroups_ >= minOutputBatchSize_) {
     output = createOutput(
-        std::min(numGroups_ - 1, static_cast<size_t>(maxOutputBatchSize_)));
+        std::min(numGroups_, static_cast<size_t>(maxOutputBatchSize_)));
   }
 
   prevInput_ = input_;

velox/experimental/stateful/GroupWindowAggregator.cpp

@@ -24,7 +24,7 @@ GroupWindowAggregator::GroupWindowAggregator(
     std::vector<std::unique_ptr<StatefulOperator>> targets,
     std::unique_ptr<KeySelector> keySelector,
     std::unique_ptr<SliceAssigner> sliceAssigner,
-    long allowedLateness,
+    int64_t allowedLateness,
     bool produceUpdates,
     int rowtimeIndex,
     bool isEventTime,
@@ -82,12 +82,12 @@ void GroupWindowAggregator::getOutput() {
   }
 
   // 1. Partition input by key
-  std::map<uint32_t, RowVectorPtr> keyToData = keySelector_->partition(input_);
+  std::map<int64_t, RowVectorPtr> keyToData = keySelector_->partition(input_);
   for (const auto& [key, keyedData] : keyToData) {
     // 2. Set the current key in the context
     windowContext_->setCurrentKey(key);
     // 3. Partition the keyed data by rowtime or processing time
-    std::map<uint32_t, RowVectorPtr> timestampToData = sliceAssigner_->assignSliceEnd(keyedData);
+    std::map<int64_t, RowVectorPtr> timestampToData = sliceAssigner_->assignSliceEnd(keyedData);
     for (const auto& [timestamp, data] : timestampToData) {
       // 4. Assign data to window
       std::vector<TimeWindow> windows = 
@@ -143,11 +143,11 @@ void GroupWindowAggregator::close() {
 }
 
 void GroupWindowAggregator::registerCleanupTimer(uint32_t key, TimeWindow window) {
-  long cleanupTime =
+  int64_t cleanupTime =
       TimeWindowUtil::toEpochMillsForTimer(
           TimeWindowUtil::cleanupTime(window.maxTimestamp(), allowedLateness_, isEventTime_),
           shiftTimeZone_);
-  if (cleanupTime == LONG_MAX) {
+  if (cleanupTime == INT64_MAX) {
     // don't set a GC timer for "end of time"
     return;
   }
@@ -190,17 +190,17 @@ void GroupWindowAggregator::WindowTriggerContext::open() {
 }
 
 bool GroupWindowAggregator::WindowTriggerContext::onElement(
-    uint32_t key, RowVectorPtr row, long timestamp, TimeWindow window) {
+    uint32_t key, RowVectorPtr row, int64_t timestamp, TimeWindow window) {
   return trigger_->onElement(key, row, timestamp, window);
 }
 
 bool GroupWindowAggregator::WindowTriggerContext::onProcessingTime(
-    TimeWindow window, long time) {
+    TimeWindow window, int64_t time) {
   return trigger_->onProcessingTime(window, time);
 }
 
 bool GroupWindowAggregator::WindowTriggerContext::onEventTime(
-    TimeWindow window, long time) {
+    TimeWindow window, int64_t time) {
   return trigger_->onEventTime(window, time);
 }
 
@@ -209,31 +209,31 @@ void GroupWindowAggregator::WindowTriggerContext::onMerge(
   trigger_->onMerge(key, window, nullptr);
 }
 
-long GroupWindowAggregator::WindowTriggerContext::getCurrentProcessingTime() {
+int64_t GroupWindowAggregator::WindowTriggerContext::getCurrentProcessingTime() {
   return internalTimerService_->currentProcessingTime();
 }
 
-long GroupWindowAggregator::WindowTriggerContext::getCurrentWatermark() {
+int64_t GroupWindowAggregator::WindowTriggerContext::getCurrentWatermark() {
   return internalTimerService_->currentWatermark();
 }
 
 void GroupWindowAggregator::WindowTriggerContext::registerProcessingTimeTimer(
-    uint32_t key, TimeWindow window, long time) {
+    uint32_t key, TimeWindow window, int64_t time) {
   internalTimerService_->registerProcessingTimeTimer(key, window, time);
 }
 
 void GroupWindowAggregator::WindowTriggerContext::registerEventTimeTimer(
-    uint32_t key, TimeWindow window, long time) {
+    uint32_t key, TimeWindow window, int64_t time) {
   internalTimerService_->registerEventTimeTimer(key, window, time);
 }
 
 void GroupWindowAggregator::WindowTriggerContext::deleteProcessingTimeTimer(
-    uint32_t key, TimeWindow window, long time) {
+    uint32_t key, TimeWindow window, int64_t time) {
   internalTimerService_->deleteProcessingTimeTimer(key, window, time);
 }
 
 void GroupWindowAggregator::WindowTriggerContext::deleteEventTimeTimer(
-    uint32_t key, TimeWindow window, long time) {
+    uint32_t key, TimeWindow window, int64_t time) {
   internalTimerService_->deleteEventTimeTimer(key, window, time);
 }
 
@@ -265,7 +265,7 @@ WindowContext::WindowContext(
     std::shared_ptr<StreamOperatorStateHandler> stateHandler,
     int shiftTimeZone,
     bool isEventTime,
-    long allowedLateness)
+    int64_t allowedLateness)
     : windowAggregator_(windowAggregator),
       windowState_(std::move(windowState)),
       timerService_(std::move(timerService)),
@@ -280,11 +280,11 @@ StatePtr WindowContext::getPartitionedState(StateDescriptor& stateDescriptor) {
   return stateHandler_->getGroupMapState(stateDescriptor);
 }
 
-long WindowContext::currentProcessingTime() {
+int64_t WindowContext::currentProcessingTime() {
   return timerService_->currentProcessingTime();
 }
 
-long WindowContext::currentWatermark() {
+int64_t WindowContext::currentWatermark() {
   return timerService_->currentWatermark();
 }
 
@@ -316,11 +316,11 @@ void WindowContext::clearTrigger(TimeWindow window) {
 }
 
 void WindowContext::deleteCleanupTimer(TimeWindow window) {
-  long cleanupTime =
+  int64_t cleanupTime =
       TimeWindowUtil::toEpochMillsForTimer(
           TimeWindowUtil::cleanupTime(window.maxTimestamp(), allowedLateness_, isEventTime_),
           shiftTimeZone_);
-  if (cleanupTime == LONG_MAX) {
+  if (cleanupTime == INT64_MAX) {
       // no need to clean up because we didn't set one
       return;
   }

velox/experimental/stateful/GroupWindowAggregator.h

@@ -21,7 +21,6 @@
 #include "velox/experimental/stateful/TimerHeapInternalTimer.h"
 #include "velox/experimental/stateful/Triggerable.h"
 #include "velox/experimental/stateful/window/SliceAssigner.h"
-#include "velox/experimental/stateful/window/WindowBuffer.h"
 #include "velox/experimental/stateful/window/WindowProcessFunction.h"
 #include "velox/experimental/stateful/window/WindowTrigger.h"