refactor(exec): Move window files under exec/window (#17710)

Summary:
Preparatory, mechanical refactor for #17558. The window subsystem lived as a flat set of files under `velox/exec/`. This moves the internal implementation files into a dedicated `velox/exec/window/` directory and `facebook::velox::exec::window` namespace, while keeping the public-facing files in `velox/exec/` and the `facebook::velox::exec` namespace.

**Stay in `velox/exec/` (`exec` namespace):**
- `Window.h`/`.cpp` — the operator itself, alongside `HashJoin`, `Aggregate`, and `TableScan`.
- `WindowFunction.h`/`.cpp` — the public API that window function authors implement against, like `AggregateFunction.h`.

**Move under `velox/exec/window/` (`exec::window` namespace):**
- `WindowBuild` and its variants, `WindowPartition`, `AggregateWindow`, `KRangeFrameBound`, `PeerGroupComputation`, `WindowPartitionAccessor` — internal implementation details.

There are no behavior changes; only file locations, the enclosing namespace of the internal files, and references to them change. The operator and the public API reference the internal types through the `window::` namespace.

Pull Request resolved: #17710

Reviewed By: apurva-meta

Differential Revision: D107420866

Pulled By: bikramSingh91

fbshipit-source-id: 7a9a306463fd5005d5b48db395d58afde2d32ea0

Author: zhli1142015

velox/exec/Aggregate.cpp

@@ -19,7 +19,7 @@
 #include <unordered_map>
 #include "velox/exec/AggregateCompanionAdapter.h"
 #include "velox/exec/AggregateCompanionSignatures.h"
-#include "velox/exec/AggregateWindow.h"
+#include "velox/exec/window/AggregateWindow.h"
 
 namespace facebook::velox::exec {
 
@@ -86,7 +86,7 @@ AggregateRegistrationResult registerAggregateFunction(
   // If the aggregate is not a companion function, also register it as a window
   // function.
   if (!metadata.companionFunction) {
-    registerAggregateWindowFunction(sanitizedName);
+    window::registerAggregateWindowFunction(sanitizedName);
   }
 
   // Register companion function if needed.

velox/exec/CMakeLists.txt

@@ -19,7 +19,6 @@ velox_add_library(
   AggregateCompanionSignatures.cpp
   AggregateFunctionRegistry.cpp
   AggregateInfo.cpp
-  AggregateWindow.cpp
   AggregationMasks.cpp
   ArrowStream.cpp
   AssignUniqueId.cpp
@@ -76,19 +75,15 @@ velox_add_library(
   OutputBufferManager.cpp
   ParallelProject.cpp
   PartitionFunction.cpp
-  PartitionStreamingWindowBuild.cpp
   PartitionedOutput.cpp
   PlanNodeStats.cpp
   PrefixSort.cpp
   ProbeOperatorState.cpp
-  SubPartitionedSortWindowBuild.cpp
   RowContainer.cpp
   RowNumber.cpp
-  RowsStreamingWindowBuild.cpp
   ScaleWriterLocalPartition.cpp
   ScaledScanController.cpp
   SortBuffer.cpp
-  SortWindowBuild.cpp
   SortedAggregations.cpp
   SpatialJoinBuild.cpp
   SpatialJoinProbe.cpp
@@ -111,7 +106,6 @@ velox_add_library(
   Values.cpp
   VectorHasher.cpp
   Window.cpp
-  WindowBuild.cpp
   WindowFunction.cpp
   WindowPartition.cpp
   HEADERS
@@ -122,7 +116,6 @@ velox_add_library(
   AggregateCompanionSignatures.h
   AggregateFunctionRegistry.h
   AggregateInfo.h
-  AggregateWindow.h
   AggregationMasks.h
   ArrowStream.h
   AssignUniqueId.h
@@ -156,7 +149,6 @@ velox_add_library(
   IndexLookupJoin.h
   IndexLookupJoinBridge.h
   JoinBridge.h
-  KRangeFrameBound.h
   Limit.h
   LocalPartition.h
   LocalPlanner.h
@@ -182,24 +174,20 @@ velox_add_library(
   OutputBuffer.h
   OutputBufferManager.h
   ParallelProject.h
-  PeerGroupComputation.h
   PartitionFunction.h
-  PartitionStreamingWindowBuild.h
   PartitionedOutput.h
   PlanNodeStats.h
   PrefixSort.h
   ProbeOperatorState.h
   RoundRobinPartitionFunction.h
   RowContainer.h
   RowNumber.h
-  RowsStreamingWindowBuild.h
   ScaleWriterLocalPartition.h
   ScaledScanController.h
   SerializedPage.h
   SetAccumulator.h
   SimpleAggregateAdapter.h
   SortBuffer.h
-  SortWindowBuild.h
   SortedAggregations.h
   SpatialIndex.h
   SpatialJoinBuild.h
@@ -211,7 +199,6 @@ velox_add_library(
   StreamingAggregation.h
   StreamingEnforceDistinct.h
   Strings.h
-  SubPartitionedSortWindowBuild.h
   TableScan.h
   TableWriteMerge.h
   TableWriter.h
@@ -229,10 +216,8 @@ velox_add_library(
   VectorHasher-inl.h
   VectorHasher.h
   Window.h
-  WindowBuild.h
   WindowFunction.h
   WindowPartition.h
-  WindowPartitionAccessor.h
 )
 
 velox_add_library(velox_exec_spill_stats SpillStats.cpp HEADERS SpillStats.h)
@@ -247,6 +232,7 @@ velox_link_libraries(
   velox_connector
   velox_connector_registry
   velox_exec_spill_stats
+  velox_exec_window
   velox_expression
   velox_file
   velox_presto_serializer
@@ -279,6 +265,7 @@ endif()
 add_subdirectory(prefixsort)
 add_subdirectory(rpc)
 add_subdirectory(trace)
+add_subdirectory(window)
 
 velox_add_library(velox_aggregate_util INTERFACE HEADERS AggregateUtil.h)
 

velox/exec/Window.cpp

@@ -16,11 +16,11 @@
 #include "velox/exec/Window.h"
 #include "velox/exec/OperatorType.h"
 #include "velox/exec/OperatorUtils.h"
-#include "velox/exec/PartitionStreamingWindowBuild.h"
-#include "velox/exec/RowsStreamingWindowBuild.h"
-#include "velox/exec/SortWindowBuild.h"
-#include "velox/exec/SubPartitionedSortWindowBuild.h"
 #include "velox/exec/Task.h"
+#include "velox/exec/window/PartitionStreamingWindowBuild.h"
+#include "velox/exec/window/RowsStreamingWindowBuild.h"
+#include "velox/exec/window/SortWindowBuild.h"
+#include "velox/exec/window/SubPartitionedSortWindowBuild.h"
 
 namespace facebook::velox::exec {
 
@@ -62,17 +62,17 @@ Window::Window(
   }
   if (windowNode->inputsSorted()) {
     if (supportRowsStreaming()) {
-      windowBuild_ = std::make_unique<RowsStreamingWindowBuild>(
+      windowBuild_ = std::make_unique<window::RowsStreamingWindowBuild>(
           windowNode_, pool(), spillConfig, &nonReclaimableSection_);
     } else {
-      windowBuild_ = std::make_unique<PartitionStreamingWindowBuild>(
+      windowBuild_ = std::make_unique<window::PartitionStreamingWindowBuild>(
           windowNode, pool(), spillConfig, &nonReclaimableSection_);
     }
   } else {
     if (auto numSubPartitions =
             operatorCtx_->driverCtx()->queryConfig().windowNumSubPartitions();
         numSubPartitions > 1) {
-      windowBuild_ = std::make_unique<SubPartitionedSortWindowBuild>(
+      windowBuild_ = std::make_unique<window::SubPartitionedSortWindowBuild>(
           windowNode,
           numSubPartitions,
           pool(),
@@ -82,7 +82,7 @@ Window::Window(
           &stats_,
           spillStats_.get());
     } else {
-      windowBuild_ = std::make_unique<SortWindowBuild>(
+      windowBuild_ = std::make_unique<window::SortWindowBuild>(
           windowNode,
           pool(),
           makePrefixSortConfig(driverCtx->queryConfig()),

velox/exec/Window.h

@@ -17,9 +17,9 @@
 
 #include "velox/exec/Operator.h"
 #include "velox/exec/RowContainer.h"
-#include "velox/exec/WindowBuild.h"
 #include "velox/exec/WindowFunction.h"
 #include "velox/exec/WindowPartition.h"
+#include "velox/exec/window/WindowBuild.h"
 
 namespace facebook::velox::exec {
 
@@ -171,7 +171,7 @@ class Window : public Operator {
 
   // WindowBuild is used to store input rows and return WindowPartitions
   // for the processing.
-  std::unique_ptr<WindowBuild> windowBuild_;
+  std::unique_ptr<window::WindowBuild> windowBuild_;
 
   // The cached window plan node used for window function initialization. It is
   // reset after the initialization.

velox/exec/WindowPartition.cpp

@@ -14,8 +14,8 @@
  * limitations under the License.
  */
 #include "velox/exec/WindowPartition.h"
-#include "velox/exec/KRangeFrameBound.h"
-#include "velox/exec/PeerGroupComputation.h"
+#include "velox/exec/window/KRangeFrameBound.h"
+#include "velox/exec/window/PeerGroupComputation.h"
 
 #include <algorithm>
 
@@ -287,7 +287,7 @@ std::pair<vector_size_t, vector_size_t> WindowPartition::computePeerBuffers(
     vector_size_t* rawPeerStarts,
     vector_size_t* rawPeerEnds) {
   RowContainerAccessor rows{*this};
-  auto result = PeerGroupComputation::compute(
+  auto result = window::PeerGroupComputation::compute(
       rows, start, end, prevPeerStart, prevPeerEnd, rawPeerStarts, rawPeerEnds);
   if (result.previousRowConsumed) {
     removePreviousRow();
@@ -308,7 +308,7 @@ void WindowPartition::computeKRangeFrameBounds(
   const auto frameType = data_->columnTypes()[inputMapping_[frameColumn]];
 
   RowContainerAccessor rows{*this};
-  KRangeFrameBound::compute(
+  window::KRangeFrameBound::compute(
       rows,
       isStartBound,
       isPreceding,

velox/exec/tests/WindowTest.cpp

@@ -21,11 +21,11 @@
 #include "velox/common/testutil/TempDirectoryPath.h"
 #include "velox/exec/OrderBy.h"
 #include "velox/exec/PlanNodeStats.h"
-#include "velox/exec/RowsStreamingWindowBuild.h"
-#include "velox/exec/SortWindowBuild.h"
 #include "velox/exec/tests/utils/AssertQueryBuilder.h"
 #include "velox/exec/tests/utils/OperatorTestBase.h"
 #include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/exec/window/RowsStreamingWindowBuild.h"
+#include "velox/exec/window/SortWindowBuild.h"
 #include "velox/functions/prestosql/window/WindowFunctionsRegistration.h"
 
 using namespace facebook::velox::exec::test;
@@ -38,7 +38,7 @@ class WindowTest : public OperatorTestBase {
  public:
   void SetUp() override {
     OperatorTestBase::SetUp();
-    window::prestosql::registerAllWindowFunctions();
+    velox::window::prestosql::registerAllWindowFunctions();
     filesystems::registerLocalFileSystem();
   }
 
@@ -348,9 +348,9 @@ DEBUG_ONLY_TEST_F(WindowTest, aggWindowResultMismatch) {
 
   std::atomic_bool isStreamCreated{false};
   SCOPED_TESTVALUE_SET(
-      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
-      std::function<void(RowsStreamingWindowBuild*)>(
-          [&](RowsStreamingWindowBuild* windowBuild) {
+      "facebook::velox::exec::window::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      std::function<void(window::RowsStreamingWindowBuild*)>(
+          [&](window::RowsStreamingWindowBuild* windowBuild) {
             isStreamCreated.store(true);
           }));
 
@@ -381,9 +381,9 @@ DEBUG_ONLY_TEST_F(WindowTest, rankRowStreamingWindowBuild) {
 
   std::atomic_bool isStreamCreated{false};
   SCOPED_TESTVALUE_SET(
-      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
-      std::function<void(RowsStreamingWindowBuild*)>(
-          [&](RowsStreamingWindowBuild* windowBuild) {
+      "facebook::velox::exec::window::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      std::function<void(window::RowsStreamingWindowBuild*)>(
+          [&](window::RowsStreamingWindowBuild* windowBuild) {
             isStreamCreated.store(true);
           }));
 
@@ -424,9 +424,9 @@ DEBUG_ONLY_TEST_F(WindowTest, valuesRowsStreamingWindowBuild) {
 
   std::atomic_bool isStreamCreated{false};
   SCOPED_TESTVALUE_SET(
-      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
-      std::function<void(RowsStreamingWindowBuild*)>(
-          [&](RowsStreamingWindowBuild* windowBuild) {
+      "facebook::velox::exec::window::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      std::function<void(window::RowsStreamingWindowBuild*)>(
+          [&](window::RowsStreamingWindowBuild* windowBuild) {
             isStreamCreated.store(true);
           }));
 
@@ -577,9 +577,9 @@ DEBUG_ONLY_TEST_F(WindowTest, aggregationWithNonDefaultFrame) {
 
   std::atomic_bool isStreamCreated{false};
   SCOPED_TESTVALUE_SET(
-      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
-      std::function<void(RowsStreamingWindowBuild*)>(
-          [&](RowsStreamingWindowBuild* windowBuild) {
+      "facebook::velox::exec::window::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      std::function<void(window::RowsStreamingWindowBuild*)>(
+          [&](window::RowsStreamingWindowBuild* windowBuild) {
             isStreamCreated.store(true);
           }));
 
@@ -610,9 +610,9 @@ DEBUG_ONLY_TEST_F(WindowTest, nonRowsStreamingWindow) {
 
   std::atomic_bool isStreamCreated{false};
   SCOPED_TESTVALUE_SET(
-      "facebook::velox::exec::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
-      std::function<void(RowsStreamingWindowBuild*)>(
-          [&](RowsStreamingWindowBuild* windowBuild) {
+      "facebook::velox::exec::window::RowsStreamingWindowBuild::RowsStreamingWindowBuild",
+      std::function<void(window::RowsStreamingWindowBuild*)>(
+          [&](window::RowsStreamingWindowBuild* windowBuild) {
             isStreamCreated.store(true);
           }));
 
@@ -889,7 +889,7 @@ DEBUG_ONLY_TEST_F(WindowTest, reserveMemorySort) {
         velox::common::PrefixSortConfig{
             std::numeric_limits<int32_t>::max(), 130, 12};
     folly::Synchronized<OperatorStats> opStats;
-    auto sortWindowBuild = std::make_unique<SortWindowBuild>(
+    auto sortWindowBuild = std::make_unique<window::SortWindowBuild>(
         plan,
         pool_.get(),
         std::move(prefixSortConfig),

velox/exec/AggregateWindow.cpp velox/exec/window/AggregateWindow.cppvelox/exec/AggregateWindow.cpp renamed to velox/exec/window/AggregateWindow.cpp

@@ -14,14 +14,14 @@
  * limitations under the License.
  */
 
-#include "velox/exec/AggregateWindow.h"
+#include "velox/exec/window/AggregateWindow.h"
 #include "velox/common/base/Exceptions.h"
 #include "velox/exec/Aggregate.h"
 #include "velox/exec/WindowFunction.h"
 #include "velox/expression/FunctionSignature.h"
 #include "velox/vector/FlatVector.h"
 
-namespace facebook::velox::exec {
+namespace facebook::velox::exec::window {
 
 namespace {
 
@@ -435,4 +435,4 @@ void registerAggregateWindowFunction(const std::string& name) {
         });
   }
 }
-} // namespace facebook::velox::exec
+} // namespace facebook::velox::exec::window

velox/exec/AggregateWindow.h velox/exec/window/AggregateWindow.hvelox/exec/AggregateWindow.h renamed to velox/exec/window/AggregateWindow.h

@@ -16,8 +16,8 @@
 #pragma once
 #include <string>
 
-namespace facebook::velox::exec {
+namespace facebook::velox::exec::window {
 
 void registerAggregateWindowFunction(const std::string& name);
 
-} // namespace facebook::velox::exec
+} // namespace facebook::velox::exec::window

velox/exec/window/CMakeLists.txt

@@ -0,0 +1,35 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+velox_add_library(
+  velox_exec_window
+  AggregateWindow.cpp
+  PartitionStreamingWindowBuild.cpp
+  RowsStreamingWindowBuild.cpp
+  SortWindowBuild.cpp
+  SubPartitionedSortWindowBuild.cpp
+  WindowBuild.cpp
+  HEADERS
+  AggregateWindow.h
+  KRangeFrameBound.h
+  PartitionStreamingWindowBuild.h
+  PeerGroupComputation.h
+  RowsStreamingWindowBuild.h
+  SortWindowBuild.h
+  SubPartitionedSortWindowBuild.h
+  WindowBuild.h
+  WindowPartitionAccessor.h
+)
+
+velox_link_libraries(velox_exec_window velox_exec velox_vector)