diff --git a/velox/core/QueryConfig.h b/velox/core/QueryConfig.h
index eccc48c8047..3d38573c610 100644
--- a/velox/core/QueryConfig.h
+++ b/velox/core/QueryConfig.h
@@ -562,6 +562,13 @@ class QueryConfig {
   static constexpr const char* kFieldNamesInJsonCastEnabled =
       "field_names_in_json_cast_enabled";
 
+  static constexpr const char* kStatefulTaskParallelism =
+      "stateful_task_parallelism";
+
+  int32_t statefulTaskParallelism() const {
+    return get<int32_t>(kStatefulTaskParallelism, 0);
+  }
+
   bool selectiveNimbleReaderEnabled() const {
     return get<bool>(kSelectiveNimbleReaderEnabled, false);
   }
diff --git a/velox/exec/HashPartitionFunction.cpp b/velox/exec/HashPartitionFunction.cpp
index 896facc4efa..a879dccc195 100644
--- a/velox/exec/HashPartitionFunction.cpp
+++ b/velox/exec/HashPartitionFunction.cpp
@@ -109,11 +109,11 @@ std::optional<uint32_t> HashPartitionFunction::partition(
   } else {
     if (localExchange_) {
       for (auto i = 0; i < size; ++i) {
-        partitions[i] = localExchangeHash(hashes_[i]) % numPartitions_;
+        partitions[i] = numPartitions_ > 0 ? localExchangeHash(hashes_[i]) % numPartitions_ : localExchangeHash(hashes_[i]);
       }
     } else {
       for (auto i = 0; i < size; ++i) {
-        partitions[i] = hashes_[i] % numPartitions_;
+        partitions[i] = numPartitions_ > 0 ? hashes_[i] % numPartitions_ : hashes_[i];
       }
     }
   }
diff --git a/velox/exec/HashPartitionFunction.h b/velox/exec/HashPartitionFunction.h
index 7aa6a032d6b..6eae8bedb63 100644
--- a/velox/exec/HashPartitionFunction.h
+++ b/velox/exec/HashPartitionFunction.h
@@ -59,7 +59,7 @@ class HashPartitionFunction : public core::PartitionFunction {
       const std::vector<VectorPtr>& constValues);
 
   const bool localExchange_;
-  const int numPartitions_;
+  int numPartitions_{0};
   const std::optional<HashBitRange> hashBitRange_ = std::nullopt;
   std::vector<std::unique_ptr<VectorHasher>> hashers_;
 
diff --git a/velox/experimental/stateful/KeySelector.cpp b/velox/experimental/stateful/KeySelector.cpp
index c555d3915b4..c465ce725ca 100644
--- a/velox/experimental/stateful/KeySelector.cpp
+++ b/velox/experimental/stateful/KeySelector.cpp
@@ -15,76 +15,86 @@
  */
 #include "velox/experimental/stateful/KeySelector.h"
 #include "velox/experimental/stateful/window/WindowPartitionFunction.h"
+#include <folly/container/F14Map.h>
+#include <algorithm>
+#include <memory>
+#include <optional>
 
 namespace facebook::velox::stateful {
 
+namespace {
+
+void allocateIndexBuffers(
+    const folly::F14FastMap<int64_t, vector_size_t>& partitionCounts,
+    folly::F14FastMap<int64_t, BufferPtr>& keyToIndexBuffers,
+    folly::F14FastMap<int64_t, vector_size_t*>& keyToRawIndices,
+    memory::MemoryPool* pool) {
+  keyToIndexBuffers.reserve(partitionCounts.size());
+  keyToRawIndices.reserve(partitionCounts.size());
+  for (const auto& [key, num] : partitionCounts) {
+    keyToIndexBuffers[key] = allocateIndices(num, pool);
+    keyToRawIndices[key] = keyToIndexBuffers[key]->asMutable<vector_size_t>();
+  }
+}
+
+} // namespace
+
 KeySelector::KeySelector(
     std::unique_ptr<core::PartitionFunction> partitionFunction,
     memory::MemoryPool* pool,
     int numPartitions)
     : partitionFunction_(std::move(partitionFunction)),
       pool_(pool),
-      numPartitions_(numPartitions) {}
+      numPartitions_(numPartitions) {
+  windowPartitionFunction_ =
+    dynamic_cast<WindowPartitionFunction*>(partitionFunction_.get());
+}
 
-std::map<int64_t, RowVectorPtr> KeySelector::partition(
-    const RowVectorPtr& input) {
-  if (numPartitions_ == 1) {
-    return std::map<int64_t, RowVectorPtr>{{0, input}};
-  }
+std::map<int64_t, RowVectorPtr> KeySelector::partition(const RowVectorPtr& input) {
   prepareForInput(input);
-
-  // TODO: The partition function doesn't use max parallelism.
-  std::vector<int64_t> partitions(input->size());
+  const auto numInput = input->size();
+  std::vector<int64_t> partitions(numInput);
   std::optional<int64_t> res;
-  auto windowPartitionFunction = dynamic_cast<WindowPartitionFunction*>(partitionFunction_.get());
-  if (windowPartitionFunction) {
-    res = windowPartitionFunction->partition(*input, partitions);
+  if (windowPartitionFunction_) {
+    res = windowPartitionFunction_->partition(*input, partitions);
   } else {
-    std::vector<uint32_t> tmpPartitions(input->size());
+    std::vector<uint32_t> tmpPartitions(numInput);
     std::optional<uint32_t> tmpRes = partitionFunction_->partition(*input, tmpPartitions);
     if (tmpRes) {
       res = static_cast<int64_t>(*tmpRes);
     }
-    for (vector_size_t i = 0; i < tmpPartitions.size(); ++i) {
-      partitions[i] = static_cast<int64_t>(tmpPartitions[i]);
+    for (vector_size_t i = 0; i < numInput; ++i) {
+      partitions[i] =static_cast<int64_t>(tmpPartitions[i]);
     }
   }
   if (res) {
-    // TODO: this is a optimization, as the RowVector may have be partitioned in
-    // local aggregation, so need not to partition again in global agg, but need
-    // to verify whether the judge condition is enough.
     return std::map<int64_t, RowVectorPtr>{{*res, input}};
   }
-  const auto numInput = input->size();
-  std::map<int64_t, vector_size_t> numOfKeys;
-  for (auto i = 0; i < numInput; ++i) {
-    if (numOfKeys.count(partitions[i]) == 0) {
-      numOfKeys[partitions[i]] = 1;
-    } else {
-      numOfKeys[partitions[i]] = numOfKeys[partitions[i]] + 1;
-    }
+
+  folly::F14FastMap<int64_t, vector_size_t> partitionCounts;
+  partitionCounts.reserve(std::min<vector_size_t>(numInput, 4096));
+
+  for (vector_size_t i = 0; i < numInput; ++i) {
+    ++partitionCounts[partitions[i]];
   }
+  folly::F14FastMap<int64_t, BufferPtr> keyToIndexBuffers;
+  folly::F14FastMap<int64_t, vector_size_t*> keyToRawIndices;
+  allocateIndexBuffers(partitionCounts, keyToIndexBuffers, keyToRawIndices, pool_);
 
-  std::map<int64_t, BufferPtr> keyToIndexBuffers;
-  std::map<int64_t, vector_size_t*> keyToRawIndices;
-  allocateIndexBuffers(numOfKeys, keyToIndexBuffers, keyToRawIndices);
+  folly::F14FastMap<int64_t, vector_size_t> nextRowIndex;
+  nextRowIndex.reserve(partitionCounts.size());
 
-  numOfKeys.clear();
-  for (auto i = 0; i < numInput; ++i) {
-    auto partition = partitions[i];
-    int index = 0;
-    if (numOfKeys.count(partition)) {
-      index = numOfKeys[partition];
-    }
+  for (vector_size_t i = 0; i < numInput; ++i) {
+    const auto partition = partitions[i];
+    const auto index = nextRowIndex[partition];
     keyToRawIndices[partition][index] = i;
-    numOfKeys[partition] = index + 1;
+    nextRowIndex[partition] = index + 1;
   }
 
   std::map<int64_t, RowVectorPtr> results;
-  for (auto& [key, partitionSize] : numOfKeys) {
-    auto partitionData =
-        wrapChildren(input, partitionSize, keyToIndexBuffers[key]);
-    results[key] = partitionData;
+  for (const auto& [partitionKey, partitionSize] : partitionCounts) {
+    results.emplace(
+      partitionKey, wrapChildren(input, partitionSize, keyToIndexBuffers[partitionKey]));
   }
   return results;
 }
@@ -99,16 +109,6 @@ void KeySelector::prepareForInput(const RowVectorPtr& input) {
   }
 }
 
-void KeySelector::allocateIndexBuffers(
-    const std::map<int64_t, vector_size_t>& numOfKeys,
-    std::map<int64_t, BufferPtr>& keyToIndexBuffers,
-    std::map<int64_t, vector_size_t*>& keyToRawIndices) {
-  for (auto& [key, num] : numOfKeys) {
-    keyToIndexBuffers[key] = allocateIndices(num, pool_);
-    keyToRawIndices[key] = keyToIndexBuffers[key]->asMutable<vector_size_t>();
-  }
-}
-
 RowVectorPtr KeySelector::wrapChildren(
     const RowVectorPtr& input,
     vector_size_t size,
diff --git a/velox/experimental/stateful/KeySelector.h b/velox/experimental/stateful/KeySelector.h
index d415acdbbb6..2083fca797b 100644
--- a/velox/experimental/stateful/KeySelector.h
+++ b/velox/experimental/stateful/KeySelector.h
@@ -17,11 +17,13 @@
 
 #include <climits>
 #include <map>
-#include "velox/experimental/stateful/StatefulPlanNode.h"
 #include "velox/vector/ComplexVector.h"
+#include "velox/core/PlanNode.h"
 
 namespace facebook::velox::stateful {
 
+class WindowPartitionFunction;
+
 /// This class is relevant to flink KeySelector.
 /// It can partition the RowVector according to the key fields.
 class KeySelector {
@@ -29,18 +31,17 @@ class KeySelector {
   KeySelector(
       std::unique_ptr<core::PartitionFunction> partitionFunction,
       memory::MemoryPool* pool,
-      int numPartitions = INT_MAX);
+      int numPartitions = 0);
 
   std::map<int64_t, RowVectorPtr> partition(const RowVectorPtr& input);
 
+  void setNumPartitions(int numPartitions) {
+    numPartitions_ = numPartitions;
+  }
+
  private:
   void prepareForInput(const RowVectorPtr& input);
 
-  void allocateIndexBuffers(
-      const std::map<int64_t, vector_size_t>& numOfKeys,
-      std::map<int64_t, BufferPtr>& keyToIndexBuffers,
-      std::map<int64_t, vector_size_t*>& keyToRawIndices);
-
   RowVectorPtr wrapChildren(
       const RowVectorPtr& input,
       vector_size_t size,
@@ -48,7 +49,9 @@ class KeySelector {
 
   const std::unique_ptr<core::PartitionFunction> partitionFunction_;
   memory::MemoryPool* pool_;
-  const int numPartitions_;
+  int numPartitions_;
+  /// Non-owning; null if partitionFunction_ is not a WindowPartitionFunction.
+  WindowPartitionFunction* windowPartitionFunction_{nullptr};
 };
 
 } // namespace facebook::velox::stateful
diff --git a/velox/experimental/stateful/LocalWindowAggregator.cpp b/velox/experimental/stateful/LocalWindowAggregator.cpp
index 91fe3e864f5..714db4b6a86 100644
--- a/velox/experimental/stateful/LocalWindowAggregator.cpp
+++ b/velox/experimental/stateful/LocalWindowAggregator.cpp
@@ -16,7 +16,6 @@
 #include "velox/experimental/stateful/LocalWindowAggregator.h"
 #include "velox/experimental/stateful/window/SliceAssigner.h"
 #include <cstdint>
-#include "velox/experimental/stateful/window/TimeWindowUtil.h"
 
 namespace facebook::velox::stateful {
 
@@ -47,15 +46,13 @@ void LocalWindowAggregator::advance() {
   if (!input_) {
     return;
   }
-
+  // partition input by key
   std::map<int64_t, RowVectorPtr> keyToData = keySelector_->partition(input_);
   for (const auto& [key, data] : keyToData) {
-    std::map<int64_t, RowVectorPtr> sliceEndToData =
-        sliceAssigner_->assignSliceEnd(data);
-    for (const auto& [sliceEnd, data] : sliceEndToData) {
-      // TODO: addElement may have data output.
-      auto windowData = data;
-      windowBuffer_->addElement(key, sliceEnd, windowData);
+    // assign slice end to data
+    std::map<int64_t, RowVectorPtr> sliceEndToData = sliceAssigner_->assignSliceEnd(data);
+    for (auto& [sliceEnd, data] : sliceEndToData) {
+      windowBuffer_->addElement(key, sliceEnd, data);
     }
   }
   input_.reset();
@@ -68,18 +65,33 @@ void LocalWindowAggregator::processWatermarkInternal(int64_t timestamp) {
       // we only need to call advanceProgress() when current watermark may
       // trigger window
       auto windowKeyToData = windowBuffer_->advanceProgress(currentWatermark_);
+      auto* aggregator = op().get();
+      auto* aggPool = aggregator->pool();
+      int64_t windowTriggered = -1;
       for (const auto& [windowKey, datas] : windowKeyToData) {
-        if (datas.empty()) {
+        const int64_t window = windowKey.window();
+        if (datas.empty() || currentWatermark_ < window) {
+          continue;
+        }
+        RowVectorPtr mergedInput = (datas.size() == 1) ? datas.front() : TimeWindowUtil::mergeVectors(datas, aggPool);
+        if (!mergedInput) {
           continue;
         }
-        op()->addInput(TimeWindowUtil::mergeVectors(datas, op()->pool()));
-        RowVectorPtr output = op()->getOutput();
+        aggregator->addInput(mergedInput);
+        RowVectorPtr output = aggregator->getOutput();
         if (output) {
-          pushOutput(std::make_shared<StreamRecord>(
-              getPlanNodeId(),
-              addWindowEndToVector(std::move(output), windowKey.window())));
+          pushOutput(
+            std::make_shared<StreamRecord>(getPlanNodeId(),
+            windowKey.key(),
+            std::move(addWindowEndToVector(output, window))));
+          if (windowTriggered < window) {
+            windowTriggered = window;
+          }
         }
       }
+      if (windowTriggered >= 0) {
+        windowBuffer_->clear(windowTriggered);
+      }
       nextTriggerWatermark_ = TimeWindowUtil::getNextTriggerWatermark(
           currentWatermark_,
           windowInterval_,
@@ -87,10 +99,11 @@ void LocalWindowAggregator::processWatermarkInternal(int64_t timestamp) {
           useDayLightSaving_);
     }
   }
+  pushOutput(std::make_shared<Watermark>(getPlanNodeId(), timestamp));
 }
 
 void LocalWindowAggregator::close() {
-  processWatermarkInternal(INT_MAX);
+  // processWatermarkInternal(INT_MAX);
   StatefulOperator::close();
   input_.reset();
   windowBuffer_->clear();
@@ -103,7 +116,6 @@ RowVectorPtr LocalWindowAggregator::addWindowEndToVector(
     int64_t sliceEnd) {
   auto newColumn = BaseVector::create(BIGINT(), vector->size(), vector->pool());
   auto windowEndCol = newColumn->as<FlatVector<int64_t>>();
-
   for (int i = 0; i < vector->size(); ++i) {
     windowEndCol->set(i, sliceEnd);
   }
diff --git a/velox/experimental/stateful/StatefulPlanNode.cpp b/velox/experimental/stateful/StatefulPlanNode.cpp
index 18c11c71fdc..18f959637df 100644
--- a/velox/experimental/stateful/StatefulPlanNode.cpp
+++ b/velox/experimental/stateful/StatefulPlanNode.cpp
@@ -15,6 +15,7 @@
  */
 #include "velox/experimental/stateful/StatefulPlanNode.h"
 #include "velox/experimental/stateful/window/WindowPartitionFunction.h"
+#include "velox/exec/PartitionFunction.h"
 #include <cstdint>
 
 namespace facebook::velox::stateful {
@@ -62,7 +63,7 @@ core::PlanNodePtr StatefulPlanNode::create(
 }
 
 void StatefulPlanNode::registerSerDe() {
-  registerPartitionFunctionSerDe();
+  exec::registerPartitionFunctionSerDe();
 
   auto& registry = DeserializationWithContextRegistryForSharedPtr();
 
diff --git a/velox/experimental/stateful/StatefulPlanner.cpp b/velox/experimental/stateful/StatefulPlanner.cpp
index 68c52e41f16..ab9ebc7b511 100644
--- a/velox/experimental/stateful/StatefulPlanner.cpp
+++ b/velox/experimental/stateful/StatefulPlanner.cpp
@@ -260,8 +260,13 @@ StatefulOperatorPtr StatefulPlanner::transformStreamWindowAggregationOperator(
   std::unique_ptr<KeySelector> keySelectorForSliceAssigner = std::make_unique<KeySelector>(
       windowAggNode->sliceAssignerSpec()->create(INT_MAX, true), op->pool());
   std::unique_ptr<SliceAssigner> sliceAssigner =
-      std::make_unique<SliceAssigner>(std::move(keySelectorForSliceAssigner), windowAggNode->size(), windowAggNode->step(),
-      windowAggNode->offset(),Window::getType(windowAggNode->windowType()),windowAggNode->rowtimeIndex());
+      std::make_unique<SliceAssigner>(
+          std::move(keySelectorForSliceAssigner),
+          windowAggNode->size(),
+          windowAggNode->step(),
+          windowAggNode->offset(),
+          Window::getType(windowAggNode->windowType()),
+          windowAggNode->rowtimeIndex());
   if (windowAggNode->isLocalAgg()) {
     return std::make_unique<LocalWindowAggregator>(
         std::move(op),
@@ -282,6 +287,7 @@ StatefulOperatorPtr StatefulPlanner::transformStreamWindowAggregationOperator(
         std::move(keySelector),
         std::move(sliceAssigner),
         windowAggNode->windowInterval(),
+        windowAggNode->size(),
         windowAggNode->useDayLightSaving(),
         windowAggNode->isEventTime(),
         windowAggNode->windowStartIndex(),
diff --git a/velox/experimental/stateful/StatefulPlanner.h b/velox/experimental/stateful/StatefulPlanner.h
index afebee68545..7d8208ae87f 100644
--- a/velox/experimental/stateful/StatefulPlanner.h
+++ b/velox/experimental/stateful/StatefulPlanner.h
@@ -36,11 +36,12 @@ class StatefulPlanner {
 
  protected:
   StatefulPlanner(exec::DriverCtx* ctx, StateBackend* stateBackend)
-      : ctx_(ctx), stateBackend_(stateBackend) {}
+      : ctx_(ctx), stateBackend_(stateBackend), parallelism_(ctx->queryConfig().statefulTaskParallelism()) {}
 
  private:
   exec::DriverCtx* ctx_ = nullptr;
   StateBackend* stateBackend_ = nullptr;
+  int32_t parallelism_ = 0;
 
   StatefulOperatorPtr transformStatefulOperators(
       const core::PlanNodePtr& planNode);
diff --git a/velox/experimental/stateful/WindowAggregator.cpp b/velox/experimental/stateful/WindowAggregator.cpp
index 9eea120dc42..9b62dde1ade 100644
--- a/velox/experimental/stateful/WindowAggregator.cpp
+++ b/velox/experimental/stateful/WindowAggregator.cpp
@@ -31,22 +31,24 @@ WindowAggregator::WindowAggregator(
     std::vector<std::unique_ptr<StatefulOperator>> targets,
     std::unique_ptr<KeySelector> keySelector,
     std::unique_ptr<SliceAssigner> sliceAssigner,
-    const long windowInterval,
+    const int64_t windowInterval,
+    const int64_t windowSize,
     const bool useDayLightSaving,
     const bool isEventTime,
     const int windowStartIndex,
     const int windowEndIndex)
-    : StatefulOperator(std::move(globalAggregator), std::move(targets)), Triggerable<int64_t, long>(),
+    : StatefulOperator(std::move(globalAggregator), std::move(targets)), Triggerable<int64_t, int64_t>(),
       localAggregator_(std::move(localAggregator)),
       keySelector_(std::move(keySelector)),
       sliceAssigner_(std::move(sliceAssigner)),
       windowInterval_(windowInterval),
+      windowSize_(windowSize),
       useDayLightSaving_(useDayLightSaving),
       isEventTime_(isEventTime),
       windowStartIndex_(windowStartIndex),
       windowEndIndex_(windowEndIndex) {
-        windowBuffer_ = std::make_shared<RecordsWindowBuffer>();
-    }
+      windowBuffer_ = std::make_shared<RecordsWindowBuffer>();
+  }
 
 void WindowAggregator::initialize() {
   StatefulOperator::initialize();
@@ -71,11 +73,9 @@ void WindowAggregator::advance() {
   if (!input_) {
     return;
   }
-
   std::map<int64_t, RowVectorPtr> keyToData = keySelector_->partition(input_);
   for (const auto& [key, data] : keyToData) {
-    std::map<int64_t, RowVectorPtr> sliceEndToData =
-        sliceAssigner_->assignSliceEnd(data);
+    std::map<int64_t, RowVectorPtr> sliceEndToData = sliceAssigner_->assignSliceEnd(data);
     for (const auto& [sliceEnd, data] : sliceEndToData) {
       auto windowData = data;
       if (!isEventTime_) {
@@ -116,35 +116,49 @@ void WindowAggregator::processWatermarkInternal(int64_t timestamp) {
       // we only need to call advanceProgress() when current watermark may
       // trigger window
       auto windowKeyToData = windowBuffer_->advanceProgress(currentProgress_);
+      auto* globalAggregator = op().get();
+      auto* aggPool = globalAggregator->pool();
       for (const auto& [windowKey, datas] : windowKeyToData) {
         if (datas.empty()) {
           continue;
         }
-        // TODO: agg should output no matter how many rows in data.
-        VELOX_CHECK(localAggregator_, "Local aggregator not set");
-        localAggregator_->addInput(
-            TimeWindowUtil::mergeVectors(datas, op()->pool()));
+        const auto key = windowKey.key();
+        const auto window = windowKey.window();
+        RowVectorPtr localInput =
+            (datas.size() == 1) ? datas.front()
+                                : TimeWindowUtil::mergeVectors(datas, aggPool);
+        if (localInput) {
+          localAggregator_->addInput(localInput);
+        }
         RowVectorPtr localAcc = localAggregator_->getOutput();
-        auto stateAcc =
-            windowState_->value(windowKey.key(), windowKey.window());
-        std::list<RowVectorPtr> allDatas;
+        RowVectorPtr stateAcc = windowState_->value(key, window);
+
         if (!localAcc && !stateAcc) {
           continue;
+        }
+
+        // Fast path: avoid building temporary list/merge when only one side
+        // exists.
+        if (!stateAcc) {
+          globalAggregator->addInput(localAcc);
+        } else if (!localAcc) {
+          globalAggregator->addInput(stateAcc);
         } else {
-          if (localAcc) {
-            allDatas.push_back(localAcc);
-          }
-          if (stateAcc) {
-            allDatas.push_back(stateAcc);
-          }
-          op()->addInput(TimeWindowUtil::mergeVectors(allDatas, op()->pool()));
-          auto newAcc = op()->getOutput();
-          if (newAcc) {
-            windowState_->update(windowKey.key(), windowKey.window(), newAcc);
-          }
+          std::list<RowVectorPtr> allDatas;
+          allDatas.push_back(localAcc);
+          allDatas.push_back(stateAcc);
+          globalAggregator->addInput(TimeWindowUtil::mergeVectors(allDatas, aggPool));
+        }
+
+        auto newAcc = globalAggregator->getOutput();
+        if (newAcc) {
+          windowState_->update(key, window, newAcc);
         }
+        windowTimerService_->registerEventTimeTimer(key, window, window - 1);
+      }
+      if (!windowKeyToData.empty()) {
+        windowBuffer_->clear();
       }
-      windowTimerService_->advanceWatermark(currentProgress_);
       nextTriggerWatermark_ = TimeWindowUtil::getNextTriggerWatermark(
           currentProgress_,
           windowInterval_,
@@ -152,16 +166,9 @@ void WindowAggregator::processWatermarkInternal(int64_t timestamp) {
           useDayLightSaving_);
     }
   }
-}
-
-void WindowAggregator::onEventTime(
-    std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>> timer) {
-  onTimer(timer);
-}
-
-int64_t WindowAggregator::sliceStateMergeTarget(int64_t sliceToMerge) {
-  // TODO: implement it
-  return sliceToMerge;
+  int64_t t = timestamp > currentProgress_ ? timestamp : currentProgress_;
+  windowTimerService_->advanceWatermark(t);
+  pushOutput(std::make_shared<Watermark>(getPlanNodeId(), t));
 }
 
 /// Add window_start / window_end timestamp to output
@@ -234,7 +241,7 @@ void WindowAggregator::fireWindow(const K& key, int64_t timerTimestamp, int64_t
       output = addWindowTimestampToOutput(
         output,
         "window_start",
-        windowEnd - windowInterval_,
+        windowEnd - windowSize_,
         windowStartIndex_);
     }
     if (windowEndIndex_ >= 0) {
@@ -246,7 +253,7 @@ void WindowAggregator::fireWindow(const K& key, int64_t timerTimestamp, int64_t
     }
   }
   if (output) {
-    pushOutput(std::make_shared<StreamRecord>(getPlanNodeId(), std::move(output)));
+    pushOutput(std::make_shared<StreamRecord>(getPlanNodeId(), key, std::move(output)));
   }
 }
 
@@ -255,27 +262,36 @@ void WindowAggregator::clearWindow(const K& key, int64_t timerTimestamp, int64_t
   windowState_->remove(key, windowEnd);
 }
 
+void WindowAggregator::onEventTime(std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>> timer) {
+  onTimer(timer);
+}
+
 void WindowAggregator::onProcessingTime(std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>> timer) {
   if (timer->timestamp() > lastTriggeredProcessingTime_) {
     lastTriggeredProcessingTime_ = timer->timestamp();
     auto windowKeyToData = windowBuffer_->advanceProgress(timer->timestamp());
-    for (auto& [windowKey, datas] : windowKeyToData) {
+    int64_t windowTriggered = -1;
+    for (const auto& [windowKey, datas] : windowKeyToData) {
       if (datas.empty()) {
         continue;
       }
+      std::list<RowVectorPtr> allDatas(datas.begin(), datas.end());
       auto stateAcc = windowState_->value(windowKey.key(), windowKey.window());
       if (stateAcc) {
-        datas.push_back(stateAcc);
+        allDatas.push_back(stateAcc);
       }
-      RowVectorPtr opInput = TimeWindowUtil::mergeVectors(datas, op()->pool());
+      RowVectorPtr opInput = TimeWindowUtil::mergeVectors(allDatas, op()->pool());
       op()->addInput(opInput);
       auto newAcc = op()->getOutput();
       if (newAcc) {
         windowState_->update(windowKey.key(), windowKey.window(), newAcc);
       }
+      if (windowKey.window() > windowTriggered) {
+        windowTriggered = windowKey.window();
+      }
     }
     if (!windowKeyToData.empty()) {
-      windowBuffer_->clear();
+      windowBuffer_->clear(windowTriggered);
     }
   }
   onTimer(timer);
@@ -288,16 +304,32 @@ void WindowAggregator::processProcessingTimeByJni(int64_t) {
     "processElementByJni", "WindowAggOperator");
 }
 
+int64_t WindowAggregator::sliceStateMergeTarget(int64_t sliceToMerge) {
+  // TODO: implement it
+  return sliceToMerge;
+}
+
 void WindowAggregator::close() {
-  // processWatermarkInternal(std::numeric_limits<int64_t>::max());
+  // processWatermarkInternal(INT_MAX);
   StatefulOperator::close();
+  if (windowTimerService_) {
+    windowTimerService_->close();
+  }
+  input_.reset();
+  if (windowBuffer_) {
+    windowBuffer_->clear();
+  }
+  if (windowState_) {
+    windowState_->clear();
+  }
   if (localAggregator_) {
     localAggregator_->close();
+    // Release aggregator-owned memory pool immediately on close instead of
+    // waiting for WindowAggregator destruction.
+    localAggregator_.reset();
   }
-  input_.reset();
-  windowTimerService_->close();
-  windowBuffer_->clear();
-  windowState_->clear();
+  windowTimerService_.reset();
+  windowState_.reset();
   currentProgress_ = 0;
   nextTriggerWatermark_ = 0;
 }
diff --git a/velox/experimental/stateful/WindowAggregator.h b/velox/experimental/stateful/WindowAggregator.h
index 4a52329c324..ec993a1d91f 100644
--- a/velox/experimental/stateful/WindowAggregator.h
+++ b/velox/experimental/stateful/WindowAggregator.h
@@ -36,12 +36,13 @@ class WindowAggregator : public StatefulOperator,
                          public Triggerable<int64_t, int64_t> {
  public:
   WindowAggregator(
-    std::unique_ptr<exec::Operator> localAggregator,
+    std::unique_ptr<exec::Operator> mergeOperator,
     std::unique_ptr<exec::Operator> globalAggregator,
     std::vector<std::unique_ptr<StatefulOperator>> targets,
     std::unique_ptr<KeySelector> keySelector,
     std::unique_ptr<SliceAssigner> sliceAssigner,
     const int64_t windowInterval,
+    const int64_t windowSize,
     const bool useDayLightSaving,
     const bool isEventTime,
     const int32_t windowStartIndex,
@@ -62,14 +63,13 @@ class WindowAggregator : public StatefulOperator,
   }
 
   void processWatermark(int64_t timestamp) override {
-    // processWatermarkInternal(timestamp);
+    processWatermarkInternal(timestamp);
   }
 
   void onEventTime(std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>>
                        timer) override;
 
-  void onProcessingTime(
-      std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>> timer) override;
+  void onProcessingTime(std::shared_ptr<TimerHeapInternalTimer<int64_t, int64_t>> timer) override;
 
   void processProcessingTimeByJni(int64_t timestamp) override;
 
@@ -91,6 +91,7 @@ class WindowAggregator : public StatefulOperator,
   std::unique_ptr<SliceAssigner> sliceAssigner_;
   WindowBufferPtr windowBuffer_;
   const int64_t windowInterval_;
+  const int64_t windowSize_;
   const bool useDayLightSaving_;
   const int shiftTimeZone_ = 0; // TODO: support time zone shift
   const bool isEventTime_ = true;
diff --git a/velox/experimental/stateful/WindowJoin.cpp b/velox/experimental/stateful/WindowJoin.cpp
index 3135b6f97be..2d4e70f4733 100644
--- a/velox/experimental/stateful/WindowJoin.cpp
+++ b/velox/experimental/stateful/WindowJoin.cpp
@@ -15,8 +15,6 @@
  */
 #include "velox/experimental/stateful/WindowJoin.h"
 #include <cstdint>
-#include "velox/experimental/stateful/join/JoinRecordStateViews.h"
-#include "velox/expression/Expr.h"
 
 namespace facebook::velox::stateful {
 
diff --git a/velox/experimental/stateful/WindowJoin.h b/velox/experimental/stateful/WindowJoin.h
index 8b6c49639f2..93b9eada136 100644
--- a/velox/experimental/stateful/WindowJoin.h
+++ b/velox/experimental/stateful/WindowJoin.h
@@ -20,13 +20,11 @@
 #include "velox/experimental/stateful/InternalTimerService.h"
 #include "velox/experimental/stateful/KeySelector.h"
 #include "velox/experimental/stateful/StatefulOperator.h"
-#include "velox/experimental/stateful/StatefulPlanNode.h"
 #include "velox/experimental/stateful/StreamElement.h"
 #include "velox/experimental/stateful/TimerHeapInternalTimer.h"
 #include "velox/experimental/stateful/Triggerable.h"
-#include "velox/experimental/stateful/join/JoinRecordStateView.h"
 
-    namespace facebook::velox::stateful {
+namespace facebook::velox::stateful {
 
   class WindowJoin : public StatefulOperator,
                      public Triggerable<int64_t, int64_t> {
diff --git a/velox/experimental/stateful/window/SliceAssigner.cpp b/velox/experimental/stateful/window/SliceAssigner.cpp
index a549fbee5f0..149ba3fe1af 100644
--- a/velox/experimental/stateful/window/SliceAssigner.cpp
+++ b/velox/experimental/stateful/window/SliceAssigner.cpp
@@ -14,28 +14,18 @@
  * limitations under the License.
  */
 #include "velox/experimental/stateful/window/SliceAssigner.h"
-#include "velox/experimental/stateful/window/TimeWindowUtil.h"
 #include "velox/experimental/stateful/window/Window.h"
-#include "velox/type/Timestamp.h"
 #include "velox/vector/ComplexVector.h"
-#include "velox/vector/ConstantVector.h"
 #include "velox/vector/DictionaryVector.h"
-#include "velox/vector/FlatVector.h"
+#include "velox/experimental/stateful/window/TimeWindowUtil.h"
 #include <cstdint>
 #include <numeric>
-#include <unordered_map>
 #include <vector>
 
 namespace facebook::velox::stateful {
 
 namespace {
 
-void prepareChildrenLoaded(const RowVectorPtr& input) {
-  for (auto& child : input->children()) {
-    child->loadedVector();
-  }
-}
-
 RowVectorPtr wrapChildrenByIndices(
     const RowVectorPtr& input,
     vector_size_t size,
@@ -73,92 +63,65 @@ SliceAssigner::SliceAssigner(
     int64_t step,
     int64_t offset,
     WindowType windowType,
-    int rowtimeIndex)
+    int rowtimeIndex,
+    bool expandHopWindows)
     : keySelector_(std::move(keySelector)),
       size_(size),
       step_(step),
       offset_(offset),
       windowType_(windowType),
-      rowtimeIndex_(rowtimeIndex) {
-  // TODO: calculate sliceSize_ based on windowType.
-  sliceSize_ = std::gcd(size, step);
+      rowtimeIndex_(rowtimeIndex),
+      expandHopWindows_(expandHopWindows) {
+  if (windowType_ == WindowType::TUMBLE) {
+    sliceSize_ = size;
+  } else if (windowType_ == WindowType::HOP) {
+    sliceSize_ = std::gcd(size, step);
+  } else {
+    VELOX_FAIL("window type: {} is not supported", static_cast<int32_t>(windowType_));
+  }
 }
 
 std::map<int64_t, RowVectorPtr> SliceAssigner::assignSliceEnd(const RowVectorPtr& input) {
-  if (rowtimeIndex_ < 0) {
-    int64_t timestampMs = TimeWindowUtil::getCurrentProcessingTime();
-    if (windowType_ == WindowType::TUMBLE) {
+  auto calculateWindowEnd = [&]
+  (int64_t timestampMs, int64_t offset, int64_t sliceSize, int64_t windowSize, bool notExpandEventTimeWindow) -> int64_t {
+    if (notExpandEventTimeWindow) {
+       return timestampMs;
+    } else {
       int64_t utcTimestamp = TimeWindowUtil::toEpochMillsForTimer(timestampMs, 0);
-      int64_t windowStart = stateful::TimeWindowUtil::getWindowStartWithOffset(utcTimestamp, offset_, size_);
-      return {{windowStart + size_, input}};
+      return TimeWindowUtil::getWindowStartWithOffset(utcTimestamp, offset, sliceSize) + windowSize;
+    }
+  };
+  std::map<int64_t, RowVectorPtr> res;
+  std::map<int64_t, RowVectorPtr> partitionToData = keySelector_->partition(input);
+  // set window end to data
+  auto setWindowEnd = [&](int64_t windowEnd, const RowVectorPtr& data) {
+    if (res.count(windowEnd) == 0) {
+      res[windowEnd] = data;
     } else {
-      return {{timestampMs, input}};
+      res[windowEnd] = TimeWindowUtil::mergeVectors({res[windowEnd], data}, input->pool());
     }
-  } else {
-    const VectorPtr& rowtimeVector = input->childAt(rowtimeIndex_);
-    prepareChildrenLoaded(input);
-    const auto* tsSimple = rowtimeVector->as<SimpleVector<Timestamp>>();
-    const auto* tsSimpleInt = rowtimeVector->as<SimpleVector<int64_t>>();
-    VELOX_CHECK(tsSimple != nullptr || tsSimpleInt != nullptr,
-        "rowtime column must be TIMESTAMP/Int64 simple vector");
-
-    const vector_size_t numRows = rowtimeVector->size();
-    auto isNullAtRow = [&](vector_size_t row) {
-      return tsSimple ? tsSimple->isNullAt(row) : tsSimpleInt->isNullAt(row);
-    };
-    auto timestampMillisAt = [&](vector_size_t row) {
-      return tsSimple ? tsSimple->valueAt(row).toMillis() : tsSimpleInt->valueAt(row);
-    };
-
-    velox::memory::MemoryPool* pool = input->pool();
-    std::map<int64_t, RowVectorPtr> sliceEndToData;
-
+  };
+  bool isEventTimeWindow = rowtimeIndex_ >= 0 && (windowType_ == WindowType::HOP || windowType_ == WindowType::TUMBLE);
+  // assign slice end to data
+  for (auto& kv : partitionToData) {
+    int64_t windowEnd = calculateWindowEnd(kv.first, offset_, sliceSize_, size_,
+      isEventTimeWindow && !expandHopWindows_);
     if (windowType_ == WindowType::TUMBLE) {
-      std::unordered_map<int64_t, std::vector<vector_size_t>> groups;
-      for (vector_size_t i = 0; i < numRows; ++i) {
-        if (isNullAtRow(i)) {
-          continue;
-        }
-        int64_t timestampMs = timestampMillisAt(i);
-        int64_t utcTimestamp = TimeWindowUtil::toEpochMillsForTimer(timestampMs, 0);
-        int64_t windowStart =
-            stateful::TimeWindowUtil::getWindowStartWithOffset(utcTimestamp, offset_, size_);
-        const int64_t sliceEnd = windowStart + size_;
-        groups[sliceEnd].push_back(i);
-      }
-      for (auto& [sliceEnd, rowIndices] : groups) {
-        const vector_size_t n = rowIndices.size();
-        BufferPtr indicesBuf = allocateIndices(n, pool);
-        auto* raw = indicesBuf->asMutable<vector_size_t>();
-        for (vector_size_t j = 0; j < n; ++j) {
-          raw[j] = rowIndices[j];
+      setWindowEnd(windowEnd, kv.second);
+    } else if (windowType_ == WindowType::HOP) {
+      if (expandHopWindows_) {
+        int64_t end = rowtimeIndex_ < 0 ? windowEnd : windowEnd - sliceSize_;
+        for (; end >= kv.first; end -= sliceSize_) {
+          setWindowEnd(end, kv.second);
         }
-        sliceEndToData[sliceEnd] =
-            wrapChildrenByIndices(input, n, indicesBuf, pool);
+      } else {
+        setWindowEnd(windowEnd, kv.second);
       }
-      return sliceEndToData;
-    }
-    std::unordered_map<int64_t, std::vector<vector_size_t>> groups;
-    for (vector_size_t i = 0; i < numRows; ++i) {
-      if (isNullAtRow(i)) {
-        continue;
-      }
-      int64_t timestampMs = timestampMillisAt(i);
-      int64_t key = TimeWindowUtil::toEpochMillsForTimer(timestampMs, 0);
-      groups[key].push_back(i);
-    }
-    for (auto& [timeKey, rowIndices] : groups) {
-      const vector_size_t n = rowIndices.size();
-      BufferPtr indicesBuf = allocateIndices(n, pool);
-      auto* raw = indicesBuf->asMutable<vector_size_t>();
-      for (vector_size_t j = 0; j < n; ++j) {
-        raw[j] = rowIndices[j];
-      }
-      sliceEndToData[timeKey] =
-          wrapChildrenByIndices(input, n, indicesBuf, pool);
+    } else {
+      VELOX_FAIL("window type: {} is not supported", static_cast<int32_t>(windowType_));
     }
-    return sliceEndToData;
   }
+  return res;
 }
 
 int64_t SliceAssigner::getLastWindowEnd(int64_t sliceEnd) {
diff --git a/velox/experimental/stateful/window/SliceAssigner.h b/velox/experimental/stateful/window/SliceAssigner.h
index d7842f648a7..6d47fe0fa4c 100644
--- a/velox/experimental/stateful/window/SliceAssigner.h
+++ b/velox/experimental/stateful/window/SliceAssigner.h
@@ -15,7 +15,6 @@
  */
 #pragma once
 #include <cstdint>
-
 #include "velox/experimental/stateful/KeySelector.h"
 #include "velox/experimental/stateful/window/Window.h"
 
@@ -30,7 +29,8 @@ class SliceAssigner {
       int64_t step,
       int64_t offset,
       WindowType windowType,
-      int rowtimeIndex);
+      int rowtimeIndex,
+      bool expandHopWindows = true);
 
   std::map<int64_t, RowVectorPtr> assignSliceEnd(const RowVectorPtr& input);
 
@@ -50,7 +50,7 @@ class SliceAssigner {
   const WindowType windowType_;
   int64_t sliceSize_;
   int rowtimeIndex_;
+  const bool expandHopWindows_;
 };
 
 } // namespace facebook::velox::stateful
- 
diff --git a/velox/experimental/stateful/window/TimeWindowUtil.cpp b/velox/experimental/stateful/window/TimeWindowUtil.cpp
index 9cbc16ac204..08c219b31aa 100644
--- a/velox/experimental/stateful/window/TimeWindowUtil.cpp
+++ b/velox/experimental/stateful/window/TimeWindowUtil.cpp
@@ -28,7 +28,7 @@ int64_t TimeWindowUtil::getNextTriggerWatermark(
     return currentWatermark;
   }
 
-  int64_t triggerWatermark;
+  int64_t triggerWatermark = currentWatermark;
   // consider the DST timezone
   if (useDayLightSaving) {
     // TODO: support time zone
@@ -114,20 +114,36 @@ RowVectorPtr TimeWindowUtil::mergeVectors(
   if (datas.empty()) {
     return nullptr;
   }
-  // TODO: refine it
-  auto numColumns = datas.front()->childrenSize();
-  std::vector<VectorPtr> mergedColumns(numColumns);
 
+  RowVectorPtr firstNonEmpty = nullptr;
   size_t totalRows = 0;
+  size_t nonEmptyBatches = 0;
   for (const auto& data : datas) {
+    if (!data || data->size() == 0) {
+      continue;
+    }
+    if (!firstNonEmpty) {
+      firstNonEmpty = data;
+    }
     totalRows += data->size();
+    ++nonEmptyBatches;
   }
 
-  auto merged =
-      BaseVector::create<RowVector>(datas.front()->type(), totalRows, pool);
+  if (nonEmptyBatches == 0) {
+    return nullptr;
+  }
+  if (nonEmptyBatches == 1) {
+    // Fast path: avoid allocating/copying when only one input has rows.
+    return firstNonEmpty;
+  }
 
-  size_t offset = 0;
-  for (auto& data : datas) {
+  auto merged = BaseVector::create<RowVector>(firstNonEmpty->type(), totalRows, pool);
+
+  vector_size_t offset = 0;
+  for (const auto& data : datas) {
+    if (!data || data->size() == 0) {
+      continue;
+    }
     merged->copy(data.get(), offset, 0, data->size());
     offset += data->size();
   }
diff --git a/velox/experimental/stateful/window/TimeWindowUtil.h b/velox/experimental/stateful/window/TimeWindowUtil.h
index 0a8fc4b0822..407774eeeba 100644
--- a/velox/experimental/stateful/window/TimeWindowUtil.h
+++ b/velox/experimental/stateful/window/TimeWindowUtil.h
@@ -46,8 +46,7 @@ class TimeWindowUtil {
 
   static int64_t toEpochMillsForTimer(int64_t timestamp, int shiftTimeZone);
 
-  static int64_t
-  cleanupTime(int64_t maxTimestamp, int64_t allowedLateness_, bool isEventTime);
+  static int64_t cleanupTime(int64_t maxTimestamp, int64_t allowedLateness_, bool isEventTime);
 
   static int64_t getCurrentProcessingTime();
 };
diff --git a/velox/experimental/stateful/window/WindowBuffer.cpp b/velox/experimental/stateful/window/WindowBuffer.cpp
index b1abaa9a69e..1ff3425a08d 100644
--- a/velox/experimental/stateful/window/WindowBuffer.cpp
+++ b/velox/experimental/stateful/window/WindowBuffer.cpp
@@ -38,7 +38,7 @@ void RecordsWindowBuffer::addElement(
   }
 }
 
-std::unordered_map<WindowKey, std::list<RowVectorPtr>>&
+std::map<WindowKey, std::list<RowVectorPtr>>&
 RecordsWindowBuffer::advanceProgress(int64_t progress) {
   if (TimeWindowUtil::isWindowFired(minSliceEnd_, progress, shiftTimeZone_)) {
     // There should be some window to be fired, flush buffer to state first.
diff --git a/velox/experimental/stateful/window/WindowBuffer.h b/velox/experimental/stateful/window/WindowBuffer.h
index aa0a1a4fa81..355d2b9feb7 100644
--- a/velox/experimental/stateful/window/WindowBuffer.h
+++ b/velox/experimental/stateful/window/WindowBuffer.h
@@ -14,14 +14,14 @@
  * limitations under the License.
  */
 #pragma once
+#include <algorithm>
 #include <cstdint>
 #include <list>
+#include <map>
 #include <memory>
-#include <unordered_map>
 #include "velox/experimental/stateful/window/WindowKey.h"
 #include "velox/vector/ComplexVector.h"
 #include <climits>
-#include <mutex>
 
 namespace facebook::velox::stateful {
 
@@ -33,11 +33,13 @@ class WindowBuffer {
   virtual void
   addElement(uint32_t key, int64_t window, RowVectorPtr& element) = 0;
 
-  virtual std::unordered_map<WindowKey, std::list<RowVectorPtr>>&
+  virtual std::map<WindowKey, std::list<RowVectorPtr>>&
   advanceProgress(int64_t progress) = 0;
 
   virtual void clear() = 0;
 
+  virtual void clear(int64_t window) = 0;
+
   virtual int size() = 0;
 };
 
@@ -49,22 +51,43 @@ class RecordsWindowBuffer : public WindowBuffer {
   void addElement(uint32_t key, int64_t sliceEnd, RowVectorPtr& element)
       override;
 
-  std::unordered_map<WindowKey, std::list<RowVectorPtr>>& advanceProgress(
+  std::map<WindowKey, std::list<RowVectorPtr>>& advanceProgress(
       int64_t progress) override;
 
   void clear() override {
     buffer_.clear();
     minSliceEnd_ = INT64_MAX;
   }
+
+  void clear(int64_t window) override {
+    // Do not erase during a range-for over the same map; erasure
+    // invalidates iterators and causes undefined behavior / crashes.
+    for (auto it = buffer_.begin(); it != buffer_.end();) {
+      if (it->first.window() <= window) {
+        it = buffer_.erase(it);
+      } else {
+        break;
+      }
+    }
+    if (buffer_.empty()) {
+      minSliceEnd_ = INT64_MAX;
+    } else {
+      minSliceEnd_ = INT64_MAX;
+      for (const auto& entry : buffer_) {
+        minSliceEnd_ = std::min(minSliceEnd_, entry.first.window());
+      }
+    }
+  }
+
   int size() override {
     return buffer_.size();
   }
 
- private:
-  // TODO: use map to simplify.
-  std::unordered_map<WindowKey, std::list<RowVectorPtr>> buffer_;
+  private:
+  /// Ordered by WindowKey (window end, then partition key) for stable iteration.
+  std::map<WindowKey, std::list<RowVectorPtr>> buffer_;
   // This is used to return empty map when no window is fired.
-  std::unordered_map<WindowKey, std::list<RowVectorPtr>> empty_;
+  std::map<WindowKey, std::list<RowVectorPtr>> empty_;
   int64_t minSliceEnd_ = INT64_MAX;
   int shiftTimeZone_ = 0; // TODO: support time zone shift
 };
diff --git a/velox/experimental/stateful/window/WindowKey.h b/velox/experimental/stateful/window/WindowKey.h
index 7bec1d6c311..5a2d12360ca 100644
--- a/velox/experimental/stateful/window/WindowKey.h
+++ b/velox/experimental/stateful/window/WindowKey.h
@@ -17,7 +17,6 @@
 #include <cstdint>
 #include <functional>
 
-#include "velox/vector/ComplexVector.h"
 
 namespace facebook::velox::stateful {
 
@@ -38,6 +37,13 @@ class WindowKey {
     return key_ == other.key() && window_ == other.window();
   }
 
+  bool operator<(const WindowKey& other) const {
+    if (window_ != other.window_) {
+      return window_ < other.window_;
+    }
+    return key_ < other.key_;
+  }
+
  private:
   uint32_t key_;
   int64_t window_;
diff --git a/velox/experimental/stateful/window/WindowPartitionFunction.cpp b/velox/experimental/stateful/window/WindowPartitionFunction.cpp
index 5ea3b52715e..d2becf9f022 100644
--- a/velox/experimental/stateful/window/WindowPartitionFunction.cpp
+++ b/velox/experimental/stateful/window/WindowPartitionFunction.cpp
@@ -14,17 +14,17 @@
  * limitations under the License.
  */
 #include "velox/experimental/stateful/window/WindowPartitionFunction.h"
-#include <cstdint>
 #include "velox/experimental/stateful/window/TimeWindowUtil.h"
 #include "velox/experimental/stateful/window/Window.h"
 
+#include <cstdint>
 #include <numeric>
 
 namespace facebook::velox::stateful {
 
 WindowPartitionFunction::WindowPartitionFunction(
     const RowTypePtr& inputType,
-    const column_index_t rowtimeIndex,
+    const int32_t rowtimeIndex,
     int64_t size,
     int64_t step,
     int64_t offset,
@@ -35,14 +35,19 @@ WindowPartitionFunction::WindowPartitionFunction(
       step_(step),
       offset_(offset),
       windowType_(windowType) {
+  // VELOX_CHECK_GT(inputType->size(), rowtimeIndex, "rowtimeIndex invalid: {}",
+  //    rowtimeIndex);
   sliceSize_ = std::gcd(size, step);
 }
 
 std::optional<int64_t> WindowPartitionFunction::partition(
-  const RowVector& input,
-  std::vector<int64_t>& partitions) {
+    const RowVector& input,
+    std::vector<int64_t>& partitions) {
+  if (rowtimeIndex_ < 0) {
+    return std::optional<int64_t>(TimeWindowUtil::getCurrentProcessingTime());
+  }
   if (inputType_->childAt(rowtimeIndex_)->kind() == TypeKind::BIGINT) {
-    // TODO: this is a optimization, as the RowVector may have be partitioned in
+    // This is a optimization, as the RowVector may have be partitioned in
     // local aggregation, so need not to partition again in global agg, but need
     // to verify whether the judge condition is enough.
     auto child = input.childAt(rowtimeIndex_);
@@ -50,16 +55,14 @@ std::optional<int64_t> WindowPartitionFunction::partition(
     return ts;
   }
   const auto size = input.size();
-  partitions.clear();
   partitions.resize(size);
   // TODO: support more window types. Support time zone.
   for (auto i = 0; i < size; ++i) {
-    const auto& child = input.childAt(rowtimeIndex_);
+    auto child = input.childAt(rowtimeIndex_);
     auto ts = child->as<SimpleVector<Timestamp>>()->valueAt(i);
     int64_t timestamp = ts.toMillis();
     if (windowType_ == WindowType::HOP) { // Hopping window
-      int64_t start = TimeWindowUtil::getWindowStartWithOffset(
-          timestamp, offset_, sliceSize_);
+      int64_t start = TimeWindowUtil::getWindowStartWithOffset(timestamp, offset_, sliceSize_);
       partitions[i] = start + sliceSize_;
     } else if (windowType_ == WindowType::TUMBLE) { // Windowed Slice Assigner
       partitions[i] = timestamp;
@@ -71,9 +74,9 @@ std::optional<int64_t> WindowPartitionFunction::partition(
 }
 
 std::optional<uint32_t> WindowPartitionFunction::partition(
-  const RowVector& /* input */,
-  std::vector<uint32_t>& /* partitions */) {
-  VELOX_NYI();
+    const RowVector& input,
+    std::vector<uint32_t>& partitions) {
+  return std::nullopt;
 }
 
 int64_t getTimestamp(
@@ -114,22 +117,15 @@ core::PartitionFunctionSpecPtr StreamWindowPartitionFunctionSpec::deserialize(
   auto size = obj["size"].asInt();
   auto step = obj["step"].asInt();
   auto offset = obj["offset"].asInt();
-  auto windowType = Window::getType(obj["windowType"].asInt());
+  auto windowType = obj["windowType"].asInt();
 
   return std::make_shared<StreamWindowPartitionFunctionSpec>(
       ISerializable::deserialize<RowType>(obj["inputType"]),
       rowtimeIndex,
       size,
-      step,
+      step, 
       offset,
-      windowType);
-}
-
-void registerPartitionFunctionSerDe() {
-  auto& registry = DeserializationWithContextRegistryForSharedPtr();
-  registry.Register(
-      "StreamWindowPartitionFunctionSpec",
-      StreamWindowPartitionFunctionSpec::deserialize);
+      static_cast<WindowType>(windowType));
 }
 
 } // namespace facebook::velox::stateful
diff --git a/velox/experimental/stateful/window/WindowPartitionFunction.h b/velox/experimental/stateful/window/WindowPartitionFunction.h
index 7064aefb311..f825f43f6a8 100644
--- a/velox/experimental/stateful/window/WindowPartitionFunction.h
+++ b/velox/experimental/stateful/window/WindowPartitionFunction.h
@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 #pragma once
-#include <experimental/stateful/window/Window.h>
+#include "velox/experimental/stateful/window/Window.h"
 #include <cstdint>
+
 #include "velox/core/PlanNode.h"
 
 namespace facebook::velox::stateful {
@@ -26,7 +27,7 @@ class WindowPartitionFunction : public core::PartitionFunction {
  public:
   WindowPartitionFunction(
       const RowTypePtr& inputType,
-      const column_index_t rowtimeIndex,
+      const int32_t rowtimeIndex,
       int64_t size,
       int64_t step,
       int64_t offset,
@@ -37,12 +38,12 @@ class WindowPartitionFunction : public core::PartitionFunction {
       std::vector<uint32_t>& partitions) override;
 
   std::optional<int64_t> partition(
-      const RowVector& input,
-      std::vector<int64_t>& partitions) override;
+     const RowVector& input,
+     std::vector<int64_t>& partitions) override;
 
  private:
   RowTypePtr inputType_;
-  column_index_t rowtimeIndex_;
+  int32_t rowtimeIndex_;
   int64_t size_;
   int64_t step_;
   int64_t offset_;
@@ -54,7 +55,7 @@ class StreamWindowPartitionFunctionSpec : public core::PartitionFunctionSpec {
  public:
   StreamWindowPartitionFunctionSpec(
       const RowTypePtr& inputType,
-      column_index_t rowtimeIndex,
+      int32_t rowtimeIndex,
       int64_t size,
       int64_t step,
       int64_t offset,
@@ -80,13 +81,11 @@ class StreamWindowPartitionFunctionSpec : public core::PartitionFunctionSpec {
 
  private:
   RowTypePtr inputType_;
-  column_index_t rowtimeIndex_;
+  int32_t rowtimeIndex_;
   int64_t size_;
   int64_t step_;
   int64_t offset_;
   WindowType windowType_;
 };
 
-void registerPartitionFunctionSerDe();
-
 } // namespace facebook::velox::stateful
diff --git a/velox/functions/flinksql/CMakeLists.txt b/velox/functions/flinksql/CMakeLists.txt
new file mode 100644
index 00000000000..e69de29bb2d