diff --git a/presto-docs/src/main/sphinx/presto_cpp/properties-session.rst b/presto-docs/src/main/sphinx/presto_cpp/properties-session.rst index 6aa098e11bcde..02d6b417ac7ce 100644 --- a/presto-docs/src/main/sphinx/presto_cpp/properties-session.rst +++ b/presto-docs/src/main/sphinx/presto_cpp/properties-session.rst @@ -419,3 +419,23 @@ writer scaling if it is detected as overloaded by scale writer exchange. Minimum amount of data processed by all the logical table partitions to trigger skewed partition rebalancing by scale writer exchange. + +``native_table_scan_scaled_processing_enabled`` +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +* **Type:** ``boolean`` +* **Default value:** ``false`` + +If set to ``true``, enables scaling the table scan concurrency on each worker. + +``native_table_scan_scale_up_memory_usage_ratio`` +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +* **Type:** ``double`` +* **Minimum value:** ``0`` +* **Maximum value:** ``1`` +* **Default value:** ``0.7`` + +Controls the ratio of available memory that can be used for scaling up table scans. +A higher value allows more memory to be allocated for scaling up table scans, +while a lower value limits the amount of memory used. diff --git a/presto-main/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java b/presto-main/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java index 126a793827829..4f65719bb1c11 100644 --- a/presto-main/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java +++ b/presto-main/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java @@ -73,6 +73,8 @@ public class NativeWorkerSessionPropertyProvider public static final String NATIVE_SCALED_WRITER_MAX_PARTITIONS_PER_WRITER = "native_scaled_writer_max_partitions_per_writer"; public static final String NATIVE_SCALED_WRITER_MIN_PARTITION_PROCESSED_BYTES_REBALANCE_THRESHOLD = "native_scaled_writer_min_partition_processed_bytes_rebalance_threshold"; public static final String NATIVE_SCALED_WRITER_MIN_PROCESSED_BYTES_REBALANCE_THRESHOLD = "native_scaled_writer_min_processed_bytes_rebalance_threshold"; + public static final String NATIVE_TABLE_SCAN_SCALED_PROCESSING_ENABLED = "native_table_scan_scaled_processing_enabled"; + public static final String NATIVE_TABLE_SCAN_SCALE_UP_MEMORY_USAGE_RATIO = "native_table_scan_scale_up_memory_usage_ratio"; private final List> sessionProperties; @Inject @@ -311,6 +313,20 @@ public NativeWorkerSessionPropertyProvider(FeaturesConfig featuresConfig) "Minimum amount of data processed by all the logical table partitions " + "to trigger skewed partition rebalancing by scale writer exchange.", 256L << 20, + !nativeExecution), + booleanProperty( + NATIVE_TABLE_SCAN_SCALED_PROCESSING_ENABLED, + "If set to true, enables scaling the table scan concurrency on each worker.", + false, + !nativeExecution), + doubleProperty( + NATIVE_TABLE_SCAN_SCALE_UP_MEMORY_USAGE_RATIO, + "The query memory usage ratio used by scan controller to decide if it can " + + "increase the number of running scan threads. When the query memory usage " + + "is below this ratio, the scan controller keeps increasing the running scan " + + "thread for scale up, and stop once exceeds this ratio. The value is in the " + + "range of [0, 1].", + 0.7, !nativeExecution)); } diff --git a/presto-native-execution/presto_cpp/main/SessionProperties.cpp b/presto-native-execution/presto_cpp/main/SessionProperties.cpp index fab89fda0bf7a..ade7e46f91d87 100644 --- a/presto-native-execution/presto_cpp/main/SessionProperties.cpp +++ b/presto-native-execution/presto_cpp/main/SessionProperties.cpp @@ -443,6 +443,23 @@ SessionProperties::SessionProperties() { false, QueryConfig::kScaleWriterMinProcessedBytesRebalanceThreshold, std::to_string(c.scaleWriterMinProcessedBytesRebalanceThreshold())); + + addSessionProperty( + kTableScanScaledProcessingEnabled, + "If set to true, enables scaled processing for table scans.", + BOOLEAN(), + false, + QueryConfig::kTableScanScaledProcessingEnabled, + std::to_string(c.tableScanScaledProcessingEnabled())); + + addSessionProperty( + kTableScanScaleUpMemoryUsageRatio, + "Controls the ratio of available memory that can be used for scaling up table scans. " + "The value is in the range of [0, 1].", + DOUBLE(), + false, + QueryConfig::kTableScanScaleUpMemoryUsageRatio, + std::to_string(c.tableScanScaleUpMemoryUsageRatio())); } const std::unordered_map>& diff --git a/presto-native-execution/presto_cpp/main/SessionProperties.h b/presto-native-execution/presto_cpp/main/SessionProperties.h index 6d77bb8026f33..1d9b27dae6b8e 100644 --- a/presto-native-execution/presto_cpp/main/SessionProperties.h +++ b/presto-native-execution/presto_cpp/main/SessionProperties.h @@ -287,6 +287,15 @@ class SessionProperties { static constexpr const char* kShuffleCompressionEnabled = "exchange_compression"; + /// If set to true, enables scaled processing for table scans. + static constexpr const char* kTableScanScaledProcessingEnabled = + "native_table_scan_scaled_processing_enabled"; + + /// Controls the ratio of available memory that can be used for scaling up + /// table scans. The value is in the range of [0, 1]. + static constexpr const char* kTableScanScaleUpMemoryUsageRatio = + "native_table_scan_scale_up_memory_usage_ratio"; + SessionProperties(); const std::unordered_map>& diff --git a/presto-native-execution/presto_cpp/main/tests/SessionPropertiesTest.cpp b/presto-native-execution/presto_cpp/main/tests/SessionPropertiesTest.cpp index 6b6ad65334660..96f9f04c4bc2c 100644 --- a/presto-native-execution/presto_cpp/main/tests/SessionPropertiesTest.cpp +++ b/presto-native-execution/presto_cpp/main/tests/SessionPropertiesTest.cpp @@ -31,7 +31,9 @@ TEST_F(SessionPropertiesTest, validateMapping) { SessionProperties::kScaleWriterMaxPartitionsPerWriter, SessionProperties:: kScaleWriterMinPartitionProcessedBytesRebalanceThreshold, - SessionProperties::kScaleWriterMinProcessedBytesRebalanceThreshold}; + SessionProperties::kScaleWriterMinProcessedBytesRebalanceThreshold, + SessionProperties::kTableScanScaledProcessingEnabled, + SessionProperties::kTableScanScaleUpMemoryUsageRatio}; const std::vector veloxConfigNames = { core::QueryConfig::kAdjustTimestampToTimezone, core::QueryConfig::kDriverCpuTimeSliceLimitMs, @@ -40,7 +42,9 @@ TEST_F(SessionPropertiesTest, validateMapping) { core::QueryConfig::kScaleWriterMaxPartitionsPerWriter, core::QueryConfig:: kScaleWriterMinPartitionProcessedBytesRebalanceThreshold, - core::QueryConfig::kScaleWriterMinProcessedBytesRebalanceThreshold}; + core::QueryConfig::kScaleWriterMinProcessedBytesRebalanceThreshold, + core::QueryConfig::kTableScanScaledProcessingEnabled, + core::QueryConfig::kTableScanScaleUpMemoryUsageRatio}; auto sessionProperties = SessionProperties().getSessionProperties(); const auto len = names.size(); for (auto i = 0; i < len; i++) {