From 3c84c2391f9d2279694858ac96114dd3d9592b80 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:19:59 +0000 Subject: [PATCH 1/4] Report correct number of input RGs when all rows are pruned --- .../parquet/experimental/hybrid_scan_impl.cpp | 56 ++++++++++++--- .../io/experimental/hybrid_scan_test.cpp | 68 +++++++++++++++++++ 2 files changed, 116 insertions(+), 8 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index d572107609a3..01bacff3b543 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -39,10 +39,18 @@ using parquet::detail::PageNestingDecodeInfo; using text::byte_range_info; namespace { -// Tests the passed in logical type for a FIXED_LENGTH_BYTE_ARRAY column to see if it should -// be treated as a string. Currently the only logical type that has special handling is DECIMAL. -// Other valid types in the future would be UUID (still treated as string) and FLOAT16 (which -// for now would also be treated as a string). + +/** + * @brief Tests the logical type for a FLBA column to see if it should be + * treated as a string. + * + * Currently the only logical type that has special handling is DECIMAL. Other valid types in the + * future would be UUID (still treated as string) and FLOAT16 (which for now would also be treated + * as a string). + * + * @param logical_type The logical type to test + * @return Boolean indicating if the logical type should be treated as a string + */ [[maybe_unused]] inline bool is_treat_fixed_length_as_string( cuda::std::optional const& logical_type) { @@ -50,6 +58,12 @@ namespace { return logical_type->type != LogicalType::DECIMAL; } +/** + * @brief Get the output types from the output buffer template + * + * @param output_buffer_template Output buffer template + * @return Output types + */ [[nodiscard]] std::vector get_output_types( std::span output_buffer_template) { @@ -62,6 +76,22 @@ namespace { return output_dtypes; } +/** + * @brief Count the number of row groups in the input + * + * @param row_group_indices Row group indices + * @return Number of row groups + */ +[[nodiscard]] inline size_type count_row_groups( + cudf::host_span const> row_group_indices) +{ + return std::accumulate( + row_group_indices.begin(), + row_group_indices.end(), + size_type{0}, + [](auto sum, auto const& rgs) { return sum + static_cast(rgs.size()); }); +} + } // namespace hybrid_scan_reader_impl::hybrid_scan_reader_impl( @@ -335,7 +365,8 @@ hybrid_scan_reader_impl::filter_row_groups_with_dictionary_pages( auto const mr = cudf::get_current_device_resource_ref(); auto decompressed_dictionary_page_data = std::optional{}; if (has_compressed_data) { - // Use the `decompress_page_data` utility to decompress dictionary pages (passed as pass_pages) + // Use the `decompress_page_data` utility to decompress dictionary pages (passed as + // pass_pages) decompressed_dictionary_page_data = std::get<0>(parquet::detail::decompress_page_data(chunks, pages, {}, {}, stream, mr)); pages.host_to_device_async(stream); @@ -511,6 +542,8 @@ table_with_metadata hybrid_scan_reader_impl::materialize_filter_columns( auto const empty_row_groups = std::vector>(row_group_indices.size(), std::vector{}); prepare_data(read_mode::READ_ALL, empty_row_groups, {}, {}); + // Set correct number of input row groups to the output metadata + _file_itm_data.num_input_row_groups = count_row_groups(row_group_indices); return read_chunk_internal(read_mode::READ_ALL, read_columns_mode::FILTER_COLUMNS, row_mask); } @@ -547,6 +580,8 @@ table_with_metadata hybrid_scan_reader_impl::materialize_payload_columns( auto const empty_row_groups = std::vector>(row_group_indices.size(), std::vector{}); prepare_data(read_mode::READ_ALL, empty_row_groups, {}, {}); + // Set correct number of input row groups to the output metadata + _file_itm_data.num_input_row_groups = count_row_groups(row_group_indices); return read_chunk_internal(read_mode::READ_ALL, read_columns_mode::PAYLOAD_COLUMNS, row_mask); } @@ -616,6 +651,8 @@ void hybrid_scan_reader_impl::setup_chunking_for_filter_columns( auto const empty_row_groups = std::vector>(row_group_indices.size(), std::vector{}); prepare_data(read_mode::CHUNKED_READ, empty_row_groups, {}, {}); + // Set correct number of input row groups to the output metadata + _file_itm_data.num_input_row_groups = count_row_groups(row_group_indices); return; } @@ -674,6 +711,8 @@ void hybrid_scan_reader_impl::setup_chunking_for_payload_columns( auto const empty_row_groups = std::vector>(row_group_indices.size(), std::vector{}); prepare_data(read_mode::CHUNKED_READ, empty_row_groups, {}, {}); + // Set correct number of input row groups to the output metadata + _file_itm_data.num_input_row_groups = count_row_groups(row_group_indices); return; } @@ -959,7 +998,8 @@ table_with_metadata hybrid_scan_reader_impl::read_chunk_internal( // Copy number of total input row groups and number of surviving row groups from predicate // pushdown. out_metadata.num_input_row_groups = _file_itm_data.num_input_row_groups; - // Copy the number surviving row groups from each predicate pushdown only if the filter has value + // Copy the number surviving row groups from each predicate pushdown only if the filter has + // value if (_expr_conv.get_converted_expr().has_value()) { out_metadata.num_row_groups_after_stats_filter = _file_itm_data.surviving_row_groups.after_stats_filter; @@ -987,8 +1027,8 @@ table_with_metadata hybrid_scan_reader_impl::read_chunk_internal( // computes: // PageNestingInfo::batch_size for each level of nesting, for each page, taking row bounds into // account. PageInfo::skipped_values, which tells us where to start decoding in the input to - // respect the user bounds. It is only necessary to do this second pass if uses_custom_row_bounds - // is set (if the user has specified artificial bounds). + // respect the user bounds. It is only necessary to do this second pass if + // uses_custom_row_bounds is set (if the user has specified artificial bounds). if (uses_custom_row_bounds(mode)) { compute_page_sizes(subpass.pages, pass.chunks, diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp index f7cd45b926c3..9379bf8d295e 100644 --- a/cpp/tests/io/experimental/hybrid_scan_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp @@ -878,6 +878,74 @@ TEST_F(HybridScanTest, StructChildFilterColumn) std::invalid_argument); } +TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) +{ + using cudf::io::parquet::experimental::use_data_page_mask; + + auto constexpr num_rows = 10; + + // Single row group, single filter column (col0) and single payload column (col1) + auto values = cuda::counting_iterator{0}; + cudf::test::fixed_width_column_wrapper col0(values, values + num_rows); + cudf::test::fixed_width_column_wrapper col1(values, values + num_rows); + auto table = cudf::table_view{{col0, col1}}; + + std::string const filepath = temp_env->get_temp_filepath("AllRowsPruned.parquet"); + { + cudf::io::table_input_metadata input_metadata(table); + input_metadata.column_metadata[0].set_name("col0"); + auto out_opts = cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, table) + .metadata(std::move(input_metadata)) + .build(); + cudf::io::write_parquet(out_opts); + } + + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + + auto scalar = cudf::numeric_scalar(0, true, stream); + auto literal = cudf::ast::literal(scalar); + auto col_ref_0 = cudf::ast::column_name_reference("col0"); + auto filter_expression = + cudf::ast::operation(cudf::ast::ast_operator::GREATER_EQUAL, col_ref_0, literal); + + auto options = cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + + auto datasource = cudf::io::datasource::create(filepath); + auto const footer_buffer = cudf::io::parquet::fetch_footer_to_host(*datasource); + auto reader = + std::make_unique(*footer_buffer, options); + + auto const row_groups = reader->all_row_groups(options); + auto false_iter = cuda::make_constant_iterator(false); + auto row_mask = + cudf::test::fixed_width_column_wrapper(false_iter, false_iter + num_rows).release(); + auto row_mask_view = row_mask->mutable_view(); + + auto const filter_byte_ranges = reader->filter_column_chunks_byte_ranges(row_groups, options); + auto [filter_buffers, filter_data, filter_tasks] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + *datasource, filter_byte_ranges, stream, mr); + filter_tasks.get(); + + auto const filter_result = reader->materialize_filter_columns( + row_groups, filter_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); + EXPECT_EQ(filter_result.tbl->num_rows(), 0); + EXPECT_EQ(filter_result.metadata.num_input_row_groups, row_groups.size()); + + // Payload columns: same expectation + auto const payload_byte_ranges = reader->payload_column_chunks_byte_ranges(row_groups, options); + auto [payload_buffers, payload_data, payload_tasks] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + *datasource, payload_byte_ranges, stream, mr); + payload_tasks.get(); + + auto const payload_result = reader->materialize_payload_columns( + row_groups, payload_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); + EXPECT_EQ(payload_result.tbl->num_rows(), 0); + EXPECT_EQ(payload_result.metadata.num_input_row_groups, row_groups.size()); +} + TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass) { using T = uint32_t; From 0e66f69af6c118782156b4ce9dd74af5c904fa0a Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:26:58 +0000 Subject: [PATCH 2/4] Minor --- cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 01bacff3b543..c99f28c64457 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -83,7 +83,7 @@ namespace { * @return Number of row groups */ [[nodiscard]] inline size_type count_row_groups( - cudf::host_span const> row_group_indices) + std::span const> row_group_indices) { return std::accumulate( row_group_indices.begin(), @@ -365,8 +365,7 @@ hybrid_scan_reader_impl::filter_row_groups_with_dictionary_pages( auto const mr = cudf::get_current_device_resource_ref(); auto decompressed_dictionary_page_data = std::optional{}; if (has_compressed_data) { - // Use the `decompress_page_data` utility to decompress dictionary pages (passed as - // pass_pages) + // Use the `decompress_page_data` utility to decompress dictionary pages (passed as pass_pages) decompressed_dictionary_page_data = std::get<0>(parquet::detail::decompress_page_data(chunks, pages, {}, {}, stream, mr)); pages.host_to_device_async(stream); @@ -998,8 +997,7 @@ table_with_metadata hybrid_scan_reader_impl::read_chunk_internal( // Copy number of total input row groups and number of surviving row groups from predicate // pushdown. out_metadata.num_input_row_groups = _file_itm_data.num_input_row_groups; - // Copy the number surviving row groups from each predicate pushdown only if the filter has - // value + // Copy the number surviving row groups from each predicate pushdown only if the filter has value if (_expr_conv.get_converted_expr().has_value()) { out_metadata.num_row_groups_after_stats_filter = _file_itm_data.surviving_row_groups.after_stats_filter; @@ -1027,8 +1025,8 @@ table_with_metadata hybrid_scan_reader_impl::read_chunk_internal( // computes: // PageNestingInfo::batch_size for each level of nesting, for each page, taking row bounds into // account. PageInfo::skipped_values, which tells us where to start decoding in the input to - // respect the user bounds. It is only necessary to do this second pass if - // uses_custom_row_bounds is set (if the user has specified artificial bounds). + // respect the user bounds. It is only necessary to do this second pass if uses_custom_row_bounds + // is set (if the user has specified artificial bounds). if (uses_custom_row_bounds(mode)) { compute_page_sizes(subpass.pages, pass.chunks, From 470f8c1b077d0911c6a5543e58068abed73f0152 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Wed, 1 Jul 2026 17:14:52 +0000 Subject: [PATCH 3/4] Address comments --- cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp | 2 +- cpp/tests/io/experimental/hybrid_scan_test.cpp | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index c99f28c64457..08341df6c314 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -41,7 +41,7 @@ using text::byte_range_info; namespace { /** - * @brief Tests the logical type for a FLBA column to see if it should be + * @brief Tests the logical type for a fixed length byte array column to see if it should be * treated as a string. * * Currently the only logical type that has special handling is DECIMAL. Other valid types in the diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp index 9379bf8d295e..4d554651d463 100644 --- a/cpp/tests/io/experimental/hybrid_scan_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp @@ -931,7 +931,7 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) auto const filter_result = reader->materialize_filter_columns( row_groups, filter_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); EXPECT_EQ(filter_result.tbl->num_rows(), 0); - EXPECT_EQ(filter_result.metadata.num_input_row_groups, row_groups.size()); + EXPECT_EQ(filter_result.metadata.num_input_row_groups, static_cast(row_groups.size())); // Payload columns: same expectation auto const payload_byte_ranges = reader->payload_column_chunks_byte_ranges(row_groups, options); @@ -943,7 +943,7 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) auto const payload_result = reader->materialize_payload_columns( row_groups, payload_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); EXPECT_EQ(payload_result.tbl->num_rows(), 0); - EXPECT_EQ(payload_result.metadata.num_input_row_groups, row_groups.size()); + EXPECT_EQ(payload_result.metadata.num_input_row_groups, static_cast(row_groups.size())); } TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass) From 3a53fb90a47868594dfcc636fb8d4c8fe49497a9 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Wed, 1 Jul 2026 17:44:03 +0000 Subject: [PATCH 4/4] style --- cpp/tests/io/experimental/hybrid_scan_test.cpp | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp index 4d554651d463..e580994b6d34 100644 --- a/cpp/tests/io/experimental/hybrid_scan_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp @@ -931,7 +931,8 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) auto const filter_result = reader->materialize_filter_columns( row_groups, filter_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); EXPECT_EQ(filter_result.tbl->num_rows(), 0); - EXPECT_EQ(filter_result.metadata.num_input_row_groups, static_cast(row_groups.size())); + EXPECT_EQ(filter_result.metadata.num_input_row_groups, + static_cast(row_groups.size())); // Payload columns: same expectation auto const payload_byte_ranges = reader->payload_column_chunks_byte_ranges(row_groups, options); @@ -943,7 +944,8 @@ TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) auto const payload_result = reader->materialize_payload_columns( row_groups, payload_data, row_mask_view, use_data_page_mask::YES, options, stream, mr); EXPECT_EQ(payload_result.tbl->num_rows(), 0); - EXPECT_EQ(payload_result.metadata.num_input_row_groups, static_cast(row_groups.size())); + EXPECT_EQ(payload_result.metadata.num_input_row_groups, + static_cast(row_groups.size())); } TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass)