Add native_max_local_exchange_partition_count session property

presto-docs/src/main/sphinx/presto_cpp/properties-session.rst

@@ -258,3 +258,12 @@ The maximum size in bytes for the task's buffered output. The buffer is shared a
 The maximum bytes to buffer per PartitionedOutput operator to avoid creating tiny SerializedPages.
 For PartitionedOutputNode::Kind::kPartitioned, PartitionedOutput operator would buffer up to that number of
 bytes / number of destinations for each destination before producing a SerializedPage. Default is 32MB.
+
+``native_max_local_exchange_partition_count``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``bigint``
+* **Default value:** ``4294967296``
+
+Maximum number of partitions created by a local exchange.
+Affects concurrency for pipelines containing LocalPartitionNode.

presto-main/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java

@@ -60,6 +60,7 @@ public class NativeWorkerSessionPropertyProvider
     public static final String NATIVE_QUERY_TRACE_NODE_IDS = "native_query_trace_node_ids";
     public static final String NATIVE_QUERY_TRACE_MAX_BYTES = "native_query_trace_max_bytes";
     public static final String NATIVE_QUERY_TRACE_REG_EXP = "native_query_trace_task_reg_exp";
+    public static final String NATIVE_MAX_LOCAL_EXCHANGE_PARTITION_COUNT = "native_max_local_exchange_partition_count";
     private final List<PropertyMetadata<?>> sessionProperties;
 
     @Inject
@@ -225,6 +226,12 @@ public NativeWorkerSessionPropertyProvider(FeaturesConfig featuresConfig)
                                 "would buffer up to that number of bytes / number of destinations for each destination before " +
                                 "producing a SerializedPage.",
                         24L << 20,
+                        !nativeExecution),
+                integerProperty(
+                        NATIVE_MAX_LOCAL_EXCHANGE_PARTITION_COUNT,
+                        "Maximum number of partitions created by a local exchange. " +
+                                "Affects concurrency for pipelines containing LocalPartitionNode",
+                        null,
                         !nativeExecution));
     }
 

presto-native-execution/presto_cpp/main/SessionProperties.cpp

@@ -344,6 +344,15 @@ SessionProperties::SessionProperties() {
       // Overrides velox default value. Set it to 1 second to be aligned with
       // Presto Java.
       std::to_string(1000));
+
+  addSessionProperty(
+      kMaxLocalExchangePartitionCount,
+      "Maximum number of partitions created by a local exchange."
+      "Affects concurrency for pipelines containing LocalPartitionNode",
+      BIGINT(),
+      false,
+      QueryConfig::kMaxLocalExchangePartitionCount,
+      std::to_string(c.maxLocalExchangePartitionCount()));
 }
 
 const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&

presto-native-execution/presto_cpp/main/SessionProperties.h

@@ -215,6 +215,11 @@ class SessionProperties {
   static constexpr const char* kMaxPartitionedOutputBufferSize =
       "native_max_page_partitioning_buffer_size";
 
+  /// Maximum number of partitions created by a local exchange.
+  /// Affects concurrency for pipelines containing LocalPartitionNode.
+  static constexpr const char* kMaxLocalExchangePartitionCount =
+      "native_max_local_exchange_partition_count";
+
   SessionProperties();
 
   const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeHiveExternalTableTpchQueries.java

@@ -41,7 +41,6 @@
 import static com.facebook.presto.nativeworker.PrestoNativeQueryRunnerUtils.createExternalTable;
 import static com.facebook.presto.nativeworker.SymlinkManifestGeneratorUtils.cleanupSymlinkData;
 import static com.facebook.presto.tpch.TpchMetadata.getPrestoType;
-import static java.lang.String.format;
 
 public abstract class AbstractTestNativeHiveExternalTableTpchQueries
         extends AbstractTestNativeTpchQueries
@@ -114,7 +113,9 @@ public void tearDown()
         for (String tableName : TPCH_TABLES) {
             dropTableIfExists(javaQueryRunner, HIVE, TPCH, tableName);
         }
-        assertUpdate(format("DROP SCHEMA IF EXISTS %s.%s", HIVE, TPCH));
+
+        // https://github.com/prestodb/presto/issues/23908
+        // assertUpdate(format("DROP SCHEMA IF EXISTS %s.%s", HIVE, TPCH));
 
         File dataDirectory = ((DistributedQueryRunner) javaQueryRunner).getCoordinator().getDataDirectory().resolve(HIVE_DATA).toFile();
         Path symlinkTableDataPath = dataDirectory.toPath().getParent().resolve(SYMLINK_FOLDER);

presto-native-sidecar-plugin/src/main/java/com/facebook/presto/sidecar/sessionpropertyproviders/NativeSystemSessionPropertyProvider.java

@@ -102,7 +102,7 @@ private List<PropertyMetadata<?>> fetchSessionProperties()
             return propertyMetadataList;
         }
         catch (Exception e) {
-            throw new PrestoException(INVALID_ARGUMENTS, "Failed to get session properties from sidecar.");
+            throw new PrestoException(INVALID_ARGUMENTS, "Failed to get session properties from sidecar.", e);
         }
     }