Add spill related configs to system configs

Summary: Adding spill configs so they can be set from configs, and then overridden by session properties.

Differential Revision: D71002997

Author: HeidiHan0000

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

@@ -45,7 +45,15 @@ void updateFromSystemConfigs(
           {core::QueryConfig::kQueryMaxMemoryPerNode,
            std::string(SystemConfig::kQueryMaxMemoryPerNode)},
           {core::QueryConfig::kSpillFileCreateConfig,
-           std::string(SystemConfig::kSpillerFileCreateConfig)}};
+           std::string(SystemConfig::kSpillerFileCreateConfig)},
+          {core::QueryConfig::kSpillEnabled,
+          std::string(SystemConfig::kSpillEnabled)},
+          {core::QueryConfig::kJoinSpillEnabled,
+          std::string(SystemConfig::kJoinSpillEnabled)},
+          {core::QueryConfig::kOrderBySpillEnabled,
+          std::string(SystemConfig::kOrderBySpillEnabled)},
+          {core::QueryConfig::kAggregationSpillEnabled,
+          std::string(SystemConfig::kAggregationSpillEnabled)}};
 
   for (const auto& configNameEntry : sessionSystemConfigMapping) {
     const auto& sessionName = configNameEntry.first;

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -250,6 +250,10 @@ SystemConfig::SystemConfig() {
           BOOL_PROP(kPlanValidatorFailOnNestedLoopJoin, false),
           STR_PROP(kPrestoDefaultNamespacePrefix, "presto.default"),
           STR_PROP(kPoolType, "DEFAULT"),
+          BOOL_PROP(kSpillEnabled, false),
+          BOOL_PROP(kJoinSpillEnabled, true),
+          BOOL_PROP(kAggregationSpillEnabled, true),
+          BOOL_PROP(kOrderBySpillEnabled, true),
       };
 }
 
@@ -313,6 +317,22 @@ std::string SystemConfig::poolType() const {
   return value;
 }
 
+bool SystemConfig::spillEnabled() const {
+  return optionalProperty<bool>(kSpillEnabled).value();
+}
+
+bool SystemConfig::joinSpillEnabled() const {
+  return optionalProperty<bool>(kJoinSpillEnabled).value();
+}
+
+bool SystemConfig::aggregationSpillEnabled() const {
+  return optionalProperty<bool>(kAggregationSpillEnabled).value();
+}
+
+bool SystemConfig::orderBySpillEnabled() const {
+  return optionalProperty<bool>(kOrderBySpillEnabled).value();
+}
+
 bool SystemConfig::mutableConfig() const {
   return optionalProperty<bool>(kMutableConfig).value();
 }

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -662,6 +662,14 @@ class SystemConfig : public ConfigBase {
 
   // Specifies the type of worker pool
   static constexpr std::string_view kPoolType{"pool-type"};
+  
+  // Spill related configs
+  static constexpr std::string_view kSpillEnabled{"spill-enabled"};
+  static constexpr std::string_view kJoinSpillEnabled{"join-spill-enabled"};
+  static constexpr std::string_view kAggregationSpillEnabled{
+      "aggregation-spill-enabled"};
+  static constexpr std::string_view kOrderBySpillEnabled{
+      "order-by-spill-enabled"};
 
   SystemConfig();
 
@@ -904,6 +912,14 @@ class SystemConfig : public ConfigBase {
   std::string prestoDefaultNamespacePrefix() const;
 
   std::string poolType() const;
+
+  bool spillEnabled() const; 
+
+  bool joinSpillEnabled() const;
+
+  bool aggregationSpillEnabled() const;
+
+  bool orderBySpillEnabled() const;
 };
 
 /// Provides access to node properties defined in node.properties file.

presto-native-execution/presto_cpp/main/tests/QueryContextManagerTest.cpp

@@ -108,34 +108,54 @@ TEST_F(QueryContextManagerTest, defaultSessionProperties) {
   EXPECT_EQ(queryConfig.maxSpillLevel(), defaultQC->maxSpillLevel());
   EXPECT_EQ(
       queryConfig.spillCompressionKind(), defaultQC->spillCompressionKind());
+  EXPECT_EQ(queryConfig.spillEnabled(), defaultQC->spillEnabled());
+  EXPECT_EQ(queryConfig.aggregationSpillEnabled(), defaultQC->aggregationSpillEnabled());
   EXPECT_EQ(queryConfig.joinSpillEnabled(), defaultQC->joinSpillEnabled());
+  EXPECT_EQ(queryConfig.orderBySpillEnabled(), defaultQC->orderBySpillEnabled());
   EXPECT_EQ(
       queryConfig.validateOutputFromOperators(),
       defaultQC->validateOutputFromOperators());
   EXPECT_EQ(
       queryConfig.spillWriteBufferSize(), defaultQC->spillWriteBufferSize());
 }
 
-TEST_F(QueryContextManagerTest, overrdingSessionProperties) {
+TEST_F(QueryContextManagerTest, overridingSessionProperties) {
   protocol::TaskId taskId = "scan.0.0.1.0";
   const auto& systemConfig = SystemConfig::instance();
   {
     protocol::SessionRepresentation session{.systemProperties = {}};
     auto queryCtx =
         taskManager_->getQueryContextManager()->findOrCreateQueryCtx(
             taskId, session);
+    // When session properties are not explicitly set, they should be set to 
+    // system config values.
     EXPECT_EQ(
         queryCtx->queryConfig().queryMaxMemoryPerNode(),
         systemConfig->queryMaxMemoryPerNode());
     EXPECT_EQ(
         queryCtx->queryConfig().spillFileCreateConfig(),
         systemConfig->spillerFileCreateConfig());
+    EXPECT_EQ(
+        queryCtx->queryConfig().spillEnabled(),
+        systemConfig->spillEnabled());
+    EXPECT_EQ(
+        queryCtx->queryConfig().aggregationSpillEnabled(),
+        systemConfig->aggregationSpillEnabled());
+    EXPECT_EQ(
+        queryCtx->queryConfig().joinSpillEnabled(),
+        systemConfig->joinSpillEnabled());
+    EXPECT_EQ(
+        queryCtx->queryConfig().orderBySpillEnabled(),
+        systemConfig->orderBySpillEnabled());
   }
   {
     protocol::SessionRepresentation session{
         .systemProperties = {
             {"query_max_memory_per_node", "1GB"},
-            {"spill_file_create_config", "encoding:replica_2"}}};
+            {"spill_file_create_config", "encoding:replica_2"},
+            {"spill_enabled", "true"},
+            {"aggregation_spill_enabled", "false"},
+            {"join_spill_enabled", "true"}}};
     auto queryCtx =
         taskManager_->getQueryContextManager()->findOrCreateQueryCtx(
             taskId, session);
@@ -144,6 +164,28 @@ TEST_F(QueryContextManagerTest, overrdingSessionProperties) {
         1UL * 1024 * 1024 * 1024);
     EXPECT_EQ(
         queryCtx->queryConfig().spillFileCreateConfig(), "encoding:replica_2");
+    // Override with different value
+    EXPECT_EQ(
+        queryCtx->queryConfig().spillEnabled(), true);
+    EXPECT_NE(
+        queryCtx->queryConfig().spillEnabled(),
+        systemConfig->spillEnabled());
+    // Override with different value
+    EXPECT_EQ(
+        queryCtx->queryConfig().aggregationSpillEnabled(), false);
+    EXPECT_NE(
+        queryCtx->queryConfig().aggregationSpillEnabled(),
+        systemConfig->aggregationSpillEnabled());
+    // Override with same value
+    EXPECT_EQ(
+        queryCtx->queryConfig().joinSpillEnabled(), true);
+    EXPECT_EQ(
+        queryCtx->queryConfig().joinSpillEnabled(),
+        systemConfig->joinSpillEnabled());
+    // No override
+    EXPECT_EQ(
+        queryCtx->queryConfig().orderBySpillEnabled(),
+        systemConfig->orderBySpillEnabled());
   }
 }