[native] Make presto-on-spark native config up-to-date

Author: tanjialiang

presto-spark-base/src/main/java/com/facebook/presto/spark/PrestoSparkModule.java

@@ -122,7 +122,6 @@
 import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionNodeConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionSystemConfig;
-import com.facebook.presto.spark.execution.property.NativeExecutionVeloxConfig;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleReadInfo;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleWriteInfo;
 import com.facebook.presto.spark.execution.task.PrestoSparkNativeTaskExecutorFactory;
@@ -284,7 +283,6 @@ protected void setup(Binder binder)
         configBinder(binder).bindConfig(SessionPropertyProviderConfig.class);
         configBinder(binder).bindConfig(PrestoSparkConfig.class);
         configBinder(binder).bindConfig(TracingConfig.class);
-        configBinder(binder).bindConfig(NativeExecutionVeloxConfig.class);
         configBinder(binder).bindConfig(NativeExecutionSystemConfig.class);
         configBinder(binder).bindConfig(NativeExecutionNodeConfig.class);
         configBinder(binder).bindConfig(NativeExecutionConnectorConfig.class);

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/DetachedNativeExecutionProcess.java

@@ -47,7 +47,7 @@ public DetachedNativeExecutionProcess(
             ScheduledExecutorService errorRetryScheduledExecutor,
             JsonCodec<ServerInfo> serverInfoCodec,
             Duration maxErrorDuration,
-            WorkerProperty<?, ?, ?, ?> workerProperty)
+            WorkerProperty<?, ?, ?> workerProperty)
             throws IOException
     {
         super(executablePath,

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/DetachedNativeExecutionProcessFactory.java

@@ -40,15 +40,15 @@ public class DetachedNativeExecutionProcessFactory
     private final ExecutorService coreExecutor;
     private final ScheduledExecutorService errorRetryScheduledExecutor;
     private final JsonCodec<ServerInfo> serverInfoCodec;
-    private final WorkerProperty<?, ?, ?, ?> workerProperty;
+    private final WorkerProperty<?, ?, ?> workerProperty;
 
     @Inject
     public DetachedNativeExecutionProcessFactory(
             @ForNativeExecutionTask OkHttpClient httpClient,
             ExecutorService coreExecutor,
             ScheduledExecutorService errorRetryScheduledExecutor,
             JsonCodec<ServerInfo> serverInfoCodec,
-            WorkerProperty<?, ?, ?, ?> workerProperty,
+            WorkerProperty<?, ?, ?> workerProperty,
             FeaturesConfig featuresConfig)
     {
         super(httpClient, coreExecutor, errorRetryScheduledExecutor, serverInfoCodec, workerProperty, featuresConfig);

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/NativeExecutionModule.java

@@ -16,7 +16,6 @@
 import com.facebook.presto.spark.execution.property.NativeExecutionConnectorConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionNodeConfig;
 import com.facebook.presto.spark.execution.property.NativeExecutionSystemConfig;
-import com.facebook.presto.spark.execution.property.NativeExecutionVeloxConfig;
 import com.facebook.presto.spark.execution.property.PrestoSparkWorkerProperty;
 import com.facebook.presto.spark.execution.property.WorkerProperty;
 import com.facebook.presto.spark.execution.shuffle.PrestoSparkLocalShuffleInfoTranslator;
@@ -74,9 +73,9 @@ protected void bindShuffle(Binder binder)
 
     protected void bindWorkerProperties(Binder binder)
     {
-        newOptionalBinder(binder, new TypeLiteral<WorkerProperty<?, ?, ?, ?>>() {}).setDefault().to(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);
+        newOptionalBinder(binder, new TypeLiteral<WorkerProperty<?, ?, ?>>() {}).setDefault().to(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);
         if (connectorConfig.isPresent()) {
-            binder.bind(PrestoSparkWorkerProperty.class).toInstance(new PrestoSparkWorkerProperty(connectorConfig.get(), new NativeExecutionNodeConfig(), new NativeExecutionSystemConfig(), new NativeExecutionVeloxConfig()));
+            binder.bind(PrestoSparkWorkerProperty.class).toInstance(new PrestoSparkWorkerProperty(connectorConfig.get(), new NativeExecutionNodeConfig(), new NativeExecutionSystemConfig()));
         }
         else {
             binder.bind(PrestoSparkWorkerProperty.class).in(Scopes.SINGLETON);

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/NativeExecutionProcess.java

@@ -92,7 +92,7 @@ public class NativeExecutionProcess
     private final Executor executor;
     private final RequestErrorTracker errorTracker;
     private final OkHttpClient httpClient;
-    private final WorkerProperty<?, ?, ?, ?> workerProperty;
+    private final WorkerProperty<?, ?, ?> workerProperty;
 
     private volatile Process process;
     private volatile ProcessOutputPipe processOutputPipe;
@@ -106,7 +106,7 @@ public NativeExecutionProcess(
             ScheduledExecutorService scheduledExecutorService,
             JsonCodec<ServerInfo> serverInfoCodec,
             Duration maxErrorDuration,
-            WorkerProperty<?, ?, ?, ?> workerProperty)
+            WorkerProperty<?, ?, ?> workerProperty)
             throws IOException
     {
         this.executablePath = requireNonNull(executablePath, "executablePath is null");

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/nativeprocess/NativeExecutionProcessFactory.java

@@ -43,7 +43,7 @@ public class NativeExecutionProcessFactory
     private final ExecutorService coreExecutor;
     private final ScheduledExecutorService errorRetryScheduledExecutor;
     private final JsonCodec<ServerInfo> serverInfoCodec;
-    private final WorkerProperty<?, ?, ?, ?> workerProperty;
+    private final WorkerProperty<?, ?, ?> workerProperty;
     private final String executablePath;
     private final String programArguments;
 
@@ -55,7 +55,7 @@ public NativeExecutionProcessFactory(
             ExecutorService coreExecutor,
             ScheduledExecutorService errorRetryScheduledExecutor,
             JsonCodec<ServerInfo> serverInfoCodec,
-            WorkerProperty<?, ?, ?, ?> workerProperty,
+            WorkerProperty<?, ?, ?> workerProperty,
             FeaturesConfig featuresConfig)
     {
         this.httpClient = requireNonNull(httpClient, "httpClient is null");

presto-spark-base/src/main/java/com/facebook/presto/spark/execution/property/NativeExecutionSystemConfig.java

@@ -15,9 +15,11 @@
 
 import com.facebook.airlift.configuration.Config;
 import com.facebook.airlift.units.DataSize;
+import com.facebook.airlift.units.Duration;
 import com.google.common.collect.ImmutableMap;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import static java.util.Objects.requireNonNull;
 
@@ -34,7 +36,6 @@ public class NativeExecutionSystemConfig
     private static final String HTTP_SERVER_HTTP_PORT = "http-server.http.port";
     private static final String HTTP_SERVER_REUSE_PORT = "http-server.reuse-port";
     private static final String HTTP_SERVER_BIND_TO_NODE_INTERNAL_ADDRESS_ONLY_ENABLED = "http-server.bind-to-node-internal-address-only-enabled";
-    private static final String REGISTER_TEST_FUNCTIONS = "register-test-functions";
     // Number of I/O thread to use for serving http request on presto-native (proxygen server)
     // this excludes worker thread used by velox
     private static final String HTTP_SERVER_HTTPS_PORT = "http-server.https.port";
@@ -86,17 +87,12 @@ public class NativeExecutionSystemConfig
     // Presto-on-Spark, we set it to 'query-memory-gb' to allocate all the
     // memory arbitrator capacity to the query memory pool on its creation as
     // there is only one query running at a time.
-    private static final String MEMORY_POOL_INIT_CAPACITY = "memory-pool-init-capacity";
+    private static final String SHARED_ARBITRATOR_MEMORY_POOL_INITIAL_CAPACITY = "shared-arbitrator.memory-pool-initial-capacity";
     // Set the reserved memory capacity when we create a query memory pool. For
     // Presto-on-Spark, we set this to zero as there is only one query running
     // at a time.
-    private static final String MEMORY_POOL_RESERVED_CAPACITY = "memory-pool-reserved-capacity";
-    // Set the minimal memory capacity transfer between memory pools under
-    // memory arbitration. For Presto-on-Spark, there is only one query running
-    // so this specified how much memory to reclaim from a query when it runs
-    // out of memory.
-    private static final String MEMORY_POOL_TRANSFER_CAPACITY = "memory-pool-transfer-capacity";
-    private static final String MEMORY_RECLAIM_WAIT_MS = "memory-reclaim-wait-ms";
+    private static final String SHARED_ARBITRATOR_MEMORY_POOL_RESERVED_CAPACITY = "shared-arbitrator.memory-pool-reserved-capacity";
+    private static final String SHARED_ARBITRATOR_MAX_MEMORY_ARBITRATION_TIME = "shared-arbitrator.max-memory-arbitration-time";
     // Spilling related configs.
     private static final String SPILLER_SPILL_PATH = "experimental.spiller-spill-path";
     private static final String TASK_MAX_DRIVERS_PER_TASK = "task.max-drivers-per-task";
@@ -140,22 +136,19 @@ public class NativeExecutionSystemConfig
     // spilling and cache prefetch.
     private DataSize queryMemoryGb = new DataSize(8, DataSize.Unit.GIGABYTE);
 
-    private DataSize queryReservedMemoryGb = new DataSize(0, DataSize.Unit.GIGABYTE);
     private boolean useMmapAllocator = true;
     private String memoryArbitratorKind = "SHARED";
     private int memoryArbitratorCapacityGb = 8;
     private DataSize sharedArbitratorReservedCapacity = new DataSize(0, DataSize.Unit.GIGABYTE);
-    private long memoryPoolInitCapacity = 8L << 30;
-    private long memoryPoolReservedCapacity;
-    private long memoryPoolTransferCapacity = 2L << 30;
-    private long memoryReclaimWaitMs = 300_000;
+    private DataSize sharedArbitratorMemoryPoolInitialCapacity = new DataSize(4, DataSize.Unit.GIGABYTE);
+    private DataSize sharedArbitratorMemoryPoolReservedCapacity = new DataSize(32, DataSize.Unit.MEGABYTE);
+    private Duration sharedArbitratorMaxMemoryArbitrationTime = new Duration(5, TimeUnit.MINUTES);
     private String spillerSpillPath = "";
     private int concurrentLifespansPerTask = 5;
     private int maxDriversPerTask = 15;
     private boolean enableOldTaskCleanUp; // false;
     private String prestoVersion = "dummy.presto.version";
     private String shuffleName = "local";
-    private boolean registerTestFunctions;
     private boolean enableHttpServerAccessLog = true;
     private boolean coreOnAllocationFailureEnabled;
     private boolean spillEnabled = true;
@@ -164,6 +157,18 @@ public class NativeExecutionSystemConfig
     private boolean orderBySpillEnabled = true;
     private Long maxSpillBytes = 600L << 30;
 
+    // TODO: Deprecate following configs
+    private static final String REGISTER_TEST_FUNCTIONS = "register-test-functions";
+    private static final String MEMORY_POOL_INIT_CAPACITY = "memory-pool-init-capacity";
+    private static final String MEMORY_POOL_RESERVED_CAPACITY = "memory-pool-reserved-capacity";
+    private static final String MEMORY_POOL_TRANSFER_CAPACITY = "memory-pool-transfer-capacity";
+    private static final String MEMORY_RECLAIM_WAIT_MS = "memory-reclaim-wait-ms";
+    private boolean registerTestFunctions;
+    private long memoryPoolInitCapacity;
+    private long memoryPoolReservedCapacity;
+    private long memoryPoolTransferCapacity;
+    private long memoryReclaimWaitMs;
+
     public Map<String, String> getAllProperties()
     {
         ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
@@ -174,7 +179,6 @@ public Map<String, String> getAllProperties()
                 .put(HTTP_SERVER_HTTP_PORT, String.valueOf(getHttpServerPort()))
                 .put(HTTP_SERVER_REUSE_PORT, String.valueOf(isHttpServerReusePort()))
                 .put(HTTP_SERVER_BIND_TO_NODE_INTERNAL_ADDRESS_ONLY_ENABLED, String.valueOf(isHttpServerBindToNodeInternalAddressOnlyEnabled()))
-                .put(REGISTER_TEST_FUNCTIONS, String.valueOf(isRegisterTestFunctions()))
                 .put(HTTP_SERVER_HTTPS_PORT, String.valueOf(getHttpsServerPort()))
                 .put(HTTP_SERVER_HTTPS_ENABLED, String.valueOf(isEnableHttpsCommunication()))
                 .put(HTTPS_CIPHERS, String.valueOf(getHttpsCiphers()))
@@ -193,10 +197,9 @@ public Map<String, String> getAllProperties()
                 .put(MEMORY_ARBITRATOR_KIND, String.valueOf(getMemoryArbitratorKind()))
                 .put(MEMORY_ARBITRATOR_CAPACITY_GB, String.valueOf(getMemoryArbitratorCapacityGb()))
                 .put(SHARED_ARBITRATOR_RESERVED_CAPACITY, String.valueOf(getSharedArbitratorReservedCapacity()))
-                .put(MEMORY_POOL_INIT_CAPACITY, String.valueOf(getMemoryPoolInitCapacity()))
-                .put(MEMORY_POOL_RESERVED_CAPACITY, String.valueOf(getMemoryPoolReservedCapacity()))
-                .put(MEMORY_POOL_TRANSFER_CAPACITY, String.valueOf(getMemoryPoolTransferCapacity()))
-                .put(MEMORY_RECLAIM_WAIT_MS, String.valueOf(getMemoryReclaimWaitMs()))
+                .put(SHARED_ARBITRATOR_MEMORY_POOL_INITIAL_CAPACITY, String.valueOf(getSharedArbitratorMemoryPoolInitialCapacity()))
+                .put(SHARED_ARBITRATOR_MEMORY_POOL_RESERVED_CAPACITY, String.valueOf(getSharedArbitratorMemoryPoolReservedCapacity()))
+                .put(SHARED_ARBITRATOR_MAX_MEMORY_ARBITRATION_TIME, String.valueOf(getSharedArbitratorMaxMemoryArbitrationTime()))
                 .put(SPILLER_SPILL_PATH, String.valueOf(getSpillerSpillPath()))
                 .put(TASK_MAX_DRIVERS_PER_TASK, String.valueOf(getMaxDriversPerTask()))
                 .put(ENABLE_OLD_TASK_CLEANUP, String.valueOf(getOldTaskCleanupMs()))
@@ -208,6 +211,11 @@ public Map<String, String> getAllProperties()
                 .put(JOIN_SPILL_ENABLED, String.valueOf(getJoinSpillEnabled()))
                 .put(ORDER_BY_SPILL_ENABLED, String.valueOf(getOrderBySpillEnabled()))
                 .put(MAX_SPILL_BYTES, String.valueOf(getMaxSpillBytes()))
+                .put(REGISTER_TEST_FUNCTIONS, String.valueOf(registerTestFunctions))
+                .put(MEMORY_POOL_INIT_CAPACITY, String.valueOf(memoryPoolInitCapacity))
+                .put(MEMORY_POOL_RESERVED_CAPACITY, String.valueOf(memoryPoolReservedCapacity))
+                .put(MEMORY_POOL_TRANSFER_CAPACITY, String.valueOf(memoryPoolTransferCapacity))
+                .put(MEMORY_RECLAIM_WAIT_MS, String.valueOf(memoryReclaimWaitMs))
                 .build();
     }
 
@@ -295,18 +303,6 @@ public NativeExecutionSystemConfig setHttpServerBindToNodeInternalAddressOnlyEna
         return this;
     }
 
-    @Config(REGISTER_TEST_FUNCTIONS)
-    public NativeExecutionSystemConfig setRegisterTestFunctions(boolean registerTestFunctions)
-    {
-        this.registerTestFunctions = registerTestFunctions;
-        return this;
-    }
-
-    public boolean isRegisterTestFunctions()
-    {
-        return registerTestFunctions;
-    }
-
     @Config(HTTP_SERVER_NUM_IO_THREADS_HW_MULTIPLIER)
     public NativeExecutionSystemConfig setHttpServerNumIoThreadsHwMultiplier(double httpServerNumIoThreadsHwMultiplier)
     {
@@ -511,52 +507,40 @@ public DataSize getSharedArbitratorReservedCapacity()
         return sharedArbitratorReservedCapacity;
     }
 
-    @Config(MEMORY_POOL_INIT_CAPACITY)
-    public NativeExecutionSystemConfig setMemoryPoolInitCapacity(long memoryPoolInitCapacity)
-    {
-        this.memoryPoolInitCapacity = memoryPoolInitCapacity;
-        return this;
-    }
-
-    public long getMemoryPoolInitCapacity()
+    @Config(SHARED_ARBITRATOR_MEMORY_POOL_INITIAL_CAPACITY)
+    public NativeExecutionSystemConfig setSharedArbitratorMemoryPoolInitialCapacity(DataSize sharedArbitratorMemoryPoolInitialCapacity)
     {
-        return memoryPoolInitCapacity;
-    }
-
-    @Config(MEMORY_POOL_RESERVED_CAPACITY)
-    public NativeExecutionSystemConfig setMemoryPoolReservedCapacity(long memoryPoolReservedCapacity)
-    {
-        this.memoryPoolReservedCapacity = memoryPoolReservedCapacity;
+        this.sharedArbitratorMemoryPoolInitialCapacity = sharedArbitratorMemoryPoolInitialCapacity;
         return this;
     }
 
-    public long getMemoryPoolReservedCapacity()
+    public DataSize getSharedArbitratorMemoryPoolInitialCapacity()
     {
-        return memoryPoolReservedCapacity;
+        return sharedArbitratorMemoryPoolInitialCapacity;
     }
 
-    @Config(MEMORY_POOL_TRANSFER_CAPACITY)
-    public NativeExecutionSystemConfig setMemoryPoolTransferCapacity(long memoryPoolTransferCapacity)
+    @Config(SHARED_ARBITRATOR_MEMORY_POOL_RESERVED_CAPACITY)
+    public NativeExecutionSystemConfig setSharedArbitratorMemoryPoolReservedCapacity(DataSize sharedArbitratorMemoryPoolReservedCapacity)
     {
-        this.memoryPoolTransferCapacity = memoryPoolTransferCapacity;
+        this.sharedArbitratorMemoryPoolReservedCapacity = sharedArbitratorMemoryPoolReservedCapacity;
         return this;
     }
 
-    public long getMemoryPoolTransferCapacity()
+    public DataSize getSharedArbitratorMemoryPoolReservedCapacity()
     {
-        return memoryPoolTransferCapacity;
+        return sharedArbitratorMemoryPoolReservedCapacity;
     }
 
-    @Config(MEMORY_RECLAIM_WAIT_MS)
-    public NativeExecutionSystemConfig setMemoryReclaimWaitMs(long memoryReclaimWaitMs)
+    @Config(SHARED_ARBITRATOR_MAX_MEMORY_ARBITRATION_TIME)
+    public NativeExecutionSystemConfig setSharedArbitratorMaxMemoryArbitrationTime(Duration sharedArbitratorMaxMemoryArbitrationTime)
     {
-        this.memoryReclaimWaitMs = memoryReclaimWaitMs;
+        this.sharedArbitratorMaxMemoryArbitrationTime = sharedArbitratorMaxMemoryArbitrationTime;
         return this;
     }
 
-    public long getMemoryReclaimWaitMs()
+    public Duration getSharedArbitratorMaxMemoryArbitrationTime()
     {
-        return memoryReclaimWaitMs;
+        return sharedArbitratorMaxMemoryArbitrationTime;
     }
 
     @Config(SPILLER_SPILL_PATH)
@@ -702,4 +686,67 @@ public NativeExecutionSystemConfig setMaxSpillBytes(Long maxSpillBytes)
         this.maxSpillBytes = maxSpillBytes;
         return this;
     }
+
+    // TODO: All following configs are deprecated and will be removed after references from all
+    //  other systems are cleared.
+
+    @Config(REGISTER_TEST_FUNCTIONS)
+    public NativeExecutionSystemConfig setRegisterTestFunctions(boolean registerTestFunctions)
+    {
+        this.registerTestFunctions = registerTestFunctions;
+        return this;
+    }
+
+    public boolean isRegisterTestFunctions()
+    {
+        return registerTestFunctions;
+    }
+
+    @Config(MEMORY_POOL_INIT_CAPACITY)
+    public NativeExecutionSystemConfig setMemoryPoolInitCapacity(long memoryPoolInitCapacity)
+    {
+        this.memoryPoolInitCapacity = memoryPoolInitCapacity;
+        return this;
+    }
+
+    public long getMemoryPoolInitCapacity()
+    {
+        return memoryPoolInitCapacity;
+    }
+
+    @Config(MEMORY_POOL_RESERVED_CAPACITY)
+    public NativeExecutionSystemConfig setMemoryPoolReservedCapacity(long memoryPoolReservedCapacity)
+    {
+        this.memoryPoolReservedCapacity = memoryPoolReservedCapacity;
+        return this;
+    }
+
+    public long getMemoryPoolReservedCapacity()
+    {
+        return memoryPoolReservedCapacity;
+    }
+
+    @Config(MEMORY_POOL_TRANSFER_CAPACITY)
+    public NativeExecutionSystemConfig setMemoryPoolTransferCapacity(long memoryPoolTransferCapacity)
+    {
+        this.memoryPoolTransferCapacity = memoryPoolTransferCapacity;
+        return this;
+    }
+
+    public long getMemoryPoolTransferCapacity()
+    {
+        return memoryPoolTransferCapacity;
+    }
+
+    @Config(MEMORY_RECLAIM_WAIT_MS)
+    public NativeExecutionSystemConfig setMemoryReclaimWaitMs(long memoryReclaimWaitMs)
+    {
+        this.memoryReclaimWaitMs = memoryReclaimWaitMs;
+        return this;
+    }
+
+    public long getMemoryReclaimWaitMs()
+    {
+        return memoryReclaimWaitMs;
+    }
 }