🚧 Implement an experimental Parquet reader optimized for highly-selective hybrid scan reads

cpp/CMakeLists.txt

@@ -521,6 +521,13 @@ add_library(
   src/io/parquet/compact_protocol_reader.cpp
   src/io/parquet/compact_protocol_writer.cpp
   src/io/parquet/decode_preprocess.cu
+  src/io/parquet/experimental/dictionary_page_filter.cu
+  src/io/parquet/experimental/hybrid_scan.cpp
+  src/io/parquet/experimental/hybrid_scan_impl.cpp
+  src/io/parquet/experimental/hybrid_scan_helpers.cpp
+  src/io/parquet/experimental/hybrid_scan_preprocess.cu
+  src/io/parquet/experimental/hybrid_scan_chunking.cu
+  src/io/parquet/experimental/page_index_filter.cu
   src/io/parquet/page_data.cu
   src/io/parquet/chunk_dict.cu
   src/io/parquet/page_enc.cu
@@ -806,6 +813,7 @@ set_source_files_properties(
   PROPERTIES COMPILE_DEFINITIONS "_FILE_OFFSET_BITS=64"
 )
 
+
 set_property(
   SOURCE src/io/parquet/writer_impl.cu
   APPEND

cpp/include/cudf/io/parquet.hpp

@@ -100,12 +100,13 @@ class parquet_reader_options {
   explicit parquet_reader_options() = default;
 
   /**
-   * @brief Creates a parquet_reader_options_builder which will build parquet_reader_options.
+   * @brief Creates a `parquet_reader_options_builder` to build `parquet_reader_options`.
+   *        By default, build with empty data source info.
    *
    * @param src Source information to read parquet file
    * @return Builder to build reader options
    */
-  static parquet_reader_options_builder builder(source_info src);
+  static parquet_reader_options_builder builder(source_info src = source_info{});
 
   /**
    * @brief Returns source info.

cpp/src/io/parquet/bloom_filter_reader.cu

@@ -508,7 +508,7 @@ std::vector<Type> aggregate_reader_metadata::get_parquet_types(
 }
 
 std::optional<std::vector<std::vector<size_type>>> aggregate_reader_metadata::apply_bloom_filters(
-  std::vector<rmm::device_buffer>& bloom_filter_data,
+  cudf::host_span<rmm::device_buffer> bloom_filter_data,
   host_span<std::vector<size_type> const> input_row_group_indices,
   host_span<std::vector<ast::literal*> const> literals,
   size_type total_row_groups,

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

@@ -0,0 +1,134 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "hybrid_scan_helpers.hpp"
+
+#include <cudf/ast/detail/expression_transformer.hpp>
+#include <cudf/ast/detail/operators.hpp>
+#include <cudf/ast/expressions.hpp>
+#include <cudf/detail/cuco_helpers.hpp>
+#include <cudf/detail/transform.hpp>
+#include <cudf/hashing/detail/xxhash_64.cuh>
+#include <cudf/io/parquet_schema.hpp>
+#include <cudf/logger.hpp>
+#include <cudf/utilities/span.hpp>
+#include <cudf/utilities/traits.hpp>
+#include <cudf/utilities/type_checks.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_buffer.hpp>
+#include <rmm/exec_policy.hpp>
+
+#include <thrust/iterator/counting_iterator.h>
+#include <thrust/tabulate.h>
+
+#include <future>
+#include <numeric>
+#include <optional>
+
+namespace cudf::io::parquet::experimental::detail {
+
+dictionary_literals_and_operators_collector::dictionary_literals_and_operators_collector() =
+  default;
+
+dictionary_literals_and_operators_collector::dictionary_literals_and_operators_collector(
+  ast::expression const& expr, cudf::size_type num_input_columns)
+{
+  _num_input_columns = num_input_columns;
+  _literals.resize(num_input_columns);
+  _operators.resize(num_input_columns);
+  expr.accept(*this);
+}
+
+std::reference_wrapper<ast::expression const> dictionary_literals_and_operators_collector::visit(
+  ast::column_reference const& expr)
+{
+  CUDF_EXPECTS(expr.get_table_source() == ast::table_reference::LEFT,
+               "DictionaryAST supports only left table");
+  CUDF_EXPECTS(expr.get_column_index() < _num_input_columns,
+               "Column index cannot be more than number of columns in the table");
+  return expr;
+}
+
+std::reference_wrapper<ast::expression const> dictionary_literals_and_operators_collector::visit(
+  ast::column_name_reference const& expr)
+{
+  CUDF_FAIL("Column name reference is not supported in DictionaryAST");
+}
+
+std::reference_wrapper<ast::expression const> dictionary_literals_and_operators_collector::visit(
+  ast::operation const& expr)
+{
+  using cudf::ast::ast_operator;
+  auto const operands = expr.get_operands();
+  auto const op       = expr.get_operator();
+
+  if (auto* v = dynamic_cast<ast::column_reference const*>(&operands[0].get())) {
+    // First operand should be column reference, second should be literal.
+    CUDF_EXPECTS(cudf::ast::detail::ast_operator_arity(op) == 2,
+                 "Only binary operations are supported on column reference");
+    auto const literal_ptr = dynamic_cast<ast::literal const*>(&operands[1].get());
+    CUDF_EXPECTS(literal_ptr != nullptr,
+                 "Second operand of binary operation with column reference must be a literal");
+    v->accept(*this);
+
+    // Push to the corresponding column's literals and operators list iff EQUAL or NOT_EQUAL
+    // operator is seen
+    if (op == ast_operator::EQUAL or op == ast::ast_operator::NOT_EQUAL) {
+      auto const col_idx = v->get_column_index();
+      _literals[col_idx].emplace_back(const_cast<ast::literal*>(literal_ptr));
+      _operators[col_idx].emplace_back(op);
+    }
+  } else {
+    // Just visit the operands and ignore any output
+    std::ignore = visit_operands(operands);
+  }
+
+  return expr;
+}
+
+std::pair<std::vector<std::vector<ast::literal*>>, std::vector<std::vector<ast::ast_operator>>>
+dictionary_literals_and_operators_collector::get_literals_and_operators() &&
+{
+  return {std::move(_literals), std::move(_operators)};
+}
+
+std::optional<std::vector<std::vector<size_type>>>
+aggregate_reader_metadata::apply_dictionary_filter(
+  cudf::host_span<rmm::device_buffer> dictionaries,
+  host_span<std::vector<size_type> const> input_row_group_indices,
+  host_span<std::vector<ast::literal*> const> literals,
+  host_span<std::vector<ast::ast_operator> const> operators,
+  size_type total_row_groups,
+  host_span<data_type const> output_dtypes,
+  host_span<int const> dictionary_col_schemas,
+  std::reference_wrapper<ast::expression const> filter,
+  rmm::cuda_stream_view stream) const
+{
+  return {};
+}
+
+std::vector<rmm::device_buffer> aggregate_reader_metadata::materialize_dictionaries(
+  cudf::host_span<rmm::device_buffer> dictionary_page_data,
+  host_span<std::vector<size_type> const> input_row_group_indices,
+  host_span<data_type const> output_dtypes,
+  host_span<int const> dictionary_col_schemas,
+  rmm::cuda_stream_view stream) const
+{
+  return {};
+}
+
+}  // namespace cudf::io::parquet::experimental::detail

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

@@ -0,0 +1,181 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "cudf/utilities/error.hpp"
+#include "hybrid_scan_impl.hpp"
+
+#include <cudf/io/experimental/hybrid_scan.hpp>
+
+#include <thrust/host_vector.h>
+
+namespace cudf::io::parquet::experimental {
+
+using text::byte_range_info;
+
+hybrid_scan_reader::hybrid_scan_reader(cudf::host_span<uint8_t const> footer_bytes,
+                                       parquet_reader_options const& options)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(footer_bytes, options)}
+{
+}
+
+hybrid_scan_reader::~hybrid_scan_reader() = default;
+
+[[nodiscard]] byte_range_info hybrid_scan_reader::page_index_byte_range() const
+{
+  return _impl->page_index_byte_range();
+}
+
+[[nodiscard]] FileMetaData hybrid_scan_reader::parquet_metadata() const
+{
+  return _impl->parquet_metadata();
+}
+
+void hybrid_scan_reader::setup_page_index(cudf::host_span<uint8_t const> page_index_bytes) const
+{
+  return _impl->setup_page_index(page_index_bytes);
+}
+
+std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(
+  parquet_reader_options const& options) const
+{
+  CUDF_EXPECTS(options.get_row_groups().size() <= 1,
+               "Encountered invalid size of row group indices in parquet reader options");
+
+  // If row groups are specified in parquet reader options, return them as is
+  if (options.get_row_groups().size() == 1) { return options.get_row_groups().front(); }
+
+  return _impl->all_row_groups(options);
+}
+
+size_type hybrid_scan_reader::num_rows_in_row_groups(
+  cudf::host_span<size_type const> row_group_indices) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->num_rows_in_row_groups(input_row_group_indices);
+}
+
+std::vector<cudf::size_type> hybrid_scan_reader::filter_row_groups_with_stats(
+  cudf::host_span<size_type const> row_group_indices,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->filter_row_groups_with_stats(input_row_group_indices, options, stream).front();
+}
+
+std::pair<std::vector<byte_range_info>, std::vector<byte_range_info>>
+hybrid_scan_reader::secondary_filters_byte_ranges(
+  cudf::host_span<size_type const> row_group_indices, parquet_reader_options const& options) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->secondary_filters_byte_ranges(input_row_group_indices, options);
+}
+
+std::vector<cudf::size_type> hybrid_scan_reader::filter_row_groups_with_dictionary_pages(
+  cudf::host_span<rmm::device_buffer> dictionary_page_data,
+  cudf::host_span<size_type const> row_group_indices,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  CUDF_EXPECTS(row_group_indices.size() == dictionary_page_data.size(),
+               "Mismatch in size of input row group indices and dictionary page device buffers");
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl
+    ->filter_row_groups_with_dictionary_pages(
+      dictionary_page_data, input_row_group_indices, options, stream)
+    .front();
+}
+
+std::vector<cudf::size_type> hybrid_scan_reader::filter_row_groups_with_bloom_filters(
+  cudf::host_span<rmm::device_buffer> bloom_filter_data,
+  cudf::host_span<size_type const> row_group_indices,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  CUDF_EXPECTS(row_group_indices.size() == bloom_filter_data.size(),
+               "Mismatch in size of input row group indices and bloom filter device buffers");
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl
+    ->filter_row_groups_with_bloom_filters(
+      bloom_filter_data, input_row_group_indices, options, stream)
+    .front();
+}
+
+std::pair<std::unique_ptr<cudf::column>, std::vector<thrust::host_vector<bool>>>
+hybrid_scan_reader::filter_data_pages_with_stats(cudf::host_span<size_type const> row_group_indices,
+                                                 parquet_reader_options const& options,
+                                                 rmm::cuda_stream_view stream,
+                                                 rmm::device_async_resource_ref mr) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->filter_data_pages_with_stats(input_row_group_indices, options, stream, mr);
+}
+
+[[nodiscard]] std::vector<byte_range_info> hybrid_scan_reader::filter_column_chunks_byte_ranges(
+  cudf::host_span<size_type const> row_group_indices, parquet_reader_options const& options) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->filter_column_chunks_byte_ranges(input_row_group_indices, options).first;
+}
+
+table_with_metadata hybrid_scan_reader::materialize_filter_columns(
+  cudf::host_span<thrust::host_vector<bool> const> data_page_mask,
+  cudf::host_span<size_type const> row_group_indices,
+  std::vector<rmm::device_buffer> column_chunk_buffers,
+  cudf::mutable_column_view row_mask,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->materialize_filter_columns(data_page_mask,
+                                           input_row_group_indices,
+                                           std::move(column_chunk_buffers),
+                                           row_mask,
+                                           options,
+                                           stream);
+}
+
+[[nodiscard]] std::vector<byte_range_info> hybrid_scan_reader::payload_column_chunks_byte_ranges(
+  cudf::host_span<size_type const> row_group_indices, parquet_reader_options const& options) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+  return _impl->payload_column_chunks_byte_ranges(input_row_group_indices, options).first;
+}
+
+table_with_metadata hybrid_scan_reader::materialize_payload_columns(
+  cudf::host_span<size_type const> row_group_indices,
+  std::vector<rmm::device_buffer> column_chunk_buffers,
+  cudf::column_view row_mask,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  auto const input_row_group_indices =
+    std::vector<std::vector<size_type>>{{row_group_indices.begin(), row_group_indices.end()}};
+
+  return _impl->materialize_payload_columns(
+    input_row_group_indices, std::move(column_chunk_buffers), row_mask, options, stream);
+}
+
+}  // namespace cudf::io::parquet::experimental