Implement Window Function Spilling

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveDistributedWindowQueries.java

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+package com.facebook.presto.hive;
+
+import com.facebook.presto.tests.AbstractTestWindowQueries;
+
+import static com.facebook.presto.hive.HiveQueryRunner.createQueryRunner;
+import static io.airlift.tpch.TpchTable.getTables;
+
+public class TestHiveDistributedWindowQueries
+        extends AbstractTestWindowQueries
+{
+    public TestHiveDistributedWindowQueries()
+    {
+        super(() -> createQueryRunner(getTables()));
+    }
+}

presto-main/src/main/java/com/facebook/presto/operator/OperatorContext.java

@@ -264,6 +264,12 @@ public AggregatedMemoryContext aggregateUserMemoryContext()
         return new InternalAggregatedMemoryContext(operatorMemoryContext.aggregateUserMemoryContext(), memoryFuture, this::updatePeakMemoryReservations, false);
     }
 
+    // caller shouldn't close this context as it's managed by the OperatorContext
+    public AggregatedMemoryContext aggregateRevocableMemoryContext()
+    {
+        return new InternalAggregatedMemoryContext(operatorMemoryContext.aggregateRevocableMemoryContext(), memoryFuture, () -> {}, false);
+    }
+
     // caller should close this context as it's a new context
     public AggregatedMemoryContext newAggregateSystemMemoryContext()
     {

presto-main/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java

@@ -1009,7 +1009,10 @@ public PhysicalOperation visitWindow(WindowNode node, LocalExecutionPlanContext
                     sortOrder,
                     node.getPreSortedOrderPrefix(),
                     10_000,
-                    pagesIndexFactory);
+                    pagesIndexFactory,
+                    isSpillEnabled(session),
+                    spillerFactory,
+                    orderingCompiler);
 
             return new PhysicalOperation(operatorFactory, outputMappings.build(), context, source);
         }

presto-main/src/test/java/com/facebook/presto/operator/BenchmarkWindowOperator.java

@@ -115,7 +115,9 @@ private void createOperatorFactoryAndGenerateTestData(int numberOfPreGroupedColu
                         Ints.asList(),
                         Ints.asList(3),
                         ImmutableList.of(SortOrder.ASC_NULLS_LAST),
-                        0);
+                        0,
+                        new DummySpillerFactory(),
+                        false);
             }
             else if (numberOfPreGroupedColumns < NUMBER_OF_GROUP_COLUMNS) {
                 // Partially grouped
@@ -127,7 +129,9 @@ else if (numberOfPreGroupedColumns < NUMBER_OF_GROUP_COLUMNS) {
                         Ints.asList(1),
                         Ints.asList(3),
                         ImmutableList.of(SortOrder.ASC_NULLS_LAST),
-                        0);
+                        0,
+                        new DummySpillerFactory(),
+                        false);
             }
             else {
                 // Fully grouped and (potentially) sorted
@@ -139,7 +143,9 @@ else if (numberOfPreGroupedColumns < NUMBER_OF_GROUP_COLUMNS) {
                         Ints.asList(0, 1),
                         Ints.asList(3),
                         ImmutableList.of(SortOrder.ASC_NULLS_LAST),
-                        (numberOfPreGroupedColumns - NUMBER_OF_GROUP_COLUMNS));
+                        (numberOfPreGroupedColumns - NUMBER_OF_GROUP_COLUMNS),
+                        new DummySpillerFactory(),
+                        false);
             }
         }
 

presto-main/src/test/java/com/facebook/presto/operator/DummySpillerFactory.java

@@ -59,6 +59,7 @@ public List<Iterator<Page>> getSpills()
             @Override
             public void close()
             {
+                spills.clear();
             }
         };
     }

presto-main/src/test/java/com/facebook/presto/operator/OperatorAssertion.java

@@ -253,7 +253,17 @@ public static void assertOperatorEqualsIgnoreOrder(
             List<Page> input,
             MaterializedResult expected)
     {
-        assertOperatorEqualsIgnoreOrder(operatorFactory, driverContext, input, expected, false, Optional.empty());
+        assertOperatorEqualsIgnoreOrder(operatorFactory, driverContext, input, expected, false);
+    }
+
+    public static void assertOperatorEqualsIgnoreOrder(
+            OperatorFactory operatorFactory,
+            DriverContext driverContext,
+            List<Page> input,
+            MaterializedResult expected,
+            boolean revokeMemoryWhenAddingPages)
+    {
+        assertOperatorEqualsIgnoreOrder(operatorFactory, driverContext, input, expected, false, Optional.empty(), revokeMemoryWhenAddingPages);
     }
 
     public static void assertOperatorEqualsIgnoreOrder(