Merge branch 'main' into spark-arithmetic-ansi-mode

Author: malinjawi

CMake/resolve_dependency_modules/cudf.cmake

@@ -17,48 +17,48 @@ include_guard(GLOBAL)
 # 3.30.4 is the minimum version required by cudf
 cmake_minimum_required(VERSION 3.30.4)
 
-# rapids_cmake commit 5ec2245 from 2026-01-26
+# rapids_cmake commit 7ece71c from 2026-02-04
 set(VELOX_rapids_cmake_VERSION 26.04)
-set(VELOX_rapids_cmake_COMMIT 5ec22457e58953e0a68f0745ce7a11a896ba62b1)
+set(VELOX_rapids_cmake_COMMIT 7ece71c2f94fb0ed402d567b457ce54ecb859695)
 set(
   VELOX_rapids_cmake_BUILD_SHA256_CHECKSUM
-  bf7d4ed5885f5fe012c42fb0977e1fe1416896479ffd34baa0cf762d3e83dc80
+  02abaa8580c30a0b01eb142d5cd58b5acc85005bf58f5360f4a62efbd6e4635a
 )
 set(
   VELOX_rapids_cmake_SOURCE_URL
   "https://github.com/rapidsai/rapids-cmake/archive/${VELOX_rapids_cmake_COMMIT}.tar.gz"
 )
 velox_resolve_dependency_url(rapids_cmake)
 
-# rmm commit e728b29 from 2026-01-26
+# rmm commit 3d6669c from 2026-02-09
 set(VELOX_rmm_VERSION 26.04)
-set(VELOX_rmm_COMMIT e728b2923f748d71aad30294b6926f43cb4c826e)
+set(VELOX_rmm_COMMIT 3d6669cd21e15080a0af2dc18f991060be2a4c3c)
 set(
   VELOX_rmm_BUILD_SHA256_CHECKSUM
-  ec18d881b327514de154af67a33a1288eec7bcd86909f23c9bf2d90511b0cf2f
+  1d2575d7a0fb492feaabc6917e7db33eb5c446a8e8eea301c54bd7e3d25fe66c
 )
 set(VELOX_rmm_SOURCE_URL "https://github.com/rapidsai/rmm/archive/${VELOX_rmm_COMMIT}.tar.gz")
 velox_resolve_dependency_url(rmm)
 
-# kvikio commit 0f03349 from 2026-01-26
+# kvikio commit 593245b from 2026-02-05
 set(VELOX_kvikio_VERSION 26.04)
-set(VELOX_kvikio_COMMIT 0f03349bcaf029a2f582d9915a88d09e355ac691)
+set(VELOX_kvikio_COMMIT 593245b7799b6ea91eed77dd03ce4c9a4e158465)
 set(
   VELOX_kvikio_BUILD_SHA256_CHECKSUM
-  728868c671e2686b5e9b7b4122d1661475f803c4fb98c0852d7be65c365d7b2d
+  bcd03423b727fb0a23551a8ad3c6fcb58eaf0eb54ded7cdce914ea07a60ea1d7
 )
 set(
   VELOX_kvikio_SOURCE_URL
   "https://github.com/rapidsai/kvikio/archive/${VELOX_kvikio_COMMIT}.tar.gz"
 )
 velox_resolve_dependency_url(kvikio)
 
-# cudf commit 68a0714 from 2026-01-27
+# cudf commit fc213fc from 2026-02-10
 set(VELOX_cudf_VERSION 26.04 CACHE STRING "cudf version")
-set(VELOX_cudf_COMMIT 68a0714a3701431041cb47bf1163706f597f9f48)
+set(VELOX_cudf_COMMIT fc213fc1ad889e2edf291b5555764ce677cb5dfa)
 set(
   VELOX_cudf_BUILD_SHA256_CHECKSUM
-  0c723d7fd04eab60336dd4bcce41e225821d13b54cdabe485ec54517f3aa8b15
+  f2c8eec90cb8571188e5bdd32222da208c353bbe1066c56bb3052a988feda1a2
 )
 set(VELOX_cudf_SOURCE_URL "https://github.com/rapidsai/cudf/archive/${VELOX_cudf_COMMIT}.tar.gz")
 velox_resolve_dependency_url(cudf)

velox/common/base/tests/SkewedPartitionBalancerTest.cpp

@@ -76,6 +76,10 @@ class SkewedPartitionRebalancerTestHelper {
 
 class SkewedPartitionRebalancerTest : public testing::Test {
  protected:
+  static void SetUpTestSuite() {
+    TestValue::enable();
+  }
+
   std::unique_ptr<SkewedPartitionRebalancer> createBalancer(
       uint32_t numPartitions = 128,
       uint32_t numTasks = 8,
@@ -356,7 +360,7 @@ DEBUG_ONLY_TEST_F(SkewedPartitionRebalancerTest, serializedRebalanceExecution) {
   // there. This ensures that when we call rebalance() from the main thread,
   // rebalancing_ is already true and our rebalance() call will return early.
   mainThreadRebalancerWait.await(
-      [&] { return mainThreadRebalancerWaitFlag.load(); });
+      [&] { return !mainThreadRebalancerWaitFlag.load(); });
 
   balancer->rebalance();
 

velox/common/caching/AsyncDataCache.cpp

@@ -935,11 +935,15 @@ void AsyncDataCache::possibleSsdSave(uint64_t bytes) {
 
   ssdSaveable_ += bytes;
   if (memory::AllocationTraits::numPages(ssdSaveable_) >
-      std::max<int32_t>(
-          static_cast<int32_t>(
-              memory::AllocationTraits::numPages(opts_.minSsdSavableBytes)),
-          static_cast<int32_t>(
-              static_cast<double>(cachedPages_) * opts_.ssdSavableRatio))) {
+          std::max<int32_t>(
+              static_cast<int32_t>(
+                  memory::AllocationTraits::numPages(opts_.minSsdSavableBytes)),
+              static_cast<int32_t>(
+                  static_cast<double>(cachedPages_) * opts_.ssdSavableRatio)) ||
+      (opts_.ssdFlushThresholdBytes > 0 &&
+       memory::AllocationTraits::numPages(ssdSaveable_) >
+           static_cast<int32_t>(memory::AllocationTraits::numPages(
+               opts_.ssdFlushThresholdBytes)))) {
     // Do not start a new save if another one is in progress.
     if (!ssdCache_->startWrite()) {
       return;

velox/common/caching/AsyncDataCache.h

@@ -714,11 +714,13 @@ class AsyncDataCache : public memory::Cache {
         double _maxWriteRatio = 0.7,
         double _ssdSavableRatio = 0.125,
         int32_t _minSsdSavableBytes = 1 << 24,
-        int32_t _numShards = kDefaultNumShards)
+        int32_t _numShards = kDefaultNumShards,
+        uint64_t _ssdFlushThresholdBytes = 0)
         : maxWriteRatio(_maxWriteRatio),
           ssdSavableRatio(_ssdSavableRatio),
           minSsdSavableBytes(_minSsdSavableBytes),
-          numShards(_numShards) {}
+          numShards(_numShards),
+          ssdFlushThresholdBytes(_ssdFlushThresholdBytes) {}
 
     /// The max ratio of the number of in-memory cache entries being written to
     /// SSD cache over the total number of cache entries. This is to control SSD
@@ -742,6 +744,11 @@ class AsyncDataCache : public memory::Cache {
     /// shards to decrease contention on the mutex for the key to entry mapping
     /// and other housekeeping. Must be a power of 2.
     int32_t numShards;
+
+    /// 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).
+    uint64_t ssdFlushThresholdBytes;
   };
 
   AsyncDataCache(

velox/common/caching/tests/AsyncDataCacheTest.cpp

@@ -1504,6 +1504,65 @@ TEST_P(AsyncDataCacheTest, ssdWriteOptions) {
   }
 }
 
+TEST_P(AsyncDataCacheTest, ssdFlushThresholdBytes) {
+  constexpr uint64_t kRamBytes = 16UL << 20; // 16 MB
+  constexpr uint64_t kSsdBytes = 64UL << 20; // 64 MB
+
+  struct {
+    double maxWriteRatio;
+    double ssdSavableRatio;
+    int32_t minSsdSavableBytes;
+    uint64_t ssdFlushThresholdBytes;
+    bool expectedSaveToSsd;
+
+    std::string debugString() const {
+      return fmt::format(
+          "maxWriteRatio {}, ssdSavableRatio {}, minSsdSavableBytes {}, ssdFlushThresholdBytes {}, expectedSaveToSsd {}",
+          maxWriteRatio,
+          ssdSavableRatio,
+          minSsdSavableBytes,
+          ssdFlushThresholdBytes,
+          expectedSaveToSsd);
+    }
+  } testSettings[] = {
+      // Ratio-based threshold not met, ssdFlushThresholdBytes disabled (0).
+      // No flush expected.
+      {0.8, 0.95, 32 << 20, 0, false},
+      // Ratio-based threshold not met, but ssdFlushThresholdBytes is small
+      // (1MB).
+      // Flush expected due to absolute threshold.
+      {0.8, 0.95, 32 << 20, 1UL << 20, true},
+      // Ratio-based threshold met. ssdFlushThresholdBytes disabled.
+      // Flush expected due to ratio.
+      {0.8, 0.3, 4 << 20, 0, true},
+      // Both thresholds could trigger. Flush expected.
+      {0.8, 0.3, 4 << 20, 1UL << 20, true}};
+
+  for (const auto& testData : testSettings) {
+    SCOPED_TRACE(testData.debugString());
+    initializeCache(
+        kRamBytes,
+        kSsdBytes,
+        0,
+        true,
+        AsyncDataCache::Options(
+            testData.maxWriteRatio,
+            testData.ssdSavableRatio,
+            testData.minSsdSavableBytes,
+            AsyncDataCache::kDefaultNumShards,
+            testData.ssdFlushThresholdBytes));
+    // Load data half of the in-memory capacity.
+    loadLoop(0, kRamBytes / 2);
+    waitForPendingLoads();
+    auto stats = cache_->refreshStats();
+    if (testData.expectedSaveToSsd) {
+      EXPECT_GT(stats.ssdStats->entriesWritten, 0);
+    } else {
+      EXPECT_EQ(stats.ssdStats->entriesWritten, 0);
+    }
+  }
+}
+
 TEST_P(AsyncDataCacheTest, appendSsdSaveable) {
   constexpr uint64_t kRamBytes = 64UL << 20; // 64 MB
   constexpr uint64_t kSsdBytes = 128UL << 20; // 128 MB

velox/connectors/hive/storage_adapters/abfs/RegisterAbfsFileSystem.cpp

@@ -52,7 +52,8 @@ std::unique_ptr<velox::dwio::common::FileSink> abfsWriteFileSinkGenerator(
         fileSystem->openFileForWrite(fileURI),
         fileURI,
         options.metricLogger,
-        options.stats);
+        options.stats,
+        options.fileSystemStats);
   }
   return nullptr;
 }

velox/connectors/hive/storage_adapters/gcs/RegisterGcsFileSystem.cpp

@@ -103,7 +103,8 @@ std::unique_ptr<velox::dwio::common::FileSink> gcsWriteFileSinkGenerator(
         fileSystem->openFileForWrite(fileURI, {{}, options.pool, std::nullopt}),
         fileURI,
         options.metricLogger,
-        options.stats);
+        options.stats,
+        options.fileSystemStats);
   }
   return nullptr;
 }

velox/connectors/hive/storage_adapters/hdfs/RegisterHdfsFileSystem.cpp

@@ -89,7 +89,8 @@ hdfsWriteFileSinkGenerator() {
               fileSystem->openFileForWrite(pathSuffix),
               fileURI,
               options.metricLogger,
-              options.stats);
+              options.stats,
+              options.fileSystemStats);
         }
         return static_cast<std::unique_ptr<dwio::common::WriteFileSink>>(
             nullptr);

velox/connectors/hive/storage_adapters/s3fs/RegisterS3FileSystem.cpp

@@ -102,7 +102,8 @@ std::unique_ptr<velox::dwio::common::FileSink> s3WriteFileSinkGenerator(
         fileSystem->openFileForWrite(fileURI, {{}, options.pool, std::nullopt}),
         fileURI,
         options.metricLogger,
-        options.stats);
+        options.stats,
+        options.fileSystemStats);
   }
   return nullptr;
 }

velox/core/Expressions.cpp

@@ -14,6 +14,9 @@
  * limitations under the License.
  */
 #include "velox/core/Expressions.h"
+
+#include <folly/hash/Hash.h>
+
 #include "velox/common/Casts.h"
 #include "velox/common/encode/Base64.h"
 #include "velox/vector/ComplexVector.h"
@@ -424,7 +427,8 @@ uint64_t hashImpl(const TypePtr& type, const Variant& value) {
 } // namespace
 
 size_t ConstantTypedExpr::localHash() const {
-  static const size_t kBaseHash = std::hash<const char*>()("ConstantTypedExpr");
+  static const size_t kBaseHash =
+      folly::hasher<std::string_view>()("ConstantTypedExpr");
 
   uint64_t h;