Forward Merge #24115 with conflict resolution

.pre-commit-config.yaml

@@ -43,7 +43,7 @@ repos:
     hooks:
       - id: mypy
         additional_dependencies: [
-          "polars>=1.35,<1.43",
+          "polars>=1.35,<1.45",
           "numpy>=1.26",
           "pyarrow-stubs>=19.0",
           "pyarrow>=19.0.0,<24.0.0",  # https://github.com/NVIDIA/cudf/issues/22229

conda/environments/all_cuda-129_arch-aarch64.yaml

@@ -78,7 +78,7 @@ dependencies:
 - packaging
 - pandas>=3.0.0,<3.1.0
 - pandoc
-- polars>=1.35,<1.43
+- polars>=1.35,<1.45
 - pre-commit
 - psutil
 - pyarrow>=19.0.0

conda/environments/all_cuda-129_arch-x86_64.yaml

@@ -78,7 +78,7 @@ dependencies:
 - packaging
 - pandas>=3.0.0,<3.1.0
 - pandoc
-- polars>=1.35,<1.43
+- polars>=1.35,<1.45
 - pre-commit
 - psutil
 - pyarrow>=19.0.0

conda/environments/all_cuda-133_arch-aarch64.yaml

@@ -78,7 +78,7 @@ dependencies:
 - packaging
 - pandas>=3.0.0,<3.1.0
 - pandoc
-- polars>=1.35,<1.43
+- polars>=1.35,<1.45
 - pre-commit
 - psutil
 - pyarrow>=19.0.0

conda/environments/all_cuda-133_arch-x86_64.yaml

@@ -78,7 +78,7 @@ dependencies:
 - packaging
 - pandas>=3.0.0,<3.1.0
 - pandoc
-- polars>=1.35,<1.43
+- polars>=1.35,<1.45
 - pre-commit
 - psutil
 - pyarrow>=19.0.0

conda/recipes/cudf-polars/recipe.yaml

@@ -39,7 +39,7 @@ requirements:
     - rapidsmpf =${{ minor_version }}
     - kvikio =${{ minor_version }}
     - cudf-streaming =${{ version }}
-    - polars>=1.35,<1.43
+    - polars>=1.35,<1.45
     - packaging
     - ${{ pin_compatible("cuda-version", upper_bound="x", lower_bound="x") }}
     - if: cuda_major == "12"

cpp/examples/hybrid_scan_io/hybrid_scan_multifile_single_step.cpp

@@ -16,6 +16,7 @@
 #include <rmm/cuda_stream_pool.hpp>
 #include <rmm/mr/statistics_resource_adaptor.hpp>
 
+#include <cassert>
 #include <ranges>
 #include <stdexcept>
 #include <string>
@@ -60,7 +61,8 @@ struct hybrid_scan_single_step_fn {
           input_sources[source_idx], {}, filters, false, stream, mr);
       }
     }
-    stream.synchronize_no_throw();
+    [[maybe_unused]] auto const status = cudaStreamSynchronize(stream.get());
+    assert(status == cudaSuccess);
     if (verbose) {
       std::cout << "Thread " << tid << " ";
       timer.print_elapsed_millis();

cpp/examples/hybrid_scan_io/hybrid_scan_multifile_two_step.cpp

@@ -16,6 +16,7 @@
 #include <rmm/cuda_stream_pool.hpp>
 #include <rmm/mr/statistics_resource_adaptor.hpp>
 
+#include <cassert>
 #include <ranges>
 #include <stdexcept>
 #include <string>
@@ -62,7 +63,8 @@ struct hybrid_scan_two_step_fn {
         input_sources[source_idx], filter_expression_opt, filters, false, stream, mr);
     }
 
-    stream.synchronize_no_throw();
+    [[maybe_unused]] auto const status = cudaStreamSynchronize(stream.get());
+    assert(status == cudaSuccess);
 
     if (verbose) {
       std::cout << "Thread " << tid << " ";
@@ -157,7 +159,7 @@ int main(int argc, char const** argv)
   // Create filter expressions (one per thread; reused circularly if needed)
   auto const column_reference = cudf::ast::column_name_reference(column_name);
   auto scalar                 = cudf::string_scalar(literal_value, true, default_stream);
-  default_stream.synchronize();
+  default_stream.sync();
   auto literal = cudf::ast::literal(scalar);
 
   std::vector<cudf::ast::operation> filter_expressions;

cpp/examples/parquet_io/parquet_io_multithreaded.cpp

@@ -147,7 +147,7 @@ std::vector<table_t> read_parquet_multithreaded(std::vector<io_source> const& in
   if (read_mode == read_mode::CONCATENATE_ALL) {
     auto stream    = stream_pool.get_stream();
     auto final_tbl = concatenate_tables(std::move(tables), stream);
-    stream.synchronize();
+    stream.sync();
     tables.clear();
     tables.emplace_back(std::move(final_tbl));
   }

cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp

@@ -61,6 +61,8 @@ class hybrid_scan_multifile {
   /**
    * @brief Constructor for the multi-file experimental Parquet reader
    *
+   * @throws std::invalid_argument if no sources are provided
+   *
    * @param footer_bytes Host span of Parquet file footer byte spans, one per source
    * @param options Parquet reader options
    */
@@ -70,12 +72,25 @@ class hybrid_scan_multifile {
   /**
    * @brief Constructor for the multi-file experimental Parquet reader
    *
+   * @throws std::invalid_argument if no sources are provided
+   *
    * @param parquet_metadata Host span of pre-populated Parquet file metadata, one per source
    * @param options Parquet reader options
    */
   explicit hybrid_scan_multifile(cudf::host_span<FileMetaData const> parquet_metadata,
                                  parquet_reader_options const& options);
 
+  /**
+   * @brief Constructor that takes ownership of pre-populated Parquet file metadata
+   *
+   * @throws std::invalid_argument if no sources are provided
+   *
+   * @param parquet_metadata Pre-populated Parquet file metadata, one per source
+   * @param options Parquet reader options
+   */
+  explicit hybrid_scan_multifile(std::vector<FileMetaData>&& parquet_metadata,
+                                 parquet_reader_options const& options);
+
   /**
    * @brief Destructor for the multi-file experimental Parquet reader
    */

cpp/include/cudf/utilities/span.hpp

@@ -107,26 +107,28 @@ struct host_span {
 
   /// Constructor from container
   /// @param in The container to construct the span from
-  template <typename C,
-            // Only supported containers of types convertible to T
-            std::enable_if_t<is_host_span_supported_container<C>::value &&
-                             std::is_convertible_v<
-                               std::remove_pointer_t<decltype(thrust::raw_pointer_cast(  // NOLINT
-                                 std::declval<C&>().data()))> (*)[],
-                               T (*)[]>>* = nullptr>  // NOLINT
+  template <
+    typename C,
+    // Only supported containers of types convertible to T
+    std::enable_if_t<
+      is_host_span_supported_container<C>::value &&
+      std::is_convertible_v<std::remove_pointer_t<decltype(thrust::raw_pointer_cast(  // NOLINT
+                              std::declval<C&>().data()))> (*)[],  // NOLINT(modernize-type-traits)
+                            T (*)[]>>* = nullptr>                  // NOLINT
   constexpr host_span(C& in) : _span{thrust::raw_pointer_cast(in.data()), in.size()}
   {
   }
 
   /// Constructor from const container
   /// @param in The container to construct the span from
-  template <typename C,
-            // Only supported containers of types convertible to T
-            std::enable_if_t<is_host_span_supported_container<C>::value &&
-                             std::is_convertible_v<
-                               std::remove_pointer_t<decltype(thrust::raw_pointer_cast(  // NOLINT
-                                 std::declval<C&>().data()))> (*)[],
-                               T (*)[]>>* = nullptr>  // NOLINT
+  template <
+    typename C,
+    // Only supported containers of types convertible to T
+    std::enable_if_t<
+      is_host_span_supported_container<C>::value &&
+      std::is_convertible_v<std::remove_pointer_t<decltype(thrust::raw_pointer_cast(  // NOLINT
+                              std::declval<C&>().data()))> (*)[],  // NOLINT(modernize-type-traits)
+                            T (*)[]>>* = nullptr>                  // NOLINT
   constexpr host_span(C const& in) : _span{thrust::raw_pointer_cast(in.data()), in.size()}
   {
   }

cpp/libcudf_streaming/src/bloom_filter.cpp

@@ -118,7 +118,7 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
   auto storage = (co_await bloom_filter->receive()).release<rmm::device_buffer>();
   RAPIDSMPF_EXPECTS((co_await bloom_filter->receive()).empty(),
                     "Bloom filter channel contained more than one message");
-  auto stream = cuda::stream_ref{storage.stream().get()};
+  auto stream = storage.stream();
   rapidsmpf::CudaEvent event;
   auto filter = cudf_streaming::detail::device_bloom_filter(filter_size_, seed_, storage.data());
   auto meta   = co_await ch_in->receive_metadata();

cpp/src/io/parquet/compact_protocol_reader.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2018-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -8,12 +8,17 @@
 #include "parquet_common.hpp"
 
 #include <cudf/io/parquet_schema.hpp>
+#include <cudf/utilities/error.hpp>
 #include <cudf/utilities/export.hpp>
 
 #include <cuda/std/bit>
 
 #include <algorithm>
+#include <concepts>
 #include <cstddef>
+#include <limits>
+#include <stdexcept>
+#include <type_traits>
 #include <utility>
 
 namespace CUDF_EXPORT cudf {
@@ -36,24 +41,41 @@ class CompactProtocolReader {
   explicit CompactProtocolReader(uint8_t const* base = nullptr, size_t len = 0) { init(base, len); }
   void init(uint8_t const* base, size_t len)
   {
+    // A null base is valid only for an empty buffer; a positive length would then have no backing
+    // storage. This keeps every later pointer op defined (the empty state has all-null pointers).
+    CUDF_EXPECTS(base != nullptr || len == 0,
+                 "CompactProtocolReader requires a non-null buffer when length is non-zero",
+                 std::invalid_argument);
     m_base = m_cur = base;
-    m_end          = base + len;
+    // Guard against `nullptr + len` (undefined) so a zero-length buffer stays fully defined.
+    m_end = base != nullptr ? base + len : base;
+  }
+  [[nodiscard]] ptrdiff_t bytecount() const noexcept
+  {
+    // Avoid `nullptr - nullptr` on a null-base reader; it has consumed nothing.
+    return m_base != nullptr ? m_cur - m_base : 0;
   }
-  [[nodiscard]] ptrdiff_t bytecount() const noexcept { return m_cur - m_base; }
   unsigned int getb() noexcept { return (m_cur < m_end) ? *m_cur++ : 0; }
   void skip_bytes(size_t bytecnt) noexcept
   {
     bytecnt = std::min(bytecnt, (size_t)(m_end - m_cur));
     m_cur += bytecnt;
   }
 
-  // returns a varint encoded integer
-  template <typename T>
-  T get_varint() noexcept
+  // Returns a varint-encoded integer. `T` is constrained to unsigned so `numeric_limits<T>::digits`
+  // is the full value width; a signed `T` would drop the sign bit and misplace the overflow bound.
+  template <std::unsigned_integral T>
+  T get_varint()
   {
     T v = 0;
     for (uint32_t l = 0;; l += 7) {
-      T c = getb();
+      T const c = getb();
+      // The byte's value, shifted into place, must fit in `T`; `l < digits` also keeps `max() >> l`
+      // itself in range. Comparing the raw byte, not the masked payload, is intentional: it also
+      // rejects a continuation byte whose successor group could not fit.
+      CUDF_EXPECTS(l < std::numeric_limits<T>::digits && c <= (std::numeric_limits<T>::max() >> l),
+                   "Parquet varint exceeds the width of its target type",
+                   std::overflow_error);
       v |= (c & 0x7f) << l;
       if (c < 0x80) { break; }
     }
@@ -62,22 +84,22 @@ class CompactProtocolReader {
 
   // returns a zigzag encoded signed integer
   template <typename T>
-  T get_zigzag() noexcept
+  T get_zigzag()
   {
     using U   = std::make_unsigned_t<T>;
     U const u = get_varint<U>();
     return static_cast<T>((u >> 1u) ^ -static_cast<T>(u & 1));
   }
 
   // thrift spec says to use zigzag i32 for i16 types
-  int32_t get_i16() noexcept { return get_zigzag<int32_t>(); }
-  int32_t get_i32() noexcept { return get_zigzag<int32_t>(); }
-  int64_t get_i64() noexcept { return get_zigzag<int64_t>(); }
+  int32_t get_i16() { return get_zigzag<int32_t>(); }
+  int32_t get_i32() { return get_zigzag<int32_t>(); }
+  int64_t get_i64() { return get_zigzag<int64_t>(); }
 
-  uint32_t get_u32() noexcept { return get_varint<uint32_t>(); }
-  uint64_t get_u64() noexcept { return get_varint<uint64_t>(); }
+  uint32_t get_u32() { return get_varint<uint32_t>(); }
+  uint64_t get_u64() { return get_varint<uint64_t>(); }
 
-  [[nodiscard]] std::pair<uint8_t, uint32_t> get_listh() noexcept
+  [[nodiscard]] std::pair<uint8_t, uint32_t> get_listh()
   {
     uint32_t const c = getb();
     uint32_t sz      = c >> 4;

cpp/src/io/parquet/experimental/hybrid_scan.cpp

@@ -18,7 +18,7 @@ hybrid_scan_metadata::hybrid_scan_metadata(cudf::host_span<uint8_t const> footer
   : _metadata{std::make_shared<detail::aggregate_reader_metadata>(
       std::vector<cudf::host_span<uint8_t const>>{footer_bytes},
       options.is_enabled_use_arrow_schema(),
-      options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas())}
+      options.is_enabled_allow_mismatched_pq_schemas())}
 {
 }
 
@@ -27,7 +27,7 @@ hybrid_scan_metadata::hybrid_scan_metadata(FileMetaData const& parquet_metadata,
   : _metadata{std::make_shared<detail::aggregate_reader_metadata>(
       std::vector<FileMetaData>{parquet_metadata},
       options.is_enabled_use_arrow_schema(),
-      options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas())}
+      options.is_enabled_allow_mismatched_pq_schemas())}
 {
 }
 

cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp

@@ -36,22 +36,6 @@ using text::byte_range_info;
 
 namespace {
 
-// Construct a vector of FileMetaData from the input footer bytes
-[[nodiscard]] std::vector<FileMetaData> parquet_metadatas_from_footer_bytes(
-  cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes)
-{
-  std::vector<FileMetaData> parquet_metadatas;
-  parquet_metadatas.reserve(footer_bytes.size());
-  std::transform(footer_bytes.begin(),
-                 footer_bytes.end(),
-                 std::back_inserter(parquet_metadatas),
-                 [](auto const& footer_bytes) {
-                   metadata parsed_metadata{footer_bytes};
-                   return FileMetaData{std::move(parsed_metadata)};
-                 });
-  return parquet_metadatas;
-}
-
 // Construct a vector of all row group indices from the input vectors
 [[nodiscard]] auto all_row_group_indices(
   std::span<std::vector<cudf::size_type> const> row_group_indices)
@@ -77,8 +61,10 @@ namespace {
 }
 
 // Compute the page index (column index and/or offset index) byte range
-[[nodiscard]] byte_range_info page_index_byte_range(FileMetaData const& file_metadata)
+[[nodiscard]] byte_range_info page_index_byte_range(parquet::detail::metadata const& file_metadata)
 {
+  if (file_metadata.is_page_index_setup()) { return {}; }
+
   auto const& row_groups = file_metadata.row_groups;
   if (row_groups.empty() or row_groups.front().columns.empty()) { return {}; }
 
@@ -132,25 +118,31 @@ aggregate_reader_metadata::aggregate_reader_metadata(
   cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
   bool use_arrow_schema,
   bool has_cols_from_mismatched_srcs)
-  : aggregate_reader_metadata_base(parquet_metadatas_from_footer_bytes(footer_bytes),
-                                   use_arrow_schema,
-                                   has_cols_from_mismatched_srcs)
+  : aggregate_reader_metadata(
+      parquet::detail::parallel_construct_metadatas(
+        footer_bytes, [](auto const& bytes) { return FileMetaData{metadata{bytes}}; }),
+      use_arrow_schema,
+      has_cols_from_mismatched_srcs)
 {
-  CUDF_EXPECTS(
-    not footer_bytes.empty(), "At least one source must be provided", std::invalid_argument);
 }
 
 aggregate_reader_metadata::aggregate_reader_metadata(
   cudf::host_span<FileMetaData const> parquet_metadatas,
   bool use_arrow_schema,
   bool has_cols_from_mismatched_srcs)
-  : aggregate_reader_metadata_base(
+  : aggregate_reader_metadata(
       std::vector<FileMetaData>{parquet_metadatas.begin(), parquet_metadatas.end()},
       use_arrow_schema,
       has_cols_from_mismatched_srcs)
 {
-  CUDF_EXPECTS(
-    not parquet_metadatas.empty(), "At least one source must be provided", std::invalid_argument);
+}
+
+aggregate_reader_metadata::aggregate_reader_metadata(std::vector<FileMetaData>&& parquet_metadatas,
+                                                     bool use_arrow_schema,
+                                                     bool has_cols_from_mismatched_srcs)
+  : aggregate_reader_metadata_base(
+      std::move(parquet_metadatas), use_arrow_schema, has_cols_from_mismatched_srcs)
+{
 }
 
 std::vector<text::byte_range_info> aggregate_reader_metadata::page_index_byte_ranges() const