diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 24992f447fab..3a1d8043d605 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -18,6 +18,8 @@ #include #include +#include +#include #include /** @@ -157,20 +159,64 @@ class hybrid_scan_multifile { cuda::stream_ref stream) const; /** - * @brief Get byte ranges of bloom filters and dictionary pages (secondary filters) for row group - * pruning + * @brief Get byte ranges of bloom filters for row group pruning * * @note Device buffers for bloom filter byte ranges must be allocated using a 32 byte * aligned memory resource * * @param row_group_indices Span of vectors of input row group indices, one per source * @param options Parquet reader options - * @return Pair of vectors of byte ranges of column chunk with bloom filters and dictionary - * pages subject to filter predicate + * @return A pair of vectors containing bloom filter byte ranges and corresponding source indices */ - [[nodiscard]] std::pair, std::vector> - secondary_filters_byte_ranges(cudf::host_span const> row_group_indices, - parquet_reader_options const& options) const; + [[nodiscard]] std::pair, std::vector> + bloom_filters_byte_ranges(std::span const> row_group_indices, + parquet_reader_options const& options) const; + + /** + * @brief Filter the row groups using column chunk bloom filters + * + * @param bloom_filter_data Device spans of header-stripped bloom filter bitsets of column + * chunks with an equality predicate, ordered to match the bloom filter + * byte ranges returned by `bloom_filters_byte_ranges` + * @param row_group_indices Input row group indices + * @param options Parquet reader options + * @param stream CUDA stream used for device memory operations and kernel launches + * @return Vectors of filtered per-source row group indices, one per source + */ + [[nodiscard]] std::vector> filter_row_groups_with_bloom_filters( + std::span const> bloom_filter_data, + std::span const> row_group_indices, + parquet_reader_options const& options, + cuda::stream_ref stream) const; + + /** + * @brief Get byte ranges of column chunk dictionary pages for row group pruning + * + * @param row_group_indices Span of vectors of input row group indices, one per source + * @param options Parquet reader options + * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter + * predicate and their corresponding source indices + */ + [[nodiscard]] std::pair, std::vector> + dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const; + + /** + * @brief Filter the row groups using column chunk dictionary pages + * + * @param dictionary_page_data Device spans of dictionary page data of column chunks with an + * (in)equality predicate, in the same order as the byte ranges returned by + * `dictionary_pages_byte_ranges` including empty spans against empty byte ranges + * @param row_group_indices Span of vectors of input row group indices, one per source + * @param options Parquet reader options + * @param stream CUDA stream used for device memory operations and kernel launches + * @return Vector of vectors of filtered row group indices, one per source + */ + [[nodiscard]] std::vector> filter_row_groups_with_dictionary_pages( + cudf::host_span const> dictionary_page_data, + cudf::host_span const> row_group_indices, + parquet_reader_options const& options, + cuda::stream_ref stream) const; /** * @brief Builds a boolean survival column of size equal to the total number of rows in the row @@ -455,35 +501,6 @@ class hybrid_scan_multifile { */ [[nodiscard]] bool has_next_table_chunk() const; - /** - * @brief Get byte ranges of column chunk dictionary pages for row group pruning - * - * @param row_group_indices Span of vectors of input row group indices, one per source - * @param options Parquet reader options - * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter - * predicate and their corresponding source indices - */ - [[nodiscard]] std::pair, std::vector> - dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, - parquet_reader_options const& options) const; - - /** - * @brief Filter the row groups using column chunk dictionary pages - * - * @param dictionary_page_data Device spans of dictionary page data of column chunks with an - * (in)equality predicate, in the same order as the byte ranges returned by - * `dictionary_pages_byte_ranges` including empty spans against empty byte ranges - * @param row_group_indices Span of vectors of input row group indices, one per source - * @param options Parquet reader options - * @param stream CUDA stream used for device memory operations and kernel launches - * @return Vector of vectors of filtered row group indices, one per source - */ - [[nodiscard]] std::vector> filter_row_groups_with_dictionary_pages( - cudf::host_span const> dictionary_page_data, - cudf::host_span const> row_group_indices, - parquet_reader_options const& options, - cuda::stream_ref stream) const; - private: std::unique_ptr _impl; }; diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index cda57fa2f294..708561aad26e 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -439,7 +439,8 @@ std::vector> aggregate_reader_metadata::filter_row_ return stats_filtered_row_group_indices.value_or(all_row_group_indices(row_group_indices)); } -std::vector aggregate_reader_metadata::get_bloom_filter_bytes( +std::pair, std::vector> +aggregate_reader_metadata::bloom_filters_byte_ranges( std::span const> row_group_indices, std::span output_dtypes, std::span output_column_schemas, @@ -463,7 +464,7 @@ std::vector aggregate_reader_metadata::get_bloom_filter_bytes( std::back_inserter(bloom_filter_col_schemas), [](auto& bloom_filter_literals) { return not bloom_filter_literals.empty(); }); - // No equality literals found, return empty vector + // No equality literals found, return empty pair if (bloom_filter_col_schemas.empty()) { return {}; } // Compute total number of input row groups @@ -476,36 +477,42 @@ std::vector aggregate_reader_metadata::get_bloom_filter_bytes( std::vector bloom_filter_bytes; bloom_filter_bytes.reserve(num_chunks); + // Parallel map identifying the source each emitted byte range must be fetched from + std::vector bloom_filter_source_map; + bloom_filter_source_map.reserve(num_chunks); + // Flag to check if we have at least one valid bloom filter offset auto have_bloom_filters = false; // For all sources - std::for_each(cuda::counting_iterator{0}, - cuda::counting_iterator{row_group_indices.size()}, - [&](auto const src_index) { - // Get all row group indices in the data source - auto const& rg_indices = row_group_indices[src_index]; - // For all row groups - std::for_each(rg_indices.cbegin(), rg_indices.cend(), [&](auto const rg_index) { - // For all column chunks - std::for_each( - bloom_filter_col_schemas.begin(), - bloom_filter_col_schemas.end(), - [&](auto const schema_idx) { - auto& col_meta = get_column_metadata(rg_index, src_index, schema_idx); - // Get bloom filter offsets and sizes - bloom_filter_bytes.emplace_back(col_meta.bloom_filter_offset.value_or(0), - col_meta.bloom_filter_length.value_or(0)); - - // Set `have_bloom_filters` if `bloom_filter_offset` is valid - if (col_meta.bloom_filter_offset.has_value()) { have_bloom_filters = true; } - }); - }); - }); + std::for_each( + cuda::counting_iterator{0}, + cuda::counting_iterator{row_group_indices.size()}, + [&](auto const src_index) { + // Get all row group indices in the data source + auto const& rg_indices = row_group_indices[src_index]; + // For all row groups + std::for_each(rg_indices.cbegin(), rg_indices.cend(), [&](auto const rg_index) { + // For all column chunks + std::for_each( + bloom_filter_col_schemas.begin(), + bloom_filter_col_schemas.end(), + [&](auto const schema_idx) { + auto& col_meta = get_column_metadata(rg_index, src_index, schema_idx); + // Get bloom filter offsets and sizes + bloom_filter_bytes.emplace_back(col_meta.bloom_filter_offset.value_or(0), + col_meta.bloom_filter_length.value_or(0)); + bloom_filter_source_map.emplace_back(static_cast(src_index)); + + // Set `have_bloom_filters` if `bloom_filter_offset` is valid + if (col_meta.bloom_filter_offset.has_value()) { have_bloom_filters = true; } + }); + }); + }); if (not have_bloom_filters) { return {}; } - return bloom_filter_bytes; + return {std::move(bloom_filter_bytes), std::move(bloom_filter_source_map)}; } std::pair, std::vector> @@ -699,6 +706,13 @@ aggregate_reader_metadata::filter_row_groups_with_bloom_filters( // Compute total number of input row groups auto const total_row_groups = compute_total_row_groups(row_group_indices); + // Ensure there is one bloom filter data span per eligible column in each row group + CUDF_EXPECTS(bloom_filter_data.size() == + static_cast(total_row_groups) * bloom_filter_col_schemas.size(), + "Bloom filter data size must match the number of row groups times the number of " + "columns with bloom filters and an equality predicate", + std::invalid_argument); + // Transform bloom filter data to cuda::std::byte type for apply_bloom_filters std::vector> transformed_bloom_filter_data; transformed_bloom_filter_data.reserve(bloom_filter_data.size()); diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp index d4b17613394f..135246802c35 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp @@ -226,13 +226,14 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base { * @param output_column_schemas schema indices of output columns * @param filter AST expression to filter row groups based on bloom filters * - * @return Byte ranges of bloom filters, one per column chunk with equality predicate + * @return A pair of vectors containing bloom filter byte ranges and corresponding source indices */ - [[nodiscard]] std::vector get_bloom_filter_bytes( - std::span const> row_group_indices, - std::span output_dtypes, - std::span output_column_schemas, - std::reference_wrapper filter); + [[nodiscard]] std::pair, + std::vector> + bloom_filters_byte_ranges(std::span const> row_group_indices, + std::span output_dtypes, + std::span output_column_schemas, + std::reference_wrapper filter); /** * @brief Get the dictionary page byte ranges, one per column chunk with (in)equality predicate diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 0b887eca7aad..c8259909e106 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -284,11 +284,14 @@ hybrid_scan_reader_impl::secondary_filters_byte_ranges( CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered"); auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options); + // Single source: keep only the bloom filter byte ranges, not the source map auto const bloom_filter_bytes = - _extended_metadata->get_bloom_filter_bytes(row_group_indices, - output_dtypes, - _output_column_schemas, - expr_conv.get_converted_expr().value()); + _extended_metadata + ->bloom_filters_byte_ranges(row_group_indices, + output_dtypes, + _output_column_schemas, + expr_conv.get_converted_expr().value()) + .first; auto const dictionary_page_bytes = _extended_metadata ->dictionary_pages_byte_ranges(row_group_indices, @@ -314,6 +317,19 @@ hybrid_scan_reader_impl::dictionary_pages_byte_ranges( expr_conv.get_converted_expr().value()); } +std::pair, std::vector> +hybrid_scan_reader_impl::bloom_filters_byte_ranges( + std::span const> row_group_indices, parquet_reader_options const& options) +{ + CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered"); + auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options); + + return _extended_metadata->bloom_filters_byte_ranges(row_group_indices, + output_dtypes, + _output_column_schemas, + expr_conv.get_converted_expr().value()); +} + std::vector> hybrid_scan_reader_impl::filter_row_groups_with_dictionary_pages( std::span const> dictionary_page_data, diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index 4c2bf670953d..e0e7d1160352 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -108,12 +108,19 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { cuda::stream_ref stream); /** - * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::secondary_filters_byte_ranges + * @copydoc cudf::io::parquet::experimental::hybrid_scan_reader::secondary_filters_byte_ranges */ [[nodiscard]] std::pair, std::vector> secondary_filters_byte_ranges(std::span const> row_group_indices, parquet_reader_options const& options); + /** + * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::bloom_filters_byte_ranges + */ + [[nodiscard]] std::pair, std::vector> + bloom_filters_byte_ranges(std::span const> row_group_indices, + parquet_reader_options const& options); + /** * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index 169146c6382d..96be4bf890f0 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -77,13 +77,24 @@ std::vector> hybrid_scan_multifile::filter_row_groups_wit return _impl->filter_row_groups_with_stats(row_group_indices, options, stream); } -std::pair, std::vector> -hybrid_scan_multifile::secondary_filters_byte_ranges( - cudf::host_span const> row_group_indices, +std::pair, std::vector> +hybrid_scan_multifile::bloom_filters_byte_ranges( + std::span const> row_group_indices, parquet_reader_options const& options) const { CUDF_FUNC_RANGE(); - return _impl->secondary_filters_byte_ranges(row_group_indices, options); + return _impl->bloom_filters_byte_ranges(row_group_indices, options); +} + +std::vector> hybrid_scan_multifile::filter_row_groups_with_bloom_filters( + std::span const> bloom_filter_data, + std::span const> row_group_indices, + parquet_reader_options const& options, + cuda::stream_ref stream) const +{ + CUDF_FUNC_RANGE(); + return _impl->filter_row_groups_with_bloom_filters( + bloom_filter_data, row_group_indices, options, stream); } std::unique_ptr hybrid_scan_multifile::build_all_true_row_mask( 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 4a42846d668a..9070272add8b 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -15,7 +15,6 @@ #include #include #include -#include #include #include #include @@ -23,8 +22,6 @@ #include #include -#include - #include #include #include @@ -483,6 +480,68 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats) EXPECT_TRUE(stats_filtered.back().empty()); } +TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithBloomFilters) +{ + using T = uint32_t; + auto constexpr num_sources = 32; + auto const stream = cudf::get_default_stream(); + + // num_sources sources, each with the same schema + std::vector> file_buffers; + file_buffers.reserve(num_sources); + for (int i = 0; i < num_sources; ++i) { + srand(0xb100 + i); + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + } + + auto inputs = multifile_inputs(build_source_info(file_buffers)); + + // An equality predicate makes col0 eligible for bloom filtering. cuDF's Parquet writer does not + // emit bloom filters, so the per-source bloom byte ranges come back empty (same as single-file). + { + auto literal_value = cudf::numeric_scalar(T{42}, true, stream); + auto literal = cudf::ast::literal(literal_value); + auto col_ref = cudf::ast::column_name_reference("col0"); + auto filter = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col_ref, literal); + + auto options = cudf::io::parquet_reader_options::builder().filter(filter).build(); + auto const reader = std::make_unique( + inputs.footer_byte_spans, options); + + auto const input_row_group_indices = reader->all_row_groups(options); + ASSERT_EQ(input_row_group_indices.size(), num_sources); + + auto const [bloom_byte_ranges, bloom_source_map] = + reader->bloom_filters_byte_ranges(input_row_group_indices, options); + // cuDF's Parquet writer does not emit bloom filters, so the ranges come back empty. The source + // map is parallel to the ranges and must always match them in length (here, both empty). + EXPECT_TRUE(bloom_byte_ranges.empty()); + EXPECT_EQ(bloom_byte_ranges.size(), bloom_source_map.size()); + } + + // Without any bloom-eligible (equality) predicate, bloom filtering is a no-op: the reader returns + // the input row groups unchanged, one inner vector per source. Validates the multifile bloom + // filter API delegation and per-source output shape. + { + auto literal_value = cudf::numeric_scalar(T{50}, true, stream); + auto literal = cudf::ast::literal(literal_value); + auto col_ref = cudf::ast::column_name_reference("col0"); + auto filter = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref, literal); + + auto options = cudf::io::parquet_reader_options::builder().filter(filter).build(); + auto const reader = std::make_unique( + inputs.footer_byte_spans, options); + + auto const input_row_group_indices = reader->all_row_groups(options); + ASSERT_EQ(input_row_group_indices.size(), num_sources); + + auto const empty_bloom_data = std::vector>{}; + auto const bloom_filtered = reader->filter_row_groups_with_bloom_filters( + empty_bloom_data, input_row_group_indices, options, stream); + EXPECT_EQ(bloom_filtered, input_row_group_indices); + } +} + TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask) { using T = uint64_t; diff --git a/python/cudf/cudf/tests/data/parquet/bloom_filter_alignment_desc_only.parquet b/python/cudf/cudf/tests/data/parquet/bloom_filter_alignment_desc_only.parquet new file mode 100644 index 000000000000..f7f4e6ce2970 Binary files /dev/null and b/python/cudf/cudf/tests/data/parquet/bloom_filter_alignment_desc_only.parquet differ diff --git a/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_stats.parquet b/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_stats.parquet new file mode 100644 index 000000000000..14ad1916831c Binary files /dev/null and b/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_stats.parquet differ diff --git a/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_with_length.parquet b/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_with_length.parquet new file mode 100644 index 000000000000..5b18eb69b7d2 Binary files /dev/null and b/python/cudf/cudf/tests/data/parquet/data_index_bloom_encoding_with_length.parquet differ diff --git a/python/cudf/cudf/tests/input_output/test_parquet.py b/python/cudf/cudf/tests/input_output/test_parquet.py index 7683452cc97b..4b0979e806e7 100644 --- a/python/cudf/cudf/tests/input_output/test_parquet.py +++ b/python/cudf/cudf/tests/input_output/test_parquet.py @@ -4820,6 +4820,48 @@ def test_parquet_bloom_filters_alignment(datadir, columns, memory_resource): assert_eq(expected, read) +@pytest.mark.parametrize( + "filename", + [ + "data_index_bloom_encoding_stats.parquet", + "data_index_bloom_encoding_with_length.parquet", + ], + ids=["length-absent", "length-present"], +) +@pytest.mark.parametrize("value", ["Hello", "not-in-this-file"]) +def test_parquet_bloom_filters_length(datadir, filename, value): + # Header may omit the bloom filter length. + # Source: apache/parquet-testing (Apache-2.0) + fname = datadir / filename + filters = [("String", "==", value)] + + expected = pq.read_table(fname, filters=filters) + read = cudf.read_parquet(fname, filters=filters).to_arrow() + + assert_eq(expected, read) + + +@pytest.mark.parametrize( + "predicate", + [ + [("r_reason_id", "==", "AAAAAAAABAAAAAAA")], # no bloom filter + [ + ("r_reason_desc", "==", "Did not like the color"), + ("r_reason_id", "==", "AAAAAAAAIAAAAAAA"), + ], # with and without a bloom filter + ], +) +def test_parquet_bloom_filters_mixed_presence(datadir, predicate): + # Source: same data as in bloom_filter_alignment.parquet, + # written with only r_reason_desc having a bloom filter + fname = datadir / "bloom_filter_alignment_desc_only.parquet" + + expected = pq.read_table(fname, filters=predicate) + read = cudf.read_parquet(fname, filters=predicate).to_arrow() + + assert_eq(expected, read) + + def test_parquet_reader_unsupported_compression(datadir): fname = datadir / "hadoop_lz4_compressed.parquet"