feat(native): Add config for asyc cache flush threshold

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

@@ -1079,7 +1079,8 @@ void PrestoServer::initializeVeloxMemory() {
         systemConfig->asyncCacheMaxSsdWriteRatio(),
         systemConfig->asyncCacheSsdSavableRatio(),
         systemConfig->asyncCacheMinSsdSavableBytes(),
-        systemConfig->asyncCacheNumShards()};
+        systemConfig->asyncCacheNumShards(),
+        systemConfig->asyncCacheSsdFlushThresholdBytes()};
     cache_ = velox::cache::AsyncDataCache::create(
         velox::memory::memoryManager()->allocator(),
         std::move(ssd),

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

@@ -210,6 +210,7 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kAsyncCacheSsdSavableRatio, 0.125),
           NUM_PROP(kAsyncCacheMinSsdSavableBytes, 1 << 24 /*16MB*/),
           NUM_PROP(kAsyncCacheNumShards, 4),
+          NUM_PROP(kAsyncCacheSsdFlushThresholdBytes, 0),
           STR_PROP(kAsyncCachePersistenceInterval, "0s"),
           BOOL_PROP(kAsyncCacheSsdDisableFileCow, false),
           BOOL_PROP(kSsdCacheChecksumEnabled, false),
@@ -719,6 +720,10 @@ int32_t SystemConfig::asyncCacheNumShards() const {
   return optionalProperty<int32_t>(kAsyncCacheNumShards).value();
 }
 
+uint64_t SystemConfig::asyncCacheSsdFlushThresholdBytes() const {
+  return optionalProperty<uint64_t>(kAsyncCacheSsdFlushThresholdBytes).value();
+}
+
 std::chrono::duration<double> SystemConfig::asyncCachePersistenceInterval()
     const {
   return velox::config::toDuration(

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

@@ -438,6 +438,12 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kAsyncCacheNumShards{
       "async-cache-num-shards"};
 
+  /// The maximum threshold in bytes for triggering SSD flush. When the
+  /// accumulated SSD-savable bytes exceed this value, a flush to SSD is
+  /// triggered. Set to 0 to disable this threshold (default).
+  static constexpr std::string_view kAsyncCacheSsdFlushThresholdBytes{
+      "async-cache-ssd-flush-threshold-bytes"};
+
   /// The interval for persisting in-memory cache to SSD. Setting this config
   /// to a non-zero value will activate periodic cache persistence.
   static constexpr std::string_view kAsyncCachePersistenceInterval{
@@ -1079,6 +1085,8 @@ class SystemConfig : public ConfigBase {
 
   int32_t asyncCacheNumShards() const;
 
+  uint64_t asyncCacheSsdFlushThresholdBytes() const;
+
   std::chrono::duration<double> asyncCachePersistenceInterval() const;
 
   bool asyncCacheSsdDisableFileCow() const;

presto-native-execution/presto_cpp/main/common/tests/ConfigTest.cpp

@@ -241,6 +241,20 @@ TEST_F(ConfigTest, asyncCacheNumShards) {
   ASSERT_EQ(config.asyncCacheNumShards(), 8);
 }
 
+TEST_F(ConfigTest, asyncCacheSsdFlushThresholdBytes) {
+  SystemConfig config;
+  init(config, {});
+  // Test default value is 0
+  ASSERT_EQ(config.asyncCacheSsdFlushThresholdBytes(), 0);
+
+  // Test custom value
+  init(
+      config,
+      {{std::string(SystemConfig::kAsyncCacheSsdFlushThresholdBytes),
+        "134217728"}});
+  ASSERT_EQ(config.asyncCacheSsdFlushThresholdBytes(), 134217728);
+}
+
 TEST_F(ConfigTest, remoteFunctionServer) {
   SystemConfig config;
   init(config, {});