[native] Fix session properties race condition

Summary: The global SessionProperties object shall not be updated by every query. It is meaningless to update the default value and it is not thread safe. Removing the updates and corresponding update method to avoid inconsistent session property display.

Differential Revision: D79859171

Author: tanjialiang

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

@@ -107,11 +107,10 @@ protocol::NodeState convertNodeState(presto::NodeState nodeState) {
 }
 
 void enableChecksum() {
-  velox::exec::OutputBufferManager::getInstanceRef()->setListenerFactory(
-      []() {
-        return std::make_unique<
-            velox::serializer::presto::PrestoOutputStreamListener>();
-      });
+  velox::exec::OutputBufferManager::getInstanceRef()->setListenerFactory([]() {
+    return std::make_unique<
+        velox::serializer::presto::PrestoOutputStreamListener>();
+  });
 }
 
 // Log only the catalog keys that are configured to avoid leaking
@@ -502,7 +501,8 @@ void PrestoServer::run() {
         << "' has " << httpSrvCpuExecutor_->numThreads() << " threads.";
     for (auto evb : httpSrvIoExecutor_->getAllEventBases()) {
       evb->setMaxLatency(
-          std::chrono::milliseconds(systemConfig->httpSrvIoEvbViolationThresholdMs()),
+          std::chrono::milliseconds(
+              systemConfig->httpSrvIoEvbViolationThresholdMs()),
           []() { RECORD_METRIC_VALUE(kCounterHttpServerIoEvbViolation, 1); },
           /*dampen=*/false);
     }
@@ -834,7 +834,8 @@ void PrestoServer::initializeThreadPools() {
                            << " threads.";
   for (auto evb : exchangeHttpIoExecutor_->getAllEventBases()) {
     evb->setMaxLatency(
-        std::chrono::milliseconds(systemConfig->exchangeIoEvbViolationThresholdMs()),
+        std::chrono::milliseconds(
+            systemConfig->exchangeIoEvbViolationThresholdMs()),
         []() { RECORD_METRIC_VALUE(kCounterExchangeIoEvbViolation, 1); },
         /*dampen=*/false);
   }

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

@@ -49,21 +49,21 @@ void updateFromSystemConfigs(
           {core::QueryConfig::kSpillFileCreateConfig,
            std::string(SystemConfig::kSpillerFileCreateConfig)},
           {core::QueryConfig::kSpillEnabled,
-          std::string(SystemConfig::kSpillEnabled)},
+           std::string(SystemConfig::kSpillEnabled)},
           {core::QueryConfig::kJoinSpillEnabled,
-          std::string(SystemConfig::kJoinSpillEnabled)},
+           std::string(SystemConfig::kJoinSpillEnabled)},
           {core::QueryConfig::kOrderBySpillEnabled,
-          std::string(SystemConfig::kOrderBySpillEnabled)},
+           std::string(SystemConfig::kOrderBySpillEnabled)},
           {core::QueryConfig::kAggregationSpillEnabled,
-          std::string(SystemConfig::kAggregationSpillEnabled)},
+           std::string(SystemConfig::kAggregationSpillEnabled)},
           {core::QueryConfig::kMaxSpillBytes,
           std::string(SystemConfig::kMaxSpillBytes)},
           {core::QueryConfig::kRequestDataSizesMaxWaitSec,
-          std::string(SystemConfig::kRequestDataSizesMaxWaitSec)},
+           std::string(SystemConfig::kRequestDataSizesMaxWaitSec)},
           {core::QueryConfig::kMaxSplitPreloadPerDriver,
-          std::string(SystemConfig::kDriverMaxSplitPreload)},
+           std::string(SystemConfig::kDriverMaxSplitPreload)},
           {core::QueryConfig::kMaxLocalExchangePartitionBufferSize,
-          std::string(SystemConfig::kMaxLocalExchangePartitionBufferSize)}};
+           std::string(SystemConfig::kMaxLocalExchangePartitionBufferSize)}};
   for (const auto& configNameEntry : veloxToSystemConfigMapping) {
     const auto& veloxConfigName = configNameEntry.first;
     const auto& systemConfigName = configNameEntry.second;
@@ -93,8 +93,7 @@ toConnectorConfigs(const protocol::TaskUpdateRequest& taskUpdateRequest) {
         taskUpdateRequest.extraCredentials.begin(),
         taskUpdateRequest.extraCredentials.end());
     connectorConfig.insert({"user", taskUpdateRequest.session.user});
-    connectorConfigs.insert(
-        {entry.first, connectorConfig});
+    connectorConfigs.insert({entry.first, connectorConfig});
   }
 
   return connectorConfigs;
@@ -252,7 +251,6 @@ QueryContextManager::toVeloxConfigs(
           velox::common::compressionKindToString(compressionKind);
     } else {
       configs[sessionProperties_.toVeloxConfig(it.first)] = it.second;
-      sessionProperties_.updateVeloxConfig(it.first, it.second);
     }
   }
 

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

@@ -316,7 +316,6 @@ SessionProperties::SessionProperties() {
       QueryConfig::kQueryTraceMaxBytes,
       std::to_string(c.queryTraceMaxBytes()));
 
-
   addSessionProperty(
       kOpTraceDirectoryCreateConfig,
       "Config used to create operator trace directory. This config is provided to"
@@ -507,31 +506,20 @@ SessionProperties::SessionProperties() {
 }
 
 const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&
-SessionProperties::getSessionProperties() {
+SessionProperties::testingSessionProperties() const {
   return sessionProperties_;
 }
 
-const std::string SessionProperties::toVeloxConfig(const std::string& name) {
+const std::string SessionProperties::toVeloxConfig(
+    const std::string& name) const {
   auto it = sessionProperties_.find(name);
   return it == sessionProperties_.end() ? name
                                         : it->second->getVeloxConfigName();
 }
 
-void SessionProperties::updateVeloxConfig(
-    const std::string& name,
-    const std::string& value) {
-  auto it = sessionProperties_.find(name);
-  // Velox config value is updated only for presto session properties.
-  if (it == sessionProperties_.end()) {
-    return;
-  }
-  it->second->updateValue(value);
-}
-
-json SessionProperties::serialize() {
+json SessionProperties::serialize() const {
   json j = json::array();
-  const auto sessionProperties = getSessionProperties();
-  for (const auto& entry : sessionProperties) {
+  for (const auto& entry : sessionProperties_) {
     j.push_back(entry.second->serialize());
   }
   return j;

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

@@ -324,18 +324,16 @@ class SessionProperties {
 
   SessionProperties();
 
-  const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&
-  getSessionProperties();
-
   /// Utility function to translate a config name in Presto to its equivalent in
   /// Velox. Returns 'name' as is if there is no mapping.
-  const std::string toVeloxConfig(const std::string& name);
+  const std::string toVeloxConfig(const std::string& name) const;
 
-  void updateVeloxConfig(const std::string& name, const std::string& value);
+  json serialize() const;
 
-  json serialize();
+  const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&
+  testingSessionProperties() const;
 
- protected:
+ private:
   void addSessionProperty(
       const std::string& name,
       const std::string& description,

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

@@ -53,7 +53,7 @@ TEST_F(SessionPropertiesTest, validateMapping) {
       core::QueryConfig::kRequestDataSizesMaxWaitSec,
       core::QueryConfig::kQueryMemoryReclaimerPriority,
       core::QueryConfig::kMaxNumSplitsListenedTo};
-  auto sessionProperties = SessionProperties().getSessionProperties();
+  auto sessionProperties = SessionProperties().testingSessionProperties();
   const auto len = names.size();
   for (auto i = 0; i < len; i++) {
     EXPECT_EQ(
@@ -68,7 +68,7 @@ TEST_F(SessionPropertiesTest, serializeProperty) {
   for (const auto& property : j) {
     auto name = property["name"];
     json expectedProperty =
-        properties.getSessionProperties().at(name)->serialize();
+        properties.testingSessionProperties().at(name)->serialize();
     EXPECT_EQ(property, expectedProperty);
   }
 }