Add experimental.spiller-spill-path property

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

@@ -41,7 +41,6 @@
 import java.util.stream.Stream;
 
 import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
-import static com.facebook.presto.spiller.BinarySpillerFactory.SPILL_PATH;
 import static com.facebook.presto.util.ImmutableCollectors.toImmutableList;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;
@@ -56,12 +55,12 @@ public class BinaryFileSpiller
     private int spillsCount;
     private final ListeningExecutorService executor;
 
-    public BinaryFileSpiller(BlockEncodingSerde blockEncodingSerde, ListeningExecutorService executor)
+    public BinaryFileSpiller(BlockEncodingSerde blockEncodingSerde, ListeningExecutorService executor, Path spillPath)
     {
         this.blockEncodingSerde = requireNonNull(blockEncodingSerde, "blockEncodingSerde is null");
         this.executor = requireNonNull(executor, "executor is null");
         try {
-            this.targetDirectory = Files.createTempDirectory(SPILL_PATH, "presto-spill");
+            this.targetDirectory = Files.createTempDirectory(spillPath, "presto-spill");
         }
         catch (IOException e) {
             throw new PrestoException(GENERIC_INTERNAL_ERROR, "Failed to create spill directory", e);

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

@@ -16,12 +16,12 @@
 
 import com.facebook.presto.spi.block.BlockEncodingSerde;
 import com.facebook.presto.spi.type.Type;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.inject.Inject;
 
 import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.List;
 
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
@@ -31,28 +31,27 @@
 public class BinarySpillerFactory
         implements SpillerFactory
 {
-    public static final Path SPILL_PATH = Paths.get("/tmp/spills");
-
     private final ListeningExecutorService executor;
     private final BlockEncodingSerde blockEncodingSerde;
+    private final Path spillPath;
 
     @Inject
-    public BinarySpillerFactory(BlockEncodingSerde blockEncodingSerde)
+    public BinarySpillerFactory(BlockEncodingSerde blockEncodingSerde, FeaturesConfig featuresConfig)
     {
-        this(blockEncodingSerde, MoreExecutors.listeningDecorator(newFixedThreadPool(4, daemonThreadsNamed("binary-spiller-%s"))));
+        this(blockEncodingSerde, MoreExecutors.listeningDecorator(newFixedThreadPool(4, daemonThreadsNamed("binary-spiller-%s"))), featuresConfig);
     }
 
-    public BinarySpillerFactory(BlockEncodingSerde blockEncodingSerde, ListeningExecutorService executor)
+    public BinarySpillerFactory(BlockEncodingSerde blockEncodingSerde, ListeningExecutorService executor, FeaturesConfig featuresConfig)
     {
         this.blockEncodingSerde = requireNonNull(blockEncodingSerde, "blockEncodingSerde is null");
         this.executor = requireNonNull(executor, "executor is null");
-
-        SPILL_PATH.toFile().mkdirs();
+        this.spillPath = featuresConfig.getSpillerSpillPath();
+        this.spillPath.toFile().mkdirs();
     }
 
     @Override
     public Spiller create(List<Type> types)
     {
-        return new BinaryFileSpiller(blockEncodingSerde, executor);
+        return new BinaryFileSpiller(blockEncodingSerde, executor, spillPath);
     }
 }

presto-main/src/main/java/com/facebook/presto/sql/analyzer/FeaturesConfig.java

@@ -22,6 +22,8 @@
 
 import javax.validation.constraints.Min;
 
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.List;
 
 import static com.facebook.presto.sql.analyzer.RegexLibrary.JONI;
@@ -58,6 +60,7 @@ public static class ProcessingOptimization
     private RegexLibrary regexLibrary = JONI;
     private boolean spillEnabled;
     private DataSize operatorMemoryLimitBeforeSpill = new DataSize(4, DataSize.Unit.MEGABYTE);
+    private Path spillerSpillPath = Paths.get(System.getProperty("java.io.tmpdir"), "presto", "spills");
 
     public boolean isResourceGroupsEnabled()
     {
@@ -281,4 +284,16 @@ public FeaturesConfig setOperatorMemoryLimitBeforeSpill(DataSize operatorMemoryL
         this.operatorMemoryLimitBeforeSpill = operatorMemoryLimitBeforeSpill;
         return this;
     }
+
+    public Path getSpillerSpillPath()
+    {
+        return spillerSpillPath;
+    }
+
+    @Config("experimental.spiller-spill-path")
+    public FeaturesConfig setSpillerSpillPath(String spillPath)
+    {
+        this.spillerSpillPath = Paths.get(spillPath);
+        return this;
+    }
 }

presto-main/src/main/java/com/facebook/presto/testing/LocalQueryRunner.java

@@ -338,7 +338,7 @@ private LocalQueryRunner(Session defaultSession, FeaturesConfig featuresConfig,
                 .put(Rollback.class, new RollbackTask())
                 .build();
 
-        this.spillerFactory = new BinarySpillerFactory(blockEncodingSerde);
+        this.spillerFactory = new BinarySpillerFactory(blockEncodingSerde, featuresConfig);
     }
 
     public static LocalQueryRunner queryRunnerWithInitialTransaction(Session defaultSession)

presto-main/src/test/java/com/facebook/presto/execution/TaskTestUtils.java

@@ -34,6 +34,7 @@
 import com.facebook.presto.spiller.BinarySpillerFactory;
 import com.facebook.presto.split.PageSinkManager;
 import com.facebook.presto.split.PageSourceManager;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.sql.gen.ExpressionCompiler;
 import com.facebook.presto.sql.gen.JoinFilterFunctionCompiler;
 import com.facebook.presto.sql.parser.SqlParser;
@@ -128,7 +129,7 @@ public static LocalExecutionPlanner createTestingPlanner()
                 new IndexJoinLookupStats(),
                 new CompilerConfig(),
                 new TaskManagerConfig(),
-                new BinarySpillerFactory(new BlockEncodingManager(metadata.getTypeManager())));
+                new BinarySpillerFactory(new BlockEncodingManager(metadata.getTypeManager()), new FeaturesConfig()));
     }
 
     public static TaskInfo updateTask(SqlTask sqlTask, List<TaskSource> taskSources, OutputBuffers outputBuffers)

presto-main/src/test/java/com/facebook/presto/spiller/TestBinaryFileSpiller.java

@@ -20,6 +20,7 @@
 import com.facebook.presto.spi.block.BlockBuilderStatus;
 import com.facebook.presto.spi.block.BlockEncodingSerde;
 import com.facebook.presto.spi.type.Type;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.type.TypeRegistry;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -45,7 +46,7 @@ public class TestBinaryFileSpiller
     private static final List<Type> TYPES = ImmutableList.of(BIGINT, VARCHAR, DOUBLE, BIGINT);
     private final ListeningExecutorService executor = listeningDecorator(newSingleThreadScheduledExecutor());
     private final BlockEncodingSerde blockEncodingSerde = new BlockEncodingManager(new TypeRegistry(ImmutableSet.of(BIGINT, DOUBLE, VARBINARY)));
-    private final BinarySpillerFactory factory = new BinarySpillerFactory(blockEncodingSerde, executor);
+    private final BinarySpillerFactory factory = new BinarySpillerFactory(blockEncodingSerde, executor, new FeaturesConfig());
 
     @Test
     public void testFileSpiller()

presto-main/src/test/java/com/facebook/presto/sql/analyzer/TestFeaturesConfig.java

@@ -18,6 +18,7 @@
 import io.airlift.units.DataSize;
 import org.testng.annotations.Test;
 
+import java.nio.file.Paths;
 import java.util.Map;
 
 import static com.facebook.presto.sql.analyzer.FeaturesConfig.ProcessingOptimization.COLUMNAR_DICTIONARY;
@@ -51,7 +52,8 @@ public void testDefaults()
                 .setRe2JDfaStatesLimit(Integer.MAX_VALUE)
                 .setRe2JDfaRetries(5)
                 .setSpillEnabled(false)
-                .setOperatorMemoryLimitBeforeSpill(DataSize.valueOf("4MB")));
+                .setOperatorMemoryLimitBeforeSpill(DataSize.valueOf("4MB"))
+                .setSpillerSpillPath(Paths.get(System.getProperty("java.io.tmpdir"), "presto", "spills").toString()));
     }
 
     @Test
@@ -76,6 +78,7 @@ public void testExplicitPropertyMappings()
                 .put("re2j.dfa-retries", "42")
                 .put("experimental.spill-enabled", "true")
                 .put("experimental.operator-memory-limit-before-spill", "100MB")
+                .put("experimental.spiller-spill-path", "/tmp/custom/spill/path")
                 .build();
         Map<String, String> properties = new ImmutableMap.Builder<String, String>()
                 .put("experimental-syntax-enabled", "true")
@@ -96,6 +99,7 @@ public void testExplicitPropertyMappings()
                 .put("re2j.dfa-retries", "42")
                 .put("experimental.spill-enabled", "true")
                 .put("experimental.operator-memory-limit-before-spill", "100MB")
+                .put("experimental.spiller-spill-path", "/tmp/custom/spill/path")
                 .build();
 
         FeaturesConfig expected = new FeaturesConfig()
@@ -116,7 +120,8 @@ public void testExplicitPropertyMappings()
                 .setRe2JDfaStatesLimit(42)
                 .setRe2JDfaRetries(42)
                 .setSpillEnabled(true)
-                .setOperatorMemoryLimitBeforeSpill(DataSize.valueOf("100MB"));
+                .setOperatorMemoryLimitBeforeSpill(DataSize.valueOf("100MB"))
+                .setSpillerSpillPath("/tmp/custom/spill/path");
 
         assertFullMapping(properties, expected);
         assertDeprecatedEquivalence(FeaturesConfig.class, properties, propertiesLegacy);