Forward-merge main into pandas3

cpp/include/cudf/context.hpp

@@ -78,4 +78,28 @@ void initialize(init_flags flags = init_flags::INIT_JIT_CACHE);
 /// teardown and that only one thread calls teardown at a time.
 void teardown();
 
+/**
+ * @brief Enable or disable the JIT program cache
+ *
+ * When disabled, the cache will not be used for
+ * storing or retrieving compiled programs, effectively bypassing the cache. When enabled, the
+ * cache will be used as normal. This can be used to temporarily disable caching without clearing
+ * the existing cache contents, allowing for easy re-enabling of the cache later.
+ *
+ * @param enable If `true`, the JIT program cache is enabled; if `false`, it is disabled.
+ */
+void enable_jit_cache(bool enable);
+
+/**
+ * @brief Clear the JIT program cache, removing all cached programs from memory and disk.
+ *
+ * This is a more expensive operation than simply disabling the cache, as it involves deleting
+ * cached files from disk, but it also frees up any memory used by the cached programs. Use
+ * `enable_jit_cache(false)` if you want to temporarily disable caching without clearing existing
+ * cache contents.
+ *
+ * @warning For benchmarking or testing purposes, prefer `enable_jit_cache`.
+ */
+void clear_jit_cache();
+
 }  // namespace CUDF_EXPORT cudf

cpp/include/cudf/detail/aggregation/aggregation.hpp

@@ -136,7 +136,8 @@ class count_aggregation final
   : public clonable<count_aggregation>::derived_from<rolling_aggregation,
                                                      groupby_aggregation,
                                                      groupby_scan_aggregation,
-                                                     reduce_aggregation> {
+                                                     reduce_aggregation,
+                                                     scan_aggregation> {
  public:
   count_aggregation(aggregation::Kind kind) : aggregation(kind) {}
 };

cpp/include/cudf/detail/unary.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2018-2024, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -8,50 +8,12 @@
 #include <cudf/column/column_factories.hpp>
 #include <cudf/unary.hpp>
 #include <cudf/utilities/default_stream.hpp>
-#include <cudf/utilities/export.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
-#include <rmm/exec_policy.hpp>
 
-#include <thrust/transform.h>
-
-namespace CUDF_EXPORT cudf {
+namespace cudf {
 namespace detail {
-/**
- * @brief Creates a column of `type_id::BOOL8` elements by applying a predicate to every element
- * between
- * [`begin, `end`) `true` indicates the value is satisfies the predicate and `false` indicates it
- * doesn't.
- *
- * @tparam InputIterator Iterator type for `begin` and `end`
- * @tparam Predicate A predicator type which will be evaluated
- * @param begin Beginning of the sequence of elements
- * @param end End of the sequence of elements
- * @param p Predicate to be applied to each element in `[begin,end)`
- * @param stream CUDA stream used for device memory operations and kernel launches.
- * @param mr Device memory resource used to allocate the returned column's device memory
- *
- * @returns A column of type `type_id::BOOL8,` with `true` representing predicate is satisfied.
- */
-
-template <typename InputIterator, typename Predicate>
-std::unique_ptr<column> true_if(InputIterator begin,
-                                InputIterator end,
-                                size_type size,
-                                Predicate p,
-                                rmm::cuda_stream_view stream,
-                                rmm::device_async_resource_ref mr)
-{
-  auto output =
-    make_numeric_column(data_type(type_id::BOOL8), size, mask_state::UNALLOCATED, stream, mr);
-  auto output_mutable_view = output->mutable_view();
-  auto output_data         = output_mutable_view.data<bool>();
-
-  thrust::transform(rmm::exec_policy_nosync(stream), begin, end, output_data, p);
-
-  return output;
-}
 
 /**
  * @copydoc cudf::unary_operation
@@ -91,4 +53,4 @@ std::unique_ptr<column> is_not_nan(cudf::column_view const& input,
                                    rmm::device_async_resource_ref mr);
 
 }  // namespace detail
-}  // namespace CUDF_EXPORT cudf
+}  // namespace cudf

cpp/src/aggregation/aggregation.cpp

@@ -121,6 +121,8 @@ template CUDF_EXPORT std::unique_ptr<groupby_scan_aggregation>
 make_count_aggregation<groupby_scan_aggregation>(null_policy null_handling);
 template CUDF_EXPORT std::unique_ptr<reduce_aggregation> make_count_aggregation<reduce_aggregation>(
   null_policy null_handling);
+template CUDF_EXPORT std::unique_ptr<scan_aggregation> make_count_aggregation<scan_aggregation>(
+  null_policy null_handling);
 
 /// Factory to create a HISTOGRAM aggregation
 template <typename Base>

cpp/src/io/parquet/experimental/page_index_filter.cu

@@ -38,6 +38,7 @@
 
 #include <algorithm>
 #include <limits>
+#include <numeric>
 
 namespace cudf::io::parquet::experimental::detail {
 
@@ -911,7 +912,7 @@ std::unique_ptr<cudf::column> aggregate_reader_metadata::build_row_mask_with_pag
     [&](auto col_idx) {
       auto const schema_idx = output_column_schemas[col_idx];
       auto const& dtype     = output_dtypes[col_idx];
-      // Only participating columns and comparable types except fixed point are supported
+      // Only participating columns and comparable types are supported
       if (not stats_columns_mask[col_idx] or
           (cudf::is_compound(dtype) && dtype.id() != cudf::type_id::STRING)) {
         // Placeholder for unsupported types and non-participating columns

cpp/src/io/parquet/predicate_pushdown.cpp

@@ -146,7 +146,7 @@ std::optional<std::vector<std::vector<size_type>>> aggregate_reader_metadata::ap
   for (size_t col_idx = 0; col_idx < output_dtypes.size(); col_idx++) {
     auto const schema_idx = output_column_schemas[col_idx];
     auto const& dtype     = output_dtypes[col_idx];
-    // Only participating columns and comparable types except fixed point are supported
+    // Only participating columns and comparable types are supported
     if (not stats_columns_mask[col_idx] or
         (cudf::is_compound(dtype) && dtype.id() != cudf::type_id::STRING)) {
       // Placeholder for unsupported types and non-participating columns

cpp/src/io/parquet/stats_filter_helpers.hpp

@@ -18,8 +18,9 @@
 #include <cudf/utilities/traits.hpp>
 
 #include <algorithm>
-#include <numeric>
-#include <string>
+#include <array>
+#include <bit>
+#include <string_view>
 
 namespace cudf::io::parquet::detail {
 
@@ -38,8 +39,34 @@ constexpr size_t initial_chars_capacity = 1024;
  */
 class stats_caster_base {
  protected:
+  static inline numeric::decimal128::rep decode_flba_decimal128(uint8_t const* stats_val)
+  {
+    auto constexpr endianness = std::endian::native;
+    static_assert(endianness == std::endian::little or endianness == std::endian::big,
+                  "Encountered unsupported endianness while decoding decimal128 from FLBA");
+    using RepType = numeric::decimal128::rep;
+    auto value    = RepType{};
+    std::memcpy(&value, stats_val, sizeof(RepType));
+    auto value_rep = std::bit_cast<std::array<std::byte, sizeof(RepType)>>(value);
+    // byte-swap to native representation on little-endian platforms
+    if constexpr (endianness == std::endian::little) { std::ranges::reverse(value_rep); }
+    return std::bit_cast<RepType>(value_rep);
+  }
+
+  template <typename T>
+  static inline T decode_fixed_width_value(uint8_t const* stats_val, size_t stats_size)
+    requires((cudf::is_integral<T>() and !cudf::is_boolean<T>()) or cudf::is_fixed_point<T>() or
+             cudf::is_chrono<T>())
+  {
+    CUDF_EXPECTS(stats_size == sizeof(T),
+                 "Parquet reader encountered a statistics vector larger than the type's size");
+    auto value = T{};
+    std::memcpy(&value, stats_val, std::min(stats_size, sizeof(T)));
+    return value;
+  }
+
   template <typename ToType, typename FromType>
-  static inline ToType targetType(FromType const value)
+  static inline ToType target_type(FromType const value)
   {
     if constexpr (cudf::is_timestamp<ToType>()) {
       return static_cast<ToType>(
@@ -52,64 +79,76 @@ class stats_caster_base {
   }
 
   // uses storage type as T
-  template <typename T, CUDF_ENABLE_IF(cudf::is_dictionary<T>() or cudf::is_nested<T>())>
+  template <typename T>
   static inline T convert(uint8_t const* stats_val, size_t stats_size, Type const type)
+    requires(cudf::is_dictionary<T>() or cudf::is_nested<T>())
   {
     CUDF_FAIL("unsupported type for stats casting");
   }
 
-  template <typename T, CUDF_ENABLE_IF(cudf::is_boolean<T>())>
+  template <typename T>
   static inline T convert(uint8_t const* stats_val, size_t stats_size, Type const type)
+    requires(cudf::is_boolean<T>())
   {
     CUDF_EXPECTS(type == Type::BOOLEAN, "Invalid type and stats combination");
-    return stats_caster_base::targetType<T>(*reinterpret_cast<bool const*>(stats_val));
+    return stats_caster_base::target_type<T>(*reinterpret_cast<bool const*>(stats_val));
   }
 
   // integral but not boolean, and fixed_point, and chrono.
-  template <typename T,
-            CUDF_ENABLE_IF((cudf::is_integral<T>() and !cudf::is_boolean<T>()) or
-                           cudf::is_fixed_point<T>() or cudf::is_chrono<T>())>
+  template <typename T>
   static inline T convert(uint8_t const* stats_val, size_t stats_size, Type const type)
+    requires((cudf::is_integral<T>() and !cudf::is_boolean<T>()) or cudf::is_fixed_point<T>() or
+             cudf::is_chrono<T>())
   {
     switch (type) {
       case Type::INT32:
-        return stats_caster_base::targetType<T>(*reinterpret_cast<int32_t const*>(stats_val));
+        return stats_caster_base::target_type<T>(
+          decode_fixed_width_value<int32_t>(stats_val, stats_size));
       case Type::INT64:
-        return stats_caster_base::targetType<T>(*reinterpret_cast<int64_t const*>(stats_val));
+        return stats_caster_base::target_type<T>(
+          decode_fixed_width_value<int64_t>(stats_val, stats_size));
       case Type::INT96:  // Deprecated in parquet specification
-        return stats_caster_base::targetType<T>(
-          static_cast<__int128_t>(reinterpret_cast<int64_t const*>(stats_val)[0]) << 32 |
-          reinterpret_cast<int32_t const*>(stats_val)[2]);
+        return stats_caster_base::target_type<T>(
+          static_cast<__int128_t>(decode_fixed_width_value<int64_t>(stats_val, stats_size)) << 32 |
+          decode_fixed_width_value<int32_t>(stats_val + sizeof(int64_t), stats_size));
       case Type::BYTE_ARRAY: [[fallthrough]];
       case Type::FIXED_LEN_BYTE_ARRAY:
         if (stats_size == sizeof(T)) {
-          // if type size == length of stats_val. then typecast and return.
           if constexpr (cudf::is_chrono<T>()) {
-            return stats_caster_base::targetType<T>(
-              *reinterpret_cast<typename T::rep const*>(stats_val));
+            return stats_caster_base::target_type<T>(
+              decode_fixed_width_value<typename T::rep>(stats_val, stats_size));
+          } else if constexpr (std::is_same_v<T, numeric::decimal128::rep>) {
+            // Decimals with physical type FLBA/BYTE_ARRAY are stored as two's complement using
+            // big-endian.
+            return stats_caster_base::target_type<T>(decode_flba_decimal128(stats_val));
           } else {
-            return stats_caster_base::targetType<T>(*reinterpret_cast<T const*>(stats_val));
+            // TODO(mh): We may need to add support for `decimal256` (two's complement using
+            // big-endian) and `UUID` types (big-endian)
+            return stats_caster_base::target_type<T>(
+              decode_fixed_width_value<T>(stats_val, stats_size));
           }
         }
         // unsupported type
       default: CUDF_FAIL("Invalid type and stats combination");
     }
   }
 
-  template <typename T, CUDF_ENABLE_IF(cudf::is_floating_point<T>())>
+  template <typename T>
   static inline T convert(uint8_t const* stats_val, size_t stats_size, Type const type)
+    requires(cudf::is_floating_point<T>())
   {
     switch (type) {
       case Type::FLOAT:
-        return stats_caster_base::targetType<T>(*reinterpret_cast<float const*>(stats_val));
+        return stats_caster_base::target_type<T>(*reinterpret_cast<float const*>(stats_val));
       case Type::DOUBLE:
-        return stats_caster_base::targetType<T>(*reinterpret_cast<double const*>(stats_val));
+        return stats_caster_base::target_type<T>(*reinterpret_cast<double const*>(stats_val));
       default: CUDF_FAIL("Invalid type and stats combination");
     }
   }
 
-  template <typename T, CUDF_ENABLE_IF(std::is_same_v<T, string_view>)>
+  template <typename T>
   static inline T convert(uint8_t const* stats_val, size_t stats_size, Type const type)
+    requires(std::is_same_v<T, string_view>)
   {
     switch (type) {
       case Type::BYTE_ARRAY: [[fallthrough]];

cpp/src/jit/cache.cpp

@@ -1,8 +1,9 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include "io/utilities/getenv_or.hpp"
 #include "runtime/context.hpp"
 
 #include <cudf/context.hpp>
@@ -95,40 +96,63 @@ std::string get_program_cache_dir()
 #endif
 }
 
-std::size_t try_parse_numeric_env_var(char const* const env_name, std::size_t default_val)
-{
-  auto const value = std::getenv(env_name);
-  return value != nullptr ? std::stoull(value) : default_val;
-}
 }  // namespace
 
 jitify2::ProgramCache<>& jit::program_cache::get(jitify2::PreprocessedProgramData const& preprog)
 {
   CUDF_FUNC_RANGE();
-  std::lock_guard<std::mutex> const caches_lock(_caches_mutex);
+  std::lock_guard caches_lock(_caches_mutex);
 
   auto existing_cache = _caches.find(preprog.name());
 
-  if (existing_cache == _caches.end()) {
-    auto const kernel_limit_proc =
-      try_parse_numeric_env_var("LIBCUDF_KERNEL_CACHE_LIMIT_PER_PROCESS", 10'000);
-    auto const kernel_limit_disk =
-      try_parse_numeric_env_var("LIBCUDF_KERNEL_CACHE_LIMIT_DISK", 100'000);
-
-    // if kernel_limit_disk is zero, jitify will assign it the value of kernel_limit_proc.
-    // to avoid this, we treat zero as "disable disk caching" by not providing the cache dir.
-    auto const cache_dir = kernel_limit_disk == 0 ? std::string{} : get_program_cache_dir();
-
-    auto const res =
-      _caches.insert({preprog.name(),
+  if (existing_cache == _caches.end() || _disabled.load(std::memory_order_seq_cst)) {
+    auto res =
+      _caches.emplace(preprog.name(),
                       std::make_unique<jitify2::ProgramCache<>>(
-                        kernel_limit_proc, preprog, nullptr, cache_dir, kernel_limit_disk)});
+                        _kernel_limit_proc, preprog, nullptr, _cache_dir, _kernel_limit_disk));
     existing_cache = res.first;
   }
 
   return *(existing_cache->second);
 }
 
+void jit::program_cache::clear()
+{
+  CUDF_FUNC_RANGE();
+  std::lock_guard caches_lock(_caches_mutex);
+
+  _caches.clear();
+
+  // non-atomic
+  std::filesystem::remove_all(_cache_dir);
+}
+
+void jit::program_cache::enable(bool enable)
+{
+  _disabled.store(!enable, std::memory_order_seq_cst);
+}
+
+bool jit::program_cache::is_enabled() const { return !_disabled.load(std::memory_order_seq_cst); }
+
+std::unique_ptr<jit::program_cache> jit::program_cache::create()
+{
+  auto const kernel_limit_proc = getenv_or("LIBCUDF_KERNEL_CACHE_LIMIT_PER_PROCESS", 10'000);
+  auto const kernel_limit_disk = getenv_or("LIBCUDF_KERNEL_CACHE_LIMIT_DISK", 100'000);
+  auto const disabled          = get_bool_env_or("LIBCUDF_KERNEL_CACHE_DISABLED", false);
+  auto const clear_cache       = get_bool_env_or("LIBCUDF_KERNEL_CACHE_CLEAR", false);
+
+  // if kernel_limit_disk is zero, jitify will assign it the value of kernel_limit_proc.
+  // to avoid this, we treat zero as "disable disk caching" by not providing the cache dir.
+  auto cache_dir = kernel_limit_disk == 0 ? std::string{} : get_program_cache_dir();
+
+  auto cache =
+    std::make_unique<jit::program_cache>(kernel_limit_proc, kernel_limit_disk, cache_dir, disabled);
+
+  if (clear_cache) { cache->clear(); }
+
+  return cache;
+}
+
 jitify2::ProgramCache<>& jit::get_program_cache(jitify2::PreprocessedProgramData const& preprog)
 {
   return cudf::get_context().program_cache().get(preprog);