Spill large blocks to separate spill files in distinct aggregates

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -127,6 +127,8 @@ public final class SystemSessionProperties
     public static final String AGGREGATION_SPILL_ENABLED = "aggregation_spill_enabled";
     public static final String DISTINCT_AGGREGATION_SPILL_ENABLED = "distinct_aggregation_spill_enabled";
     public static final String DEDUP_BASED_DISTINCT_AGGREGATION_SPILL_ENABLED = "dedup_based_distinct_aggregation_spill_enabled";
+    public static final String DISTINCT_AGGREGATION_LARGE_BLOCK_SPILL_ENABLED = "distinct_aggregation_large_block_spill_enabled";
+    public static final String DISTINCT_AGGREGATION_LARGE_BLOCK_SIZE_THRESHOLD = "distinct_aggregation_large_block_size_threshold";
     public static final String ORDER_BY_AGGREGATION_SPILL_ENABLED = "order_by_aggregation_spill_enabled";
     public static final String WINDOW_SPILL_ENABLED = "window_spill_enabled";
     public static final String ORDER_BY_SPILL_ENABLED = "order_by_spill_enabled";
@@ -645,6 +647,20 @@ public SystemSessionProperties(
                         "Perform deduplication of input data for distinct aggregates before spilling",
                         featuresConfig.isDedupBasedDistinctAggregationSpillEnabled(),
                         false),
+                booleanProperty(
+                        DISTINCT_AGGREGATION_LARGE_BLOCK_SPILL_ENABLED,
+                        "Spill large block to a separate spill file",
+                        featuresConfig.isDistinctAggregationLargeBlockSpillEnabled(),
+                        false),
+                new PropertyMetadata<>(
+                        DISTINCT_AGGREGATION_LARGE_BLOCK_SIZE_THRESHOLD,
+                        "Block size threshold beyond which it will be spilled into a separate spill file",
+                        VARCHAR,
+                        DataSize.class,
+                        featuresConfig.getDistinctAggregationLargeBlockSizeThreshold(),
+                        false,
+                        value -> DataSize.valueOf((String) value),
+                        DataSize::toString),
                 booleanProperty(
                         ORDER_BY_AGGREGATION_SPILL_ENABLED,
                         "Enable spill for order-by aggregations if spill_enabled and aggregation_spill_enabled",
@@ -1558,6 +1574,16 @@ public static boolean isDedupBasedDistinctAggregationSpillEnabled(Session sessio
         return session.getSystemProperty(DEDUP_BASED_DISTINCT_AGGREGATION_SPILL_ENABLED, Boolean.class);
     }
 
+    public static boolean isDistinctAggregationLargeBlockSpillEnabled(Session session)
+    {
+        return session.getSystemProperty(DISTINCT_AGGREGATION_LARGE_BLOCK_SPILL_ENABLED, Boolean.class);
+    }
+
+    public static DataSize getDistinctAggregationLargeBlockSizeThreshold(Session session)
+    {
+        return session.getSystemProperty(DISTINCT_AGGREGATION_LARGE_BLOCK_SIZE_THRESHOLD, DataSize.class);
+    }
+
     public static boolean isOrderByAggregationSpillEnabled(Session session)
     {
         return session.getSystemProperty(ORDER_BY_AGGREGATION_SPILL_ENABLED, Boolean.class) && isAggregationSpillEnabled(session);

presto-main/src/main/java/com/facebook/presto/operator/aggregation/AccumulatorFactoryBinder.java

@@ -17,6 +17,7 @@
 import com.facebook.presto.common.block.SortOrder;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.operator.PagesIndex;
+import com.facebook.presto.spiller.StandaloneSpillerFactory;
 import com.facebook.presto.sql.gen.JoinCompiler;
 
 import java.util.List;
@@ -35,5 +36,6 @@ AccumulatorFactory bind(
             JoinCompiler joinCompiler,
             List<LambdaProvider> lambdaProviders,
             boolean spillEnabled,
-            Session session);
+            Session session,
+            StandaloneSpillerFactory standaloneSpillerFactory);
 }

presto-main/src/main/java/com/facebook/presto/operator/aggregation/GenericAccumulatorFactoryBinder.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.operator.PagesIndex;
 import com.facebook.presto.operator.aggregation.AggregationMetadata.AccumulatorStateDescriptor;
+import com.facebook.presto.spiller.StandaloneSpillerFactory;
 import com.facebook.presto.sql.gen.JoinCompiler;
 import com.google.common.annotations.VisibleForTesting;
 
@@ -71,7 +72,8 @@ public AccumulatorFactory bind(
             JoinCompiler joinCompiler,
             List<LambdaProvider> lambdaProviders,
             boolean spillEnabled,
-            Session session)
+            Session session,
+            StandaloneSpillerFactory standaloneSpillerFactory)
     {
         return new GenericAccumulatorFactory(
                 stateDescriptors,
@@ -87,7 +89,8 @@ public AccumulatorFactory bind(
                 joinCompiler,
                 session,
                 distinct,
-                spillEnabled);
+                spillEnabled,
+                standaloneSpillerFactory);
     }
 
     @VisibleForTesting

presto-main/src/main/java/com/facebook/presto/operator/aggregation/InternalAggregationFunction.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.common.type.RowType;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.operator.PagesIndex;
+import com.facebook.presto.spiller.StandaloneSpillerFactory;
 import com.facebook.presto.sql.gen.JoinCompiler;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -140,6 +141,7 @@ public AccumulatorFactory bind(List<Integer> inputChannels, Optional<Integer> ma
                 null,
                 ImmutableList.of(),
                 false,
+                null,
                 null);
     }
 
@@ -154,9 +156,10 @@ public AccumulatorFactory bind(
             JoinCompiler joinCompiler,
             List<LambdaProvider> lambdaProviders,
             boolean spillEnabled,
-            Session session)
+            Session session,
+            StandaloneSpillerFactory standaloneSpillerFactory)
     {
-        return factory.bind(inputChannels, maskChannel, sourceTypes, orderByChannels, orderings, pagesIndexFactory, distinct, joinCompiler, lambdaProviders, spillEnabled, session);
+        return factory.bind(inputChannels, maskChannel, sourceTypes, orderByChannels, orderings, pagesIndexFactory, distinct, joinCompiler, lambdaProviders, spillEnabled, session, standaloneSpillerFactory);
     }
 
     @VisibleForTesting

presto-main/src/main/java/com/facebook/presto/operator/aggregation/LazyAccumulatorFactoryBinder.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.common.block.SortOrder;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.operator.PagesIndex;
+import com.facebook.presto.spiller.StandaloneSpillerFactory;
 import com.facebook.presto.sql.gen.JoinCompiler;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Supplier;
@@ -54,7 +55,8 @@ public AccumulatorFactory bind(
             JoinCompiler joinCompiler,
             List<LambdaProvider> lambdaProviders,
             boolean spillEnabled,
-            Session session)
+            Session session,
+            StandaloneSpillerFactory standaloneSpillerFactory)
     {
         return binder.get().bind(
                 argumentChannels,
@@ -67,6 +69,7 @@ public AccumulatorFactory bind(
                 joinCompiler,
                 lambdaProviders,
                 spillEnabled,
-                session);
+                session,
+                standaloneSpillerFactory);
     }
 }

presto-main/src/main/java/com/facebook/presto/server/ServerMainModule.java

@@ -147,7 +147,9 @@
 import com.facebook.presto.spiller.SingleStreamSpillerFactory;
 import com.facebook.presto.spiller.SpillerFactory;
 import com.facebook.presto.spiller.SpillerStats;
+import com.facebook.presto.spiller.StandaloneSpillerFactory;
 import com.facebook.presto.spiller.TempStorageSingleStreamSpillerFactory;
+import com.facebook.presto.spiller.TempStorageStandaloneSpillerFactory;
 import com.facebook.presto.split.PageSinkManager;
 import com.facebook.presto.split.PageSinkProvider;
 import com.facebook.presto.split.PageSourceManager;
@@ -644,6 +646,7 @@ public ListeningExecutorService createResourceManagerExecutor(ResourceManagerCon
 
         // Spiller
         binder.bind(SpillerFactory.class).to(GenericSpillerFactory.class).in(Scopes.SINGLETON);
+        binder.bind(StandaloneSpillerFactory.class).to(TempStorageStandaloneSpillerFactory.class).in(Scopes.SINGLETON);
         binder.bind(PartitioningSpillerFactory.class).to(GenericPartitioningSpillerFactory.class).in(Scopes.SINGLETON);
         binder.bind(SpillerStats.class).in(Scopes.SINGLETON);
         newExporter(binder).export(SpillerFactory.class).withGeneratedName();

presto-main/src/main/java/com/facebook/presto/spiller/StandaloneSpiller.java

@@ -0,0 +1,45 @@
+/*
+ * 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.spiller;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.spi.storage.SerializedStorageHandle;
+
+import java.util.Iterator;
+
+/**
+ * StandaloneSpiller is a stateless spiller interface that provides
+ * basic spill capabilities using serialized storage handle.
+ * This is different from {@link Spiller} interface which is stateful
+ * in nature.
+ */
+public interface StandaloneSpiller
+{
+    /**
+     * Initiate spilling of pages stream. Returns back serialized
+     * storage handle which identifies the spill file
+     */
+    SerializedStorageHandle spill(Iterator<Page> pageIterator);
+
+    /**
+     * Returns list of pages present in spill file
+     * specified using serialized storage handle
+     */
+    Iterator<Page> getSpilledPages(SerializedStorageHandle storageHandle);
+
+    /**
+     * Remove the spill file using storage handle
+     */
+    void remove(SerializedStorageHandle storageHandle);
+}

presto-main/src/main/java/com/facebook/presto/spiller/StandaloneSpillerFactory.java

@@ -0,0 +1,21 @@
+/*
+ * 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.spiller;
+
+import com.facebook.presto.Session;
+
+public interface StandaloneSpillerFactory
+{
+    StandaloneSpiller create(Session session);
+}