diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 8da2f535e961..744c8425fd82 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -167,6 +167,67 @@ class hybrid_scan_multifile { secondary_filters_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options) const; + /** + * @brief Builds a boolean survival column of size equal to the total number of rows in the row + * groups containing all `true` values + * + * @param row_group_indices Input per-source row group indices (one inner vector per source) + * @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 + * @return An all-true boolean (survival) column spanning all selected rows across all sources + */ + [[nodiscard]] std::unique_ptr build_all_true_row_mask( + cudf::host_span const> row_group_indices, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const; + + /** + * @brief Builds a boolean column indicating surviving rows using page-level statistics in the + * page index + * + * @param row_group_indices Input per-source row group indices (one inner vector per source) + * @param options Parquet reader options + * @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 + * @return A boolean column spanning all selected rows across all sources and indicating which + * filter column rows survive the statistics in the page index + */ + [[nodiscard]] std::unique_ptr build_row_mask_with_page_index_stats( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const; + + /** + * @brief Get byte ranges of column chunks of all (or selected) columns + * + * @param row_group_indices Input row group indices, one inner vector per source + * @param options Parquet reader options + * @return Pair of flattened byte ranges to column chunks of all (or selected) columns and their + * corresponding source indices + */ + [[nodiscard]] std::pair, std::vector> + all_column_chunks_byte_ranges(cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const; + + /** + * @brief Materializes all (or selected) columns and returns the final output table + * + * @param row_group_indices Input row group indices, one inner vector per source + * @param column_chunk_data Flattened device spans of column chunk data returned in the same order + * as `all_column_chunks_byte_ranges` + * @param options Parquet reader options + * @param stream CUDA stream used for device memory operations and kernel launches + * @param mr Device memory resource used to allocate the device memory for the output table + * @return Table of all materialized columns and metadata + */ + [[nodiscard]] table_with_metadata materialize_all_columns( + cudf::host_span const> row_group_indices, + cudf::host_span const> column_chunk_data, + parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const; + private: std::unique_ptr _impl; }; diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index d11ae1e8ddb9..4e2156bd13db 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -131,7 +131,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { rmm::cuda_stream_view stream); /** - * @copydoc cudf::io::experimental::hybrid_scan::build_all_true_row_mask + * @copydoc cudf::io::experimental::hybrid_scan_multifile::build_all_true_row_mask */ [[nodiscard]] std::unique_ptr build_all_true_row_mask( cudf::host_span const> row_group_indices, @@ -139,7 +139,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { rmm::device_async_resource_ref mr); /** - * @copydoc cudf::io::experimental::hybrid_scan::build_row_mask_with_page_index_stats + * @copydoc cudf::io::experimental::hybrid_scan_multifile::build_row_mask_with_page_index_stats */ [[nodiscard]] std::unique_ptr build_row_mask_with_page_index_stats( cudf::host_span const> row_group_indices, @@ -196,19 +196,14 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { rmm::device_async_resource_ref mr); /** - * @brief Fetches byte ranges for all (or selected) column chunks - * - * @param row_group_indices Input row groups indices - * @param options Parquet reader options - * @return Pair of a vector of byte ranges to column chunks of all (or selected) columns and a - * vector of their corresponding input source file indices + * @copydoc cudf::io::experimental::hybrid_scan_multifile::all_column_chunks_byte_ranges */ [[nodiscard]] std::pair, std::vector> all_column_chunks_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options); /** - * @copydoc cudf::io::experimental::hybrid_scan::materialize_all_columns + * @copydoc cudf::io::experimental::hybrid_scan_multifile::materialize_all_columns */ [[nodiscard]] table_with_metadata materialize_all_columns( cudf::host_span const> row_group_indices, diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index 59347367d223..75df106e7ebe 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -83,4 +83,43 @@ hybrid_scan_multifile::secondary_filters_byte_ranges( return _impl->secondary_filters_byte_ranges(row_group_indices, options); } +std::unique_ptr hybrid_scan_multifile::build_all_true_row_mask( + cudf::host_span const> row_group_indices, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const +{ + CUDF_FUNC_RANGE(); + return _impl->build_all_true_row_mask(row_group_indices, stream, mr); +} + +std::unique_ptr hybrid_scan_multifile::build_row_mask_with_page_index_stats( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const +{ + CUDF_FUNC_RANGE(); + return _impl->build_row_mask_with_page_index_stats(row_group_indices, options, stream, mr); +} + +std::pair, std::vector> +hybrid_scan_multifile::all_column_chunks_byte_ranges( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const +{ + CUDF_FUNC_RANGE(); + return _impl->all_column_chunks_byte_ranges(row_group_indices, options); +} + +table_with_metadata hybrid_scan_multifile::materialize_all_columns( + cudf::host_span const> row_group_indices, + cudf::host_span const> column_chunk_data, + parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const +{ + CUDF_FUNC_RANGE(); + return _impl->materialize_all_columns(row_group_indices, column_chunk_data, options, stream, mr); +} + } // namespace cudf::io::parquet::experimental diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index e709a52ace20..872e8060dad3 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -350,6 +350,7 @@ ConfigureTest( io/experimental/hybrid_scan_composer.cpp io/experimental/hybrid_scan_filters_test.cpp io/experimental/hybrid_scan_multifile_filters_test.cpp + io/experimental/hybrid_scan_multifile_test.cpp io/experimental/hybrid_scan_test.cpp io/parquet_common.cpp io/parquet_test.cpp diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index a64b9facb471..c31b49a7bd97 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -9,15 +9,22 @@ #include +#include #include #include #include #include #include +#include + #include #include +#include +#include +#include +#include /** * @brief Creates a strings column with a constant stringified value between 0 and 9999 @@ -35,6 +42,51 @@ cudf::test::strings_column_wrapper inline constant_strings(cudf::size_type value return cudf::test::strings_column_wrapper(elements, elements + num_ordered_rows); } +/** + * @brief Helper to construct a random list column + * + * @param gen Random engine + * @param is_str_nullable Whether the string column should be nullable + * @param is_list_nullable Whether the list column should be nullable + * + * @return Unique pointer to the constructed list column + */ +inline auto make_list_str_column(std::mt19937& gen, bool is_str_nullable, bool is_list_nullable) +{ + auto constexpr num_rows = num_ordered_rows; + auto constexpr string_per_row = 3; + auto constexpr num_string_rows = num_rows * string_per_row; + + std::vector strings{ + "abc", "x", "bananas", "gpu", "minty", "backspace", "", "cayenne", "turbine", "soft"}; + std::uniform_int_distribution uni(0, strings.size() - 1); + auto string_iter = cudf::detail::make_counting_transform_iterator( + 0, [&](cudf::size_type idx) { return strings[uni(gen)]; }); + + std::bernoulli_distribution bn(0.7f); + auto string_valids = cudf::detail::make_counting_transform_iterator( + 0, [&](int index) { return is_str_nullable ? bn(gen) : true; }); + cudf::test::strings_column_wrapper string_col{ + string_iter, string_iter + num_string_rows, string_valids}; + + auto offset_iter = cudf::detail::make_counting_transform_iterator( + 0, [](cudf::size_type idx) { return idx * string_per_row; }); + cudf::test::fixed_width_column_wrapper offsets(offset_iter, + offset_iter + num_rows + 1); + + auto list_valids = + cudf::detail::make_counting_transform_iterator(0, [&](int index) { return index % 100; }); + auto [null_mask, null_count] = [&]() { + if (is_list_nullable) { + return cudf::test::detail::make_null_mask(list_valids, list_valids + num_rows); + } else { + return std::make_pair(rmm::device_buffer{}, 0); + } + }(); + return cudf::make_lists_column( + num_rows, offsets.release(), string_col.release(), null_count, std::move(null_mask)); +} + /** * @brief Fail for types other than duration or timestamp */ diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp b/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp new file mode 100644 index 000000000000..2d033539324e --- /dev/null +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp @@ -0,0 +1,79 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include + +/** + * @brief Struct to hold multifile datasources and footer buffers along with their byte spans + */ +struct multifile_inputs { + /** + * @brief Construct datasources, datasource refs, and footer byte spans from source info + */ + explicit multifile_inputs(cudf::io::source_info const& source_info) + : datasources{cudf::io::make_datasources(source_info)} + { + datasource_refs.reserve(datasources.size()); + footer_buffers.reserve(datasources.size()); + footer_byte_spans.reserve(datasources.size()); + + for (auto const& datasource : datasources) { + datasource_refs.emplace_back(*datasource); + footer_buffers.emplace_back(cudf::io::parquet::fetch_footer_to_host(datasource_refs.back())); + footer_byte_spans.emplace_back(*footer_buffers.back()); + } + } + + std::vector> datasources; + std::vector> datasource_refs; + std::vector> footer_buffers; + std::vector> footer_byte_spans; +}; + +/** + * @brief Construct source info from host buffers + */ +template +cudf::io::source_info build_source_info(Buffers const& file_buffers) +{ + std::vector> spans; + spans.reserve(file_buffers.size()); + for (auto const& buf : file_buffers) { + spans.emplace_back(buf.data(), buf.size()); + } + return cudf::io::source_info(cudf::host_span>{spans}); +} + +/** + * @brief Fetch and set up page indexes for all sources in a multifile reader + */ +inline void setup_page_indexes(cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, + multifile_inputs const& inputs) +{ + auto const page_index_byte_ranges = reader.page_index_byte_ranges(); + std::vector> page_index_byte_spans; + page_index_byte_spans.reserve(page_index_byte_ranges.size()); + + auto const page_index_buffers = cudf::io::parquet::fetch_page_indexes_to_host( + cudf::host_span const>{inputs.datasource_refs}, + cudf::host_span{page_index_byte_ranges}); + std::transform(page_index_buffers.begin(), + page_index_buffers.end(), + std::back_inserter(page_index_byte_spans), + [](auto const& buffer) { return cudf::host_span{*buffer}; }); + + reader.setup_page_indexes( + cudf::host_span const>{page_index_byte_spans}); +} diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 29c66093b1cf..676adbadcc72 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -4,11 +4,13 @@ */ #include "hybrid_scan_common.hpp" +#include "hybrid_scan_multifile_common.hpp" #include #include #include +#include #include #include #include @@ -17,6 +19,7 @@ #include #include +#include #include #include #include @@ -24,29 +27,13 @@ namespace { /** - * @brief Struct to hold multifile datasources, and footer buffers along with their byte spans + * @brief Copy fixed-width column data to a host vector */ -struct multifile_inputs { - std::vector> datasources; - std::vector> footer_buffers; - std::vector> footer_byte_spans; -}; - -template -multifile_inputs build_multifile_inputs(Buffers const& file_buffers) +template +auto host_row_mask_data(cudf::column_view const& column, rmm::cuda_stream_view stream) { - multifile_inputs out; - out.datasources.reserve(file_buffers.size()); - out.footer_buffers.reserve(file_buffers.size()); - out.footer_byte_spans.reserve(file_buffers.size()); - for (auto const& buf : file_buffers) { - out.datasources.emplace_back(cudf::io::datasource::create(cudf::host_span( - reinterpret_cast(buf.data()), buf.size()))); - out.footer_buffers.emplace_back( - cudf::io::parquet::fetch_footer_to_host(*out.datasources.back())); - out.footer_byte_spans.emplace_back(*out.footer_buffers.back()); - } - return out; + return cudf::detail::make_host_vector( + cudf::device_span(column.data(), static_cast(column.size())), stream); } /** @@ -74,6 +61,21 @@ std::vector create_empty_parquet_with_stats() return buffer; } +/** + * @brief Build a scalar literal matching a filter column type + */ +template +auto make_scalar(cudf::size_type value, rmm::cuda_stream_view stream) +{ + if constexpr (cudf::is_timestamp()) { + return cudf::timestamp_scalar(T(typename T::duration(value)), true, stream); + } else if constexpr (cudf::is_duration()) { + return cudf::duration_scalar(T(value), true, stream); + } else { + return cudf::numeric_scalar(static_cast(value), true, stream); + } +} + } // namespace struct HybridScanMultifileFiltersTest : public cudf::test::BaseFixture {}; @@ -91,20 +93,23 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata) std::vector> file_buffers; file_buffers.reserve(num_sources); auto constexpr num_concat = 1; - srand(0xbad); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); - srand(0xf00d); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + auto constexpr seed = 0xbad; + std::transform(cuda::counting_iterator{seed}, + cuda::counting_iterator{seed + num_sources}, + std::back_inserter(file_buffers), + [](auto const src_seed) { + srand(src_seed); + return std::get<1>(create_parquet_with_stats()); + }); // Filtering AST - col0 < 100 - auto literal_value = - cudf::timestamp_scalar(T(typename T::duration(100)), true, cudf::get_default_stream()); + auto literal_value = make_scalar(100, cudf::get_default_stream()); auto literal = cudf::ast::literal(literal_value); auto col_ref_0 = cudf::ast::column_name_reference("col0"); auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref_0, literal); // Construct reader from footer bytes - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); cudf::io::parquet_reader_options options = cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); @@ -186,8 +191,6 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource) { using T = uint32_t; - srand(0xc0ffee); - // Create two parquet source. First one with non-zero rows and the second one with zero rows. auto constexpr num_sources = 2; std::vector> file_buffers; @@ -195,7 +198,7 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource) file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); file_buffers.emplace_back(create_empty_parquet_with_stats()); - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); cudf::io::parquet_reader_options options = cudf::io::parquet_reader_options::builder().build(); auto const reader = std::make_unique( @@ -224,14 +227,19 @@ TEST_F(HybridScanMultifileFiltersTest, ErrorFilterRowGroupsWithByteRanges) { using T = uint32_t; auto constexpr num_sources = 2; - srand(0xb47e); std::vector> file_buffers; file_buffers.reserve(num_sources); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + auto constexpr seed = 0xb47e; + std::transform(cuda::counting_iterator{seed}, + cuda::counting_iterator{seed + num_sources}, + std::back_inserter(file_buffers), + [](auto const src_seed) { + srand(src_seed); + return std::get<1>(create_parquet_with_stats()); + }); - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); // Setting `skip_bytes` or `num_bytes` is ambiguous when reading multiple sources. The reader is // expected to throw an exception if row groups are filtered using byte range in this case. @@ -266,15 +274,19 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats) // Two sources, each with 4 row groups and ascending strings in col2 std::vector> file_buffers; file_buffers.reserve(num_sources); - srand(0xc001); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); - srand(0xbeef); - file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + auto constexpr seed = 0xc001; + std::transform(cuda::counting_iterator{seed}, + cuda::counting_iterator{seed + num_sources}, + std::back_inserter(file_buffers), + [](auto const src_seed) { + srand(src_seed); + return std::get<1>(create_parquet_with_stats()); + }); - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); // Filter - col0 < 50 and col2 > "000010000" - auto literal_value0 = cudf::duration_scalar(T::rep(50), true, cudf::get_default_stream()); + auto literal_value0 = make_scalar(50, cudf::get_default_stream()); auto literal0 = cudf::ast::literal(literal_value0); auto col_ref0 = cudf::ast::column_reference(0); auto filter1 = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref0, literal0); @@ -314,3 +326,181 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats) EXPECT_TRUE(stats_filtered.front().empty()); EXPECT_TRUE(stats_filtered.back().empty()); } + +TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask) +{ + using T = uint64_t; + auto constexpr num_sources = 2; + + std::vector> file_buffers; + file_buffers.reserve(num_sources); + auto constexpr seed = 0xa11; + std::transform(cuda::counting_iterator{seed}, + cuda::counting_iterator{seed + num_sources}, + std::back_inserter(file_buffers), + [](auto const src_seed) { + srand(src_seed); + return std::get<1>(create_parquet_with_stats()); + }); + + auto inputs = multifile_inputs(build_source_info(file_buffers)); + + auto const options = cudf::io::parquet_reader_options::builder().build(); + auto const reader = std::make_unique( + inputs.footer_byte_spans, options); + + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + + auto test_all_true_row_mask = + [&](cudf::host_span const> row_group_indices) { + auto const row_mask = reader->build_all_true_row_mask(row_group_indices, stream, mr); + + EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8); + EXPECT_EQ(row_mask->size(), reader->total_rows_in_row_groups(row_group_indices)); + EXPECT_EQ(row_mask->null_count(), 0); + }; + + auto row_group_indices = std::vector>{{0, 2}, {1, 3}}; + test_all_true_row_mask(row_group_indices); + + row_group_indices = reader->all_row_groups(options); + test_all_true_row_mask(row_group_indices); +} + +template +struct HybridScanMultifilePageIndexRowMaskTest : public HybridScanMultifileFiltersTest {}; + +// Unsigned numeric types except booleans for page index stats tests +using SignedIntegralTypesNotBool = + cudf::test::ContainedIn>; +using PageIndexRowMaskTestTypes = + cudf::test::RemoveIf>; + +TYPED_TEST_SUITE(HybridScanMultifilePageIndexRowMaskTest, PageIndexRowMaskTestTypes); + +TYPED_TEST(HybridScanMultifilePageIndexRowMaskTest, BuildRowMaskWithPageIndexStats) +{ + using T = TypeParam; + auto constexpr num_sources = 4; + + std::vector> file_buffers; + file_buffers.reserve(num_sources); + auto constexpr seed = 0xa11b; + std::transform(cuda::counting_iterator{seed}, + cuda::counting_iterator{seed + num_sources}, + std::back_inserter(file_buffers), + [](auto const src_seed) { + srand(src_seed); + return std::get<1>(create_parquet_with_stats()); + }); + + auto inputs = multifile_inputs(build_source_info(file_buffers)); + + auto options = cudf::io::parquet_reader_options::builder().build(); + auto const reader = std::make_unique( + inputs.footer_byte_spans, options); + + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + + auto const input_row_group_indices = reader->all_row_groups(options); + + auto const test_filter_data_pages_with_stats = [&]( + cudf::ast::operation const& filter_expression, + cudf::size_type const expected_surviving_rows) { + options.set_filter(filter_expression); + reader->reset_column_selection(); + + auto const row_mask = + reader->build_row_mask_with_page_index_stats(input_row_group_indices, options, stream, mr); + + auto const expected_num_rows = reader->total_rows_in_row_groups(input_row_group_indices); + EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8); + EXPECT_EQ(row_mask->size(), expected_num_rows); + EXPECT_EQ(row_mask->null_count(), 0); + + auto const host_row_mask = host_row_mask_data(row_mask->view(), stream); + EXPECT_EQ(std::count(host_row_mask.begin(), host_row_mask.end(), true), + expected_surviving_rows); + }; + + // Calling the page-index row mask builder before setting up the page index should raise an error. + { + auto literal_value = make_scalar(100, stream); + auto const literal = cudf::ast::literal(literal_value); + auto const col_ref = cudf::ast::column_name_reference("col0"); + auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref, literal); + options.set_filter(filter_expression); + EXPECT_THROW(std::ignore = reader->build_row_mask_with_page_index_stats( + input_row_group_indices, options, stream, mr), + std::runtime_error); + } + + setup_page_indexes(*reader, inputs); + + // Filtering AST - table[0] < 100 + { + auto literal_value = make_scalar(100, stream); + auto const literal = cudf::ast::literal(literal_value); + auto const col_ref = cudf::ast::column_name_reference("col0"); + auto filter_expression = + cudf::ast::operation(cudf::ast::ast_operator::GREATER, literal, col_ref); + auto constexpr expected_surviving_rows = + num_sources * num_ordered_rows / (std::is_signed_v ? 4 : 2); + test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows); + } + + // Filtering AST - table[2] >= 10000 + { + auto literal_value = cudf::string_scalar("000010000", true, stream); + auto literal = cudf::ast::literal(literal_value); + auto col_ref = cudf::ast::column_name_reference("col2"); + auto filter_expression = + cudf::ast::operation(cudf::ast::ast_operator::GREATER_EQUAL, col_ref, literal); + auto constexpr expected_surviving_rows = + num_sources * num_ordered_rows / (std::is_signed_v ? 4 : 2); + test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows); + } + + // Filtering AST - table[0] < 50 AND table[2] < "000010000" + { + auto literal_value1 = make_scalar(50, stream); + auto const literal1 = cudf::ast::literal(literal_value1); + auto const col_ref1 = cudf::ast::column_name_reference("col0"); + auto filter_expression1 = + cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref1, literal1); + + auto literal_value2 = cudf::string_scalar("000010000", true, stream); + auto literal2 = cudf::ast::literal(literal_value2); + auto col_ref2 = cudf::ast::column_name_reference("col2"); + auto filter_expression2 = + cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref2, literal2); + + auto filter_expression = cudf::ast::operation( + cudf::ast::ast_operator::LOGICAL_AND, filter_expression1, filter_expression2); + auto constexpr expected_surviving_rows = num_sources * page_size_for_ordered_tests; + test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows); + } + + // Filtering AST - table[0] > 150 OR table[2] < "000005000" + { + auto literal_value1 = make_scalar(150, stream); + auto const literal1 = cudf::ast::literal(literal_value1); + auto const col_ref1 = cudf::ast::column_name_reference("col0"); + auto filter_expression1 = + cudf::ast::operation(cudf::ast::ast_operator::GREATER, col_ref1, literal1); + + auto literal_value2 = cudf::string_scalar("000005000", true, stream); + auto literal2 = cudf::ast::literal(literal_value2); + auto col_ref2 = cudf::ast::column_name_reference("col2"); + auto filter_expression2 = + cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref2, literal2); + + auto filter_expression = cudf::ast::operation( + cudf::ast::ast_operator::LOGICAL_OR, filter_expression1, filter_expression2); + auto constexpr expected_surviving_rows = 2 * num_sources * page_size_for_ordered_tests; + test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows); + } +} diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp new file mode 100644 index 000000000000..1273085f32a1 --- /dev/null +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp @@ -0,0 +1,216 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "hybrid_scan_common.hpp" +#include "hybrid_scan_multifile_common.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include + +namespace { + +/** + * @brief Group flattened column chunk byte ranges by their corresponding sources + */ +std::vector> column_chunks_byte_ranges_per_source( + std::pair, std::vector> const& + byte_ranges_and_source_map, + std::size_t num_sources) +{ + auto const& [byte_ranges, source_map] = byte_ranges_and_source_map; + CUDF_EXPECTS(byte_ranges.size() == source_map.size(), "Invalid source map size"); + + auto byte_ranges_per_source = + std::vector>(num_sources); + std::for_each(byte_ranges.begin(), + byte_ranges.end(), + [&, range_index = std::size_t{0}](auto const& range) mutable { + auto const source_index = source_map[range_index++]; + CUDF_EXPECTS(source_index >= 0 and static_cast(source_index) < + byte_ranges_per_source.size(), + "Invalid byte range source index"); + byte_ranges_per_source[source_index].push_back(range); + }); + return byte_ranges_per_source; +} + +/** + * @brief Helper to test multifile hybrid scan single-shot materialization + * + * Writes the input table to multiple parquet sources and compares `materialize_all_columns` output + * with the regular multi-source parquet reader. The filter expression used is `col0 >= 100`. + * + * @note The first column in the input table must be constructed with + * `cudf::test::ascending()` + */ +template +void test_hybrid_scan_multifile(std::vector const& columns, + bool case_sensitive_names = true) +{ + auto const table = cudf::table_view{columns}; + cudf::io::table_input_metadata expected_metadata(table); + expected_metadata.column_metadata[0].set_name("col0"); + + std::vector> parquet_buffers(num_sources); + for (auto& parquet_buffer : parquet_buffers) { + auto out_opts = + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&parquet_buffer}, table) + .metadata(expected_metadata) + .row_group_size_rows(num_rows) + .max_page_size_rows(page_size_for_ordered_tests) + .compression(cudf::io::compression_type::AUTO) + .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) + .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN); + cudf::io::write_parquet(out_opts); + } + + auto literal_value = cudf::numeric_scalar(100); + auto literal = cudf::ast::literal(literal_value); + auto col_ref_0 = cudf::ast::column_name_reference(case_sensitive_names ? "col0" : "Col0"); + auto filter_expression = + cudf::ast::operation(cudf::ast::ast_operator::GREATER_EQUAL, col_ref_0, literal); + + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + auto source_info = build_source_info(parquet_buffers); + auto inputs = multifile_inputs(source_info); + + auto const expected_options = cudf::io::parquet_reader_options::builder(source_info) + .filter(filter_expression) + .case_sensitive_names(case_sensitive_names) + .build(); + auto const expected = cudf::io::read_parquet(expected_options, stream, mr); + + auto const options = cudf::io::parquet_reader_options::builder() + .filter(filter_expression) + .case_sensitive_names(case_sensitive_names) + .build(); + auto reader = + cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options}; + auto const row_groups = reader.all_row_groups(options); + auto const byte_ranges_per_source = column_chunks_byte_ranges_per_source( + reader.all_column_chunks_byte_ranges(row_groups, options), inputs.datasources.size()); + auto [column_chunk_buffers, column_chunks_per_source, tasks] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + inputs.datasource_refs, + cudf::host_span const>{byte_ranges_per_source}, + stream, + mr); + tasks.get(); + (void)column_chunk_buffers; + + auto column_chunk_data = std::vector>{}; + for (auto const& source_column_chunks : column_chunks_per_source) { + column_chunk_data.insert( + column_chunk_data.end(), source_column_chunks.begin(), source_column_chunks.end()); + } + auto result = reader.materialize_all_columns(row_groups, column_chunk_data, options, stream, mr); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->view(), result.tbl->view()); +} + +} // namespace + +struct HybridScanMultifileTest : public cudf::test::BaseFixture {}; + +TEST_F(HybridScanMultifileTest, MaterializeLists) +{ + std::mt19937 gen(0xadd); + + auto constexpr num_rows = num_ordered_rows; + auto constexpr lists_per_row = 2; + auto constexpr max_vals_per_list = 3; + + auto col0 = testdata::ascending(); + auto col1 = make_parquet_list_col(gen, num_rows, max_vals_per_list, true); + auto col2 = + make_parquet_list_list_col(0, num_rows, lists_per_row, max_vals_per_list, true); + auto col3 = make_parquet_list_col(gen, num_rows, max_vals_per_list, false); + auto col4 = + make_parquet_list_list_col(0, num_rows, lists_per_row, max_vals_per_list, false); + auto col5 = make_parquet_list_list_col(0, num_rows, lists_per_row, max_vals_per_list, true); + + test_hybrid_scan_multifile({col0, *col1, *col2, *col3, *col4, *col5}); +} + +TEST_F(HybridScanMultifileTest, MaterializeListsOfStrings) +{ + std::mt19937 gen(0xc0c0a); + + auto col0 = testdata::ascending(); + auto col1 = make_list_str_column(gen, false, false); + auto col2 = make_list_str_column(gen, false, true); + auto col3 = make_list_str_column(gen, true, false); + auto col4 = make_list_str_column(gen, true, true); + + test_hybrid_scan_multifile({col0, *col1, *col2, *col3, *col4}, false); +} + +TEST_F(HybridScanMultifileTest, MaterializeStructs) +{ + std::mt19937 gen(0xbaLL); + + auto constexpr num_rows = num_ordered_rows; + + auto col0 = testdata::ascending(); + + std::bernoulli_distribution bn(0.7f); + auto valids = + cudf::detail::make_counting_transform_iterator(0, [&](int index) { return bn(gen); }); + auto struct_valids_iter = + cudf::detail::make_counting_transform_iterator(0, [&](int index) { return index % 121; }); + std::vector struct_valids(num_rows); + std::copy(struct_valids_iter, struct_valids_iter + num_rows, struct_valids.begin()); + + std::vector strings{ + "abc", "x", "bananas", "gpu", "minty", "backspace", "", "cayenne", "turbine", "soft"}; + std::uniform_int_distribution uni(0, strings.size() - 1); + auto string_iter = cudf::detail::make_counting_transform_iterator( + 0, [&](cudf::size_type idx) { return strings[uni(gen)]; }); + + auto values = cuda::counting_iterator{0}; + auto col1_list = make_list_str_column(gen, true, true); + cudf::test::fixed_width_column_wrapper col1_ints(values, values + num_rows, valids); + cudf::test::fixed_width_column_wrapper col1_floats(values, values + num_rows); + std::vector> col1_children; + col1_children.push_back(std::move(col1_list)); + col1_children.push_back(col1_ints.release()); + col1_children.push_back(col1_floats.release()); + cudf::test::structs_column_wrapper _col1(std::move(col1_children), struct_valids); + auto col1 = cudf::purge_nonempty_nulls(_col1); + + auto col2_str = cudf::test::strings_column_wrapper{string_iter, string_iter + num_rows, valids}; + auto col2_str_non_nullable = + cudf::test::strings_column_wrapper{string_iter, string_iter + num_rows}; + auto col2_bool = cudf::test::fixed_width_column_wrapper(values, values + num_rows, valids); + std::vector> col2_children; + col2_children.push_back(col2_str.release()); + col2_children.push_back(col2_str_non_nullable.release()); + col2_children.push_back(col2_bool.release()); + cudf::test::structs_column_wrapper _col2(std::move(col2_children)); + auto col2 = cudf::purge_nonempty_nulls(_col2); + + test_hybrid_scan_multifile({col0, *col1, *col2}); +} diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp index 05e7fba3d844..2ecb1ebfc1f7 100644 --- a/cpp/tests/io/experimental/hybrid_scan_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp @@ -30,52 +30,6 @@ namespace { -/** - * @brief Helper to construct a random list column - * - * @param gen Random engine - * @param is_str_nullable Whether the string column should be nullable - * @param is_list_nullable Whether the list column should be nullable - * - * @return Unique pointer to the constructed list column - */ -auto make_list_str_column(std::mt19937& gen, bool is_str_nullable, bool is_list_nullable) -{ - auto constexpr num_rows = num_ordered_rows; - auto constexpr string_per_row = 3; - auto constexpr num_string_rows = num_rows * string_per_row; - - // str and list helpers - std::vector strings{ - "abc", "x", "bananas", "gpu", "minty", "backspace", "", "cayenne", "turbine", "soft"}; - std::uniform_int_distribution uni(0, strings.size() - 1); - auto string_iter = cudf::detail::make_counting_transform_iterator( - 0, [&](cudf::size_type idx) { return strings[uni(gen)]; }); - - std::bernoulli_distribution bn(0.7f); - auto string_valids = cudf::detail::make_counting_transform_iterator( - 0, [&](int index) { return is_str_nullable ? bn(gen) : true; }); - cudf::test::strings_column_wrapper string_col{ - string_iter, string_iter + num_string_rows, string_valids}; - - auto offset_iter = cudf::detail::make_counting_transform_iterator( - 0, [](cudf::size_type idx) { return idx * string_per_row; }); - cudf::test::fixed_width_column_wrapper offsets(offset_iter, - offset_iter + num_rows + 1); - - auto list_valids = - cudf::detail::make_counting_transform_iterator(0, [&](int index) { return index % 100; }); - auto [null_mask, null_count] = [&]() { - if (is_list_nullable) { - return cudf::test::detail::make_null_mask(list_valids, list_valids + num_rows); - } else { - return std::make_pair(rmm::device_buffer{}, 0); - } - }(); - return cudf::make_lists_column( - num_rows, offsets.release(), string_col.release(), null_count, std::move(null_mask)); -} - /** * @brief Helper to test the hybrid scan reader *