[Sapphire Velox] presto side SV & Prestissimo config reconciliation

Summary:
Reconciles Sapphire-Velox and Prestissimo configurations by cleaning up `NativeExecutionSystemConfig.java`. Removes production-related configs (memory management, spill, shuffle, etc.) that are now defined in configerator's `native/config_properties.cinc`. Keeps only test-specific configs (system memory pushback, experimental spiller path, etc.) in the Java class.

**Rationale**: Production configs belong in configerator for centralized configuration management. Java configs should only contain test-specific properties needed for local development and testing.

# Release Notes
```
== NO RELEASE NOTE ==
```

Differential Revision: D85779158

Author: han-yan01

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

@@ -411,9 +411,9 @@ protected void updateWorkerMemoryProperties()
         DataSize offHeapMemoryBytes = DataSize.succinctDataSize(
                 conf.getSizeAsBytes(NATIVE_PROCESS_MEMORY_SPARK_CONF_NAME), BYTE);
         DataSize currentSystemMemory = DataSize.valueOf(workerProperty.getSystemConfig().getAllProperties()
-                .get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB) + GIGABYTE.getUnitString());
+                .get("system-memory-gb") + GIGABYTE.getUnitString());
         DataSize currentQueryMemory = DataSize.valueOf(workerProperty.getSystemConfig().getAllProperties()
-                .get(NativeExecutionSystemConfig.QUERY_MEMORY_GB) + GIGABYTE.getUnitString());
+                .get("query-memory-gb") + GIGABYTE.getUnitString());
         if (offHeapMemoryBytes.toBytes() == 0
                 || currentSystemMemory.toBytes() == 0
                 || offHeapMemoryBytes.toBytes() < currentSystemMemory.toBytes()) {
@@ -440,13 +440,13 @@ protected void updateWorkerMemoryProperties()
                 newQueryMemoryBytes);
 
         workerProperty.getSystemConfig()
-                .update(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB,
+                .update("system-memory-gb",
                         String.valueOf((int) newSystemMemory.getValue(GIGABYTE)));
         workerProperty.getSystemConfig()
-                .update(NativeExecutionSystemConfig.QUERY_MEMORY_GB,
+                .update("query-memory-gb",
                         String.valueOf((int) newQueryMemoryBytes.getValue(GIGABYTE)));
         workerProperty.getSystemConfig()
-                .update(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE,
+                .update("query.max-memory-per-node",
                         newQueryMemoryBytes.convertTo(GIGABYTE).toString());
     }
 

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

@@ -54,25 +54,18 @@ public class NativeExecutionSystemConfig
     public static final String CONNECTOR_NUM_IO_THREADS_HW_MULTIPLIER = "connector.num-io-threads-hw-multiplier";
     public static final String PRESTO_VERSION = "presto.version";
     public static final String SHUTDOWN_ONSET_SEC = "shutdown-onset-sec";
-    public static final String SYSTEM_MEMORY_GB = "system-memory-gb";
-    public static final String QUERY_MEMORY_GB = "query-memory-gb";
-    public static final String QUERY_MAX_MEMORY_PER_NODE = "query.max-memory-per-node";
-    public static final String USE_MMAP_ALLOCATOR = "use-mmap-allocator";
-    public static final String MEMORY_ARBITRATOR_KIND = "memory-arbitrator-kind";
-    public static final String SHARED_ARBITRATOR_RESERVED_CAPACITY = "shared-arbitrator.reserved-capacity";
-    public static final String SHARED_ARBITRATOR_MEMORY_POOL_INITIAL_CAPACITY = "shared-arbitrator.memory-pool-initial-capacity";
-    public static final String SHARED_ARBITRATOR_MAX_MEMORY_ARBITRATION_TIME = "shared-arbitrator.max-memory-arbitration-time";
+    // System memory pushback configs - kept for testing purposes only
+    // Production configs are in native/config_properties.cinc
+    public static final String SYSTEM_MEM_PUSHBACK_ENABLED = "system-mem-pushback-enabled";
+    public static final String SYSTEM_MEM_PUSHBACK_ABORT_ENABLED = "system-mem-pushback-abort-enabled";
+    public static final String SYSTEM_MEM_LIMIT_GB = "system-mem-limit-gb";
+    public static final String SYSTEM_MEM_SHRINK_GB = "system-mem-shrink-gb";
+    public static final String WORKER_OVERLOADED_THRESHOLD_MEM_GB = "worker-overloaded-threshold-mem-gb";
+    public static final String WORKER_OVERLOADED_THRESHOLD_CPU_PCT = "worker-overloaded-threshold-cpu-pct";
     public static final String EXPERIMENTAL_SPILLER_SPILL_PATH = "experimental.spiller-spill-path";
-    public static final String TASK_MAX_DRIVERS_PER_TASK = "task.max-drivers-per-task";
     public static final String ENABLE_OLD_TASK_CLEANUP = "enable-old-task-cleanup";
-    public static final String SHUFFLE_NAME = "shuffle.name";
     public static final String HTTP_SERVER_ENABLE_ACCESS_LOG = "http-server.enable-access-log";
     public static final String CORE_ON_ALLOCATION_FAILURE_ENABLED = "core-on-allocation-failure-enabled";
-    public static final String SPILL_ENABLED = "spill-enabled";
-    public static final String AGGREGATION_SPILL_ENABLED = "aggregation-spill-enabled";
-    public static final String JOIN_SPILL_ENABLED = "join-spill-enabled";
-    public static final String ORDER_BY_SPILL_ENABLED = "order-by-spill-enabled";
-    public static final String MAX_SPILL_BYTES = "max-spill-bytes";
     public static final String REMOTE_FUNCTION_SERVER_THRIFT_UDS_PATH = "remote-function-server.thrift.uds-path";
     public static final String REMOTE_FUNCTION_SERVER_SIGNATURE_FILES_DIRECTORY_PATH = "remote-function-server.signature.files.directory.path";
     public static final String REMOTE_FUNCTION_SERVER_SERDE = "remote-function-server.serde";
@@ -100,25 +93,16 @@ public class NativeExecutionSystemConfig
     private final String connectorNumIoThreadsHwMultiplierDefault = "0";
     private final String prestoVersionDefault = "dummy.presto.version";
     private final String shutdownOnsetSecDefault = "10";
-    private final String systemMemoryGbDefault = "10";
-    private final String queryMemoryGbDefault = "8";
-    private final String queryMaxMemoryPerNodeDefault = "8GB";
-    private final String useMmapAllocatorDefault = "true";
-    private final String memoryArbitratorKindDefault = "SHARED";
-    private final String sharedArbitratorReservedCapacityDefault = "0GB";
-    private final String sharedArbitratorMemoryPoolInitialCapacityDefault = "4GB";
-    private final String sharedArbitratorMaxMemoryArbitrationTimeDefault = "5m";
+    private final String systemMemPushbackEnabledDefault = "true";
+    private final String systemMemPushbackAbortEnabledDefault = "true";
+    private final String systemMemLimitGbDefault = "7";
+    private final String systemMemShrinkGbDefault = "1";
+    private final String workerOverloadedThresholdMemGbDefault = "6";
+    private final String workerOverloadedThresholdCpuPctDefault = "85";
     private final String experimentalSpillerSpillPathDefault = "";
-    private final String taskMaxDriversPerTaskDefault = "15";
     private final String enableOldTaskCleanupDefault = "false";
-    private final String shuffleNameDefault = "local";
     private final String httpServerEnableAccessLogDefault = "true";
     private final String coreOnAllocationFailureEnabledDefault = "false";
-    private final String spillEnabledDefault = "true";
-    private final String aggregationSpillEnabledDefault = "true";
-    private final String joinSpillEnabledDefault = "true";
-    private final String orderBySpillEnabledDefault = "true";
-    private final String maxSpillBytesDefault = String.valueOf(600L << 30);
 
     private final Map<String, String> systemConfigs;
     private final Map<String, String> defaultSystemConfigs;
@@ -163,27 +147,16 @@ public NativeExecutionSystemConfig(
                 .put(CONNECTOR_NUM_IO_THREADS_HW_MULTIPLIER, connectorNumIoThreadsHwMultiplierDefault)
                 .put(PRESTO_VERSION, prestoVersionDefault)
                 .put(SHUTDOWN_ONSET_SEC, shutdownOnsetSecDefault)
-                .put(SYSTEM_MEMORY_GB, systemMemoryGbDefault)
-                .put(QUERY_MEMORY_GB, queryMemoryGbDefault)
-                .put(QUERY_MAX_MEMORY_PER_NODE, queryMaxMemoryPerNodeDefault)
-                .put(USE_MMAP_ALLOCATOR, useMmapAllocatorDefault)
-                .put(MEMORY_ARBITRATOR_KIND, memoryArbitratorKindDefault)
-                .put(SHARED_ARBITRATOR_RESERVED_CAPACITY, sharedArbitratorReservedCapacityDefault)
-                .put(SHARED_ARBITRATOR_MEMORY_POOL_INITIAL_CAPACITY,
-                    sharedArbitratorMemoryPoolInitialCapacityDefault)
-                .put(SHARED_ARBITRATOR_MAX_MEMORY_ARBITRATION_TIME,
-                    sharedArbitratorMaxMemoryArbitrationTimeDefault)
+                .put(SYSTEM_MEM_PUSHBACK_ENABLED, systemMemPushbackEnabledDefault)
+                .put(SYSTEM_MEM_PUSHBACK_ABORT_ENABLED, systemMemPushbackAbortEnabledDefault)
+                .put(SYSTEM_MEM_LIMIT_GB, systemMemLimitGbDefault)
+                .put(SYSTEM_MEM_SHRINK_GB, systemMemShrinkGbDefault)
+                .put(WORKER_OVERLOADED_THRESHOLD_MEM_GB, workerOverloadedThresholdMemGbDefault)
+                .put(WORKER_OVERLOADED_THRESHOLD_CPU_PCT, workerOverloadedThresholdCpuPctDefault)
                 .put(EXPERIMENTAL_SPILLER_SPILL_PATH, experimentalSpillerSpillPathDefault)
-                .put(TASK_MAX_DRIVERS_PER_TASK, taskMaxDriversPerTaskDefault)
                 .put(ENABLE_OLD_TASK_CLEANUP, enableOldTaskCleanupDefault)
-                .put(SHUFFLE_NAME, shuffleNameDefault)
                 .put(HTTP_SERVER_ENABLE_ACCESS_LOG, httpServerEnableAccessLogDefault)
                 .put(CORE_ON_ALLOCATION_FAILURE_ENABLED, coreOnAllocationFailureEnabledDefault)
-                .put(SPILL_ENABLED, spillEnabledDefault)
-                .put(AGGREGATION_SPILL_ENABLED, aggregationSpillEnabledDefault)
-                .put(JOIN_SPILL_ENABLED, joinSpillEnabledDefault)
-                .put(ORDER_BY_SPILL_ENABLED, orderBySpillEnabledDefault)
-                .put(MAX_SPILL_BYTES, maxSpillBytesDefault)
                 .build();
     }
 

presto-spark-base/src/test/java/com/facebook/presto/spark/execution/TestNativeExecutionProcess.java

@@ -104,9 +104,9 @@ public void testUpdateWorkerMemoryPropertiesWithoutSparkEnv()
         process.updateWorkerMemoryProperties();
         // Verify that values remain unchanged
         Map<String, String> properties = process.getWorkerProperty().getSystemConfig().getAllProperties();
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "10");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "8");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "8GB");
+        assertEquals(properties.get("system-memory-gb"), "10");
+        assertEquals(properties.get("query-memory-gb"), "8");
+        assertEquals(properties.get("query.max-memory-per-node"), "8GB");
     }
 
     @Test
@@ -125,9 +125,9 @@ public void testUpdateWorkerMemoryPropertiesWithOffHeapMemory()
         // newSystemMemory = 20GB
         // queryMemoryFraction = 8/10 = 0.8
         // newQueryMemory = 20 * 0.8 = 16GB
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "20");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "16");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "16GB");
+        assertEquals(properties.get("system-memory-gb"), "20");
+        assertEquals(properties.get("query-memory-gb"), "16");
+        assertEquals(properties.get("query.max-memory-per-node"), "16GB");
     }
 
     @Test
@@ -142,9 +142,9 @@ public void testUpdateWorkerMemoryPropertiesWithoutOffHeapSetting()
 
         // Verify that values remain unchanged
         Map<String, String> properties = process.getWorkerProperty().getSystemConfig().getAllProperties();
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "10");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "8");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "8GB");
+        assertEquals(properties.get("system-memory-gb"), "10");
+        assertEquals(properties.get("query-memory-gb"), "8");
+        assertEquals(properties.get("query.max-memory-per-node"), "8GB");
     }
 
     @Test
@@ -159,9 +159,9 @@ public void testUpdateWorkerMemoryPropertiesWithZeroOffHeapMemory()
 
         // Verify that values remain unchanged when offHeapMemory is 0
         Map<String, String> properties = process.getWorkerProperty().getSystemConfig().getAllProperties();
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "10");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "8");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "8GB");
+        assertEquals(properties.get("system-memory-gb"), "10");
+        assertEquals(properties.get("query-memory-gb"), "8");
+        assertEquals(properties.get("query.max-memory-per-node"), "8GB");
     }
 
     @Test
@@ -176,9 +176,9 @@ public void testUpdateWorkerMemoryPropertiesWithSmallerOffHeapMemory()
 
         // Verify that values remain unchanged when offHeapMemory is smaller than current system memory
         Map<String, String> properties = process.getWorkerProperty().getSystemConfig().getAllProperties();
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "10");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "8");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "8GB");
+        assertEquals(properties.get("system-memory-gb"), "10");
+        assertEquals(properties.get("query-memory-gb"), "8");
+        assertEquals(properties.get("query.max-memory-per-node"), "8GB");
     }
 
     @Test
@@ -198,9 +198,9 @@ public void testUpdateWorkerMemoryPropertiesWithDifferentQueryMemoryFraction()
         // newSystemMemory = 30GB
         // queryMemoryFraction = 6/12 = 0.5
         // newQueryMemory = 30 * 0.5 = 15GB
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "30");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "15");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "15GB");
+        assertEquals(properties.get("system-memory-gb"), "30");
+        assertEquals(properties.get("query-memory-gb"), "15");
+        assertEquals(properties.get("query.max-memory-per-node"), "15GB");
     }
 
     @Test
@@ -215,9 +215,9 @@ public void testUpdateWorkerMemoryPropertiesWithZeroCurrentSystemMemory()
 
         // Verify that values remain unchanged when current system memory is 0
         Map<String, String> properties = process.getWorkerProperty().getSystemConfig().getAllProperties();
-        assertEquals(properties.get(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB), "0");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MEMORY_GB), "8");
-        assertEquals(properties.get(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE), "8GB");
+        assertEquals(properties.get("system-memory-gb"), "0");
+        assertEquals(properties.get("query-memory-gb"), "8");
+        assertEquals(properties.get("query.max-memory-per-node"), "8GB");
     }
 
     private TestingNativeExecutionProcess createTestingNativeExecutionProcess(
@@ -227,9 +227,9 @@ private TestingNativeExecutionProcess createTestingNativeExecutionProcess(
             SparkConf sparkConf)
     {
         Map<String, String> systemConfigs = new HashMap<>();
-        systemConfigs.put(NativeExecutionSystemConfig.SYSTEM_MEMORY_GB, systemMemoryGb);
-        systemConfigs.put(NativeExecutionSystemConfig.QUERY_MEMORY_GB, queryMemoryGb);
-        systemConfigs.put(NativeExecutionSystemConfig.QUERY_MAX_MEMORY_PER_NODE, queryMaxMemoryPerNode);
+        systemConfigs.put("system-memory-gb", systemMemoryGb);
+        systemConfigs.put("query-memory-gb", queryMemoryGb);
+        systemConfigs.put("query.max-memory-per-node", queryMaxMemoryPerNode);
 
         NativeExecutionSystemConfig systemConfig = new NativeExecutionSystemConfig(systemConfigs);
         PrestoSparkWorkerProperty workerProperty = new PrestoSparkWorkerProperty(

presto-spark-base/src/test/java/com/facebook/presto/spark/execution/property/TestNativeExecutionSystemConfig.java

@@ -60,25 +60,16 @@ public void testNativeExecutionSystemConfig()
                 .put("connector.num-io-threads-hw-multiplier", "0")
                 .put("presto.version", "dummy.presto.version")
                 .put("shutdown-onset-sec", "10")
-                .put("system-memory-gb", "10")
-                .put("query-memory-gb", "8")
-                .put("query.max-memory-per-node", "8GB")
-                .put("use-mmap-allocator", "true")
-                .put("memory-arbitrator-kind", "SHARED")
-                .put("shared-arbitrator.reserved-capacity", "0GB")
-                .put("shared-arbitrator.memory-pool-initial-capacity", "4GB")
-                .put("shared-arbitrator.max-memory-arbitration-time", "5m")
+                .put("system-mem-pushback-enabled", "true")
+                .put("system-mem-pushback-abort-enabled", "true")
+                .put("system-mem-limit-gb", "7")
+                .put("system-mem-shrink-gb", "1")
+                .put("worker-overloaded-threshold-mem-gb", "6")
+                .put("worker-overloaded-threshold-cpu-pct", "85")
                 .put("experimental.spiller-spill-path", "")
-                .put("task.max-drivers-per-task", "15")
                 .put("enable-old-task-cleanup", "false")
-                .put("shuffle.name", "local")
                 .put("http-server.enable-access-log", "true")
                 .put("core-on-allocation-failure-enabled", "false")
-                .put("spill-enabled", "true")
-                .put("aggregation-spill-enabled", "true")
-                .put("join-spill-enabled", "true")
-                .put("order-by-spill-enabled", "true")
-                .put("max-spill-bytes", String.valueOf(600L << 30))
                 .build();
         assertEquals(nativeExecutionSystemConfig.getAllProperties(), expectedConfigs);