diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 3a1d8043d605..c75fa3d186d3 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -307,6 +307,27 @@ class hybrid_scan_multifile { payload_column_chunks_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options) const; + /** + * @brief Get byte ranges of pages of payload columns + * + * Byte ranges are flattened in source, row group, column chunk, and page order. The returned + * source map has one source index per byte range. Dictionary pages precede data pages within each + * column chunk. Pruned pages are represented by empty byte ranges. + * + * @throws cudf::logic_error if any selected column chunk does not have a valid offset index + * + * @param row_group_indices Input row group indices, one vector per source + * @param row_mask Boolean mask spanning the selected row groups + * @param options Parquet reader options + * @param stream CUDA stream used to compute the page mask + * @return Pair of flattened payload page byte ranges and their corresponding source indices + */ + [[nodiscard]] std::pair, std::vector> + payload_pages_byte_ranges(cudf::host_span const> row_group_indices, + cudf::column_view const& row_mask, + parquet_reader_options const& options, + cuda::stream_ref stream) const; + /** * @brief Materialize payload columns and applies the row mask to the output table * @@ -430,6 +451,33 @@ class hybrid_scan_multifile { cuda::stream_ref stream, rmm::device_async_resource_ref mr) const; + /** + * @brief Setup chunking information for payload columns and preprocess the input data pages + * + * Input page data spans (including empty ones) must have the same shape as the byte ranges + * returned by `payload_pages_byte_ranges`. The data page mask is inferred from the data spans. + * + * @param chunk_read_limit Maximum bytes returned per output table chunk, or zero + * @param pass_read_limit Maximum read/decompression memory, or zero + * @param row_group_indices Input row group indices, one vector per source + * @param row_mask Boolean column spanning all selected rows across all sources and indicating + * which rows need to be read + * @param page_data Flattened device spans of payload page data in the same order as the byte + * ranges from `payload_pages_byte_ranges` + * @param options Parquet reader options + * @param stream CUDA stream used for preprocessing + * @param mr Device memory resource used for output table chunks + */ + void setup_chunking_for_payload_columns( + std::size_t chunk_read_limit, + std::size_t pass_read_limit, + cudf::host_span const> row_group_indices, + cudf::column_view const& row_mask, + cudf::host_span const> page_data, + parquet_reader_options const& options, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr) const; + /** * @brief Materializes a chunk of payload columns and applies the corresponding range of input row * mask to the output table chunk diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_chunking.cu b/cpp/src/io/parquet/experimental/hybrid_scan_chunking.cu index 99728cc00ae1..970f790da3e3 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_chunking.cu +++ b/cpp/src/io/parquet/experimental/hybrid_scan_chunking.cu @@ -38,10 +38,7 @@ void hybrid_scan_reader_impl::handle_chunking( // if this is our first time in here, setup the first pass. if (!_pass_itm_data) { // setup the next pass - setup_next_pass(column_chunk_data); - - // Must be called as soon as we create the pass - set_pass_page_mask(data_page_mask); + setup_next_pass(column_chunk_data, data_page_mask); } auto& pass = *_pass_itm_data; @@ -78,7 +75,8 @@ void hybrid_scan_reader_impl::handle_chunking( } void hybrid_scan_reader_impl::setup_next_pass( - std::span const> column_chunk_data) + std::span const> column_chunk_data, + std::span data_page_mask) { auto const num_passes = _file_itm_data.num_passes(); CUDF_EXPECTS(num_passes == 1, @@ -120,7 +118,16 @@ void hybrid_scan_reader_impl::setup_next_pass( pass.num_rows = _file_itm_data.global_num_rows; // Setup page information for the chunk (which we can access without decompressing) - setup_compressed_data(column_chunk_data); + if (_sparse_page_io) { + CUDF_EXPECTS(data_page_mask.empty(), + "Encountered a non-empty input data page mask in sparse I/O path.", + std::invalid_argument); + setup_sparse_compressed_data(column_chunk_data); + set_sparse_pass_page_mask(column_chunk_data); + } else { + setup_compressed_data(column_chunk_data); + set_pass_page_mask(data_page_mask); + } // detect malformed columns. // - we have seen some cases in the wild where we have a row group containing N @@ -148,12 +155,20 @@ void hybrid_scan_reader_impl::setup_next_pass( // store off how much memory we've used so far. This includes the compressed page data and the // decompressed dictionary data. we will subtract this from the available total memory for the // subpasses - auto chunk_iter = - cuda::transform_iterator(pass.chunks.d_begin(), parquet::detail::get_chunk_compressed_size{}); - pass.base_mem_size = - decomp_dict_data_size + - cudf::detail::reduce( + auto const compressed_data_size = [&] { + // In Sparse I/O case, compressed chunk size is the sum of its page data span sizes + if (_sparse_page_io) { + return std::accumulate(column_chunk_data.begin(), + column_chunk_data.end(), + std::size_t{0}, + [](auto size, auto const& page) { return size + page.size(); }); + } + auto chunk_iter = cuda::transform_iterator(pass.chunks.d_begin(), + parquet::detail::get_chunk_compressed_size{}); + return cudf::detail::reduce( chunk_iter, chunk_iter + pass.chunks.size(), size_t{0}, cuda::std::plus{}, _stream); + }(); + pass.base_mem_size = decomp_dict_data_size + compressed_data_size; // if we are doing subpass reading, generate more accurate num_row estimates for list columns. // this helps us to generate more accurate subpass splits. diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp index 135246802c35..1d061a3bc42d 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp @@ -47,19 +47,6 @@ struct metadata : public metadata_base { class aggregate_reader_metadata : public aggregate_reader_metadata_base { private: - /** - * @brief Check whether selected columns have column and offset indexes - * - * Schema indices are mapped to each source before locating the column chunks. - * - * @param row_group_indices Row group indices, one vector per source - * @param schema_indices Schema indices from the first source - * @return A pair indicating column-index and offset-index presence, respectively - */ - [[nodiscard]] std::pair page_index_presence( - std::span const> row_group_indices, - std::span schema_indices) const; - /** * @brief Filters the row groups using dictionary pages * @@ -91,6 +78,19 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base { cuda::stream_ref stream) const; public: + /** + * @brief Check whether selected columns have column and offset indexes + * + * Schema indices are mapped to each source before locating the column chunks. + * + * @param row_group_indices Row group indices, one vector per source + * @param schema_indices Schema indices from the first source + * @return A pair indicating column-index and offset-index presence, respectively + */ + [[nodiscard]] std::pair page_index_presence( + std::span const> row_group_indices, + std::span schema_indices) const; + /** * @brief Constructor for aggregate_reader_metadata * diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index c8259909e106..96ae63b61409 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -24,8 +24,10 @@ #include #include +#include #include #include +#include #include namespace cudf::io::parquet::experimental::detail { @@ -92,6 +94,28 @@ namespace { [](auto sum, auto const& rgs) { return sum + static_cast(rgs.size()); }); } +/** + * @brief Get the byte range of a column chunk's dictionary page, if present + * + * @param column Column chunk metadata with a valid offset index + * @return Dictionary page offset and size, or `std::nullopt` when no dictionary page is present + */ +[[nodiscard]] std::optional> dictionary_page_range( + ColumnChunk const& column) +{ + auto const& page_locations = column.offset_index->page_locations; + if (column.meta_data.dictionary_page_offset > 0) { + auto const offset = column.meta_data.dictionary_page_offset; + return std::pair{offset, column.meta_data.data_page_offset - offset}; + } + if (not page_locations.empty() and + column.meta_data.data_page_offset < page_locations.front().offset) { + auto const offset = column.meta_data.data_page_offset; + return std::pair{offset, page_locations.front().offset - offset}; + } + return std::nullopt; +} + } // namespace hybrid_scan_reader_impl::hybrid_scan_reader_impl( @@ -504,6 +528,125 @@ hybrid_scan_reader_impl::payload_column_chunks_byte_ranges( return get_input_column_chunk_byte_ranges(row_group_indices); } +std::pair, std::vector> +hybrid_scan_reader_impl::payload_pages_byte_ranges( + std::span const> row_group_indices, + cudf::column_view const& row_mask, + parquet_reader_options const& options, + cuda::stream_ref stream) +{ + CUDF_EXPECTS(row_group_indices.size() == _extended_metadata->get_num_sources(), + "Row group source count must match the number of input sources"); + CUDF_EXPECTS(std::cmp_equal(row_mask.size(), total_rows_in_row_groups(row_group_indices)), + "Row mask must span across all input row groups"); + CUDF_EXPECTS(row_mask.null_count() == 0, + "Row mask must not have any nulls when planning payload pages"); + + select_columns(read_columns_mode::PAYLOAD_COLUMNS, options); + + auto column_schemas = std::vector{}; + column_schemas.reserve(_input_columns.size()); + std::transform(_input_columns.begin(), + _input_columns.end(), + std::back_inserter(column_schemas), + [](auto const& col) { return col.schema_idx; }); + CUDF_EXPECTS(_extended_metadata->page_index_presence(row_group_indices, column_schemas).second, + "Page-level I/O for payload columns requires offset indexes to be present"); + + auto const num_columns = _input_columns.size(); + auto const num_chunks = + static_cast(count_row_groups(row_group_indices)) * num_columns; + + // The data page mask is ordered by column (all pages of a column, then the next column) so + // accumulate page counts per column to locate each column's portion of the mask. + auto mask_offsets = std::vector(num_columns + 1, 0); + + // For each source + for (std::size_t source_idx = 0; source_idx < row_group_indices.size(); ++source_idx) { + // For each row group in the source + auto colchunk_offsets = std::vector>(num_columns); + for (auto const row_group_idx : row_group_indices[source_idx]) { + // For each selected column chunk in the row group + auto const& row_group = _extended_metadata->get_row_group(row_group_idx, source_idx); + for (std::size_t col_idx = 0; col_idx < num_columns; ++col_idx) { + auto const schema_idx = + _extended_metadata->map_schema_index(column_schemas[col_idx], source_idx); + auto& colchunk_offset = colchunk_offsets[col_idx]; + colchunk_offset = + parquet::detail::find_colchunk_iter_offset(row_group, schema_idx, colchunk_offset); + // Accumulate page counts per column + mask_offsets[col_idx + 1] += + row_group.columns[colchunk_offset.value()].offset_index->page_locations.size(); + } + } + } + + // Accumulate page counts per column + std::partial_sum(mask_offsets.begin(), mask_offsets.end(), mask_offsets.begin()); + + // Compute the data page mask + auto const mask_size = mask_offsets.back(); + auto data_page_mask = _extended_metadata->compute_data_page_mask( + row_mask, row_group_indices, _input_columns, 0, stream); + CUDF_EXPECTS(data_page_mask.empty() or data_page_mask.size() == mask_size, + "Computed data page mask does not match offset indexes"); + + // Generate page ranges (row group wise) and the corresponding source map + auto page_ranges = std::vector{}; + auto source_map = std::vector{}; + page_ranges.reserve(mask_size + num_chunks); + source_map.reserve(mask_size + num_chunks); + + // For each source + for (std::size_t source_idx = 0; source_idx < row_group_indices.size(); ++source_idx) { + auto colchunk_offsets = std::vector>(num_columns); + // For each row group in the source + for (auto const row_group_idx : row_group_indices[source_idx]) { + auto const& row_group = _extended_metadata->get_row_group(row_group_idx, source_idx); + // For each selected column chunk in the row group + for (std::size_t col_idx = 0; col_idx < num_columns; ++col_idx) { + auto const schema_idx = + _extended_metadata->map_schema_index(column_schemas[col_idx], source_idx); + auto& colchunk_offset = colchunk_offsets[col_idx]; + colchunk_offset = + parquet::detail::find_colchunk_iter_offset(row_group, schema_idx, colchunk_offset); + auto const& column_chunk = row_group.columns[colchunk_offset.value()]; + auto const& page_locations = column_chunk.offset_index->page_locations; + auto const mask_offset = mask_offsets[col_idx]; + auto const any_data_page_retained = + data_page_mask.empty() or + std::any_of(data_page_mask.begin() + mask_offset, + data_page_mask.begin() + mask_offset + page_locations.size(), + cuda::std::identity{}); + + // Helper lambda to add a page's byte range and source index to the output vectors + auto add_page_range = [&](bool is_page_retained, int64_t offset, int64_t size) { + page_ranges.emplace_back(offset, is_page_retained ? size : 0); + source_map.push_back(static_cast(source_idx)); + }; + + // Add dictionary page range if any of the data pages are also retained + if (auto const dict_page_range = dictionary_page_range(column_chunk); + dict_page_range.has_value()) { + add_page_range(any_data_page_retained, dict_page_range->first, dict_page_range->second); + } + + // Add data page ranges + auto mask_iter = data_page_mask.cbegin() + mask_offset; + for (auto const& location : page_locations) { + add_page_range(data_page_mask.empty() or *mask_iter++, + location.offset, + static_cast(location.compressed_page_size)); + } + // Update the mask offset for the next column + mask_offsets[col_idx] += page_locations.size(); + } + } + } + + return {std::move(page_ranges), std::move(source_map)}; +} + std::pair, std::vector> hybrid_scan_reader_impl::all_column_chunks_byte_ranges( std::span const> row_group_indices, parquet_reader_options const& options) @@ -721,6 +864,56 @@ void hybrid_scan_reader_impl::setup_chunking_for_payload_columns( prepare_data(read_mode::CHUNKED_READ, row_group_indices, column_chunk_data, data_page_mask); } +void hybrid_scan_reader_impl::setup_chunking_for_payload_columns( + std::size_t chunk_read_limit, + std::size_t pass_read_limit, + std::span const> row_group_indices, + cudf::column_view const& row_mask, + std::span const> page_data, + parquet_reader_options const& options, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr) +{ + CUDF_EXPECTS(std::cmp_equal(row_mask.size(), total_rows_in_row_groups(row_group_indices)), + "Row mask must span across all input row groups"); + CUDF_EXPECTS(row_mask.null_count() == 0, + "Row mask must not have any nulls when materializing payload column"); + + prepare_materialization( + read_columns_mode::PAYLOAD_COLUMNS, row_group_indices.size(), options, stream, mr); + + _input_pass_read_limit = pass_read_limit; + _output_chunk_read_limit = chunk_read_limit; + + // Return early if all rows are pruned + if (are_all_rows_pruned(row_mask, stream)) { + 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; + } + + // Check if offset indexes are present + auto const num_columns = _input_columns.size(); + auto column_schemas = std::vector{}; + column_schemas.reserve(num_columns); + std::transform(_input_columns.begin(), + _input_columns.end(), + std::back_inserter(column_schemas), + [](auto const& col) { return col.schema_idx; }); + CUDF_EXPECTS(_extended_metadata->page_index_presence(row_group_indices, column_schemas).second, + "Page-level I/O for payload columns requires offset indexes to be present"); + + // Mark that we are using page-level I/O for payload columns + _sparse_page_io = true; + + // Data page mask in sparse mode will be computed directly from the page data span inside + // `prepare_data() -> setup_sparse_compressed_data()` + prepare_data(read_mode::CHUNKED_READ, row_group_indices, page_data, {}); +} + table_with_metadata hybrid_scan_reader_impl::materialize_payload_columns_chunk( cudf::column_view const& row_mask) { @@ -881,6 +1074,7 @@ void hybrid_scan_reader_impl::reset_internal_state() _pass_page_mask.clear(); _subpass_page_mask.reset(); _output_metadata.reset(); + _sparse_page_io = false; _options.timestamp_type = cudf::data_type{}; _options.decimal_width = type_id::EMPTY; @@ -1213,9 +1407,6 @@ void hybrid_scan_reader_impl::set_pass_page_mask(std::span data_page cuda::counting_iterator{_input_columns.size()}, [&](auto col_idx) { for (std::size_t chunk_idx = col_idx; chunk_idx < chunks.size(); chunk_idx += num_columns) { - // Insert a true value for each dictionary page - if (chunks[chunk_idx].num_dict_pages > 0) { _pass_page_mask.push_back(true); } - // Number of data pages in this column chunk auto const num_data_pages_this_col_chunk = chunks[chunk_idx].num_data_pages; @@ -1224,6 +1415,9 @@ void hybrid_scan_reader_impl::set_pass_page_mask(std::span data_page data_page_mask.size() >= num_inserted_data_pages + num_data_pages_this_col_chunk, "Encountered invalid data page mask size"); + // Insert a true value for each dictionary page + _pass_page_mask.insert(_pass_page_mask.end(), chunks[chunk_idx].num_dict_pages, true); + // Insert page mask for this column chunk _pass_page_mask.insert( _pass_page_mask.end(), @@ -1239,4 +1433,52 @@ void hybrid_scan_reader_impl::set_pass_page_mask(std::span data_page "Encountered mismatch in number of pass pages and page mask size"); } +void hybrid_scan_reader_impl::set_sparse_pass_page_mask( + std::span const> page_data) +{ + auto const& pass = _pass_itm_data; + auto const& chunks = pass->chunks; + + _pass_page_mask = cudf::detail::make_empty_host_vector(pass->pages.size(), _stream); + + // Find the first logical page-data span for every column chunk. + auto page_offsets = std::vector{}; + page_offsets.reserve(chunks.size()); + auto const num_logical_pages = std::accumulate( + chunks.begin(), chunks.end(), std::size_t{0}, [&](auto offset, auto const& chunk) { + page_offsets.push_back(offset); + return offset + chunk.num_dict_pages + chunk.num_data_pages; + }); + CUDF_EXPECTS(page_data.size() == num_logical_pages, + "Sparse page span count does not match the number of logical pages"); + + auto const num_columns = _input_columns.size(); + // Build the internal mask in column/chunk order. + std::for_each( + cuda::counting_iterator{0}, + cuda::counting_iterator{num_columns}, + [&](auto col_idx) { + for (std::size_t chunk_idx = col_idx; chunk_idx < chunks.size(); chunk_idx += num_columns) { + auto const& chunk = chunks[chunk_idx]; + auto const data_page_idx = page_offsets[chunk_idx] + chunk.num_dict_pages; + auto const data_page_end = data_page_idx + chunk.num_data_pages; + + // Retain a dictionary whenever the column chunk has a retained data page. + if (chunk.num_dict_pages > 0) { + _pass_page_mask.push_back(std::any_of(page_data.begin() + data_page_idx, + page_data.begin() + data_page_end, + [](auto const& page) { return not page.empty(); })); + } + // Insert page-mask values directly from the corresponding data-page spans. + std::transform(page_data.begin() + data_page_idx, + page_data.begin() + data_page_end, + std::back_inserter(_pass_page_mask), + [](auto const& page) { return not page.empty(); }); + } + }); + // Make sure we inserted exactly the number of pages for this pass. + CUDF_EXPECTS(_pass_page_mask.size() == pass->pages.size(), + "Encountered mismatch in number of pass pages and page mask size"); +} + } // namespace cudf::io::parquet::experimental::detail diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index e0e7d1160352..a6588f5c2ee8 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -200,6 +200,15 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { payload_column_chunks_byte_ranges(std::span const> row_group_indices, parquet_reader_options const& options); + /** + * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::payload_pages_byte_ranges + */ + [[nodiscard]] std::pair, std::vector> + payload_pages_byte_ranges(std::span const> row_group_indices, + cudf::column_view const& row_mask, + parquet_reader_options const& options, + cuda::stream_ref stream); + /** * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::materialize_payload_columns */ @@ -266,6 +275,20 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { cuda::stream_ref stream, rmm::device_async_resource_ref mr); + /** + * @copydoc + * cudf::io::parquet::experimental::hybrid_scan_multifile::setup_chunking_for_payload_columns + */ + void setup_chunking_for_payload_columns( + std::size_t chunk_read_limit, + std::size_t pass_read_limit, + std::span const> row_group_indices, + cudf::column_view const& row_mask, + std::span const> page_data, + parquet_reader_options const& options, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr); + /** * @copydoc * cudf::io::parquet::experimental::hybrid_scan_multifile::materialize_payload_columns_chunk @@ -353,10 +376,17 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { /** * @brief Set the page mask for the pass pages * - * @param data_page_mask Input data page mask from page-pruning step + * @param data_page_mask Input data page mask for the current pass */ void set_pass_page_mask(std::span data_page_mask); + /** + * @brief Set the page mask using sparse (page-level) data spans for the current pass + * + * @param page_data Span of device spans of sparse page data + */ + void set_sparse_pass_page_mask(std::span const> page_data); + /** * @brief Select the columns to be read based on the read mode * @@ -408,7 +438,8 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { * * @param mode Value indicating if the data sources are read all at once or chunk by chunk * @param row_group_indices Row group indices to read - * @param column_chunk_data Device spans of buffers containing column chunk data + * @param column_chunk_data Device spans containing column chunk data, or page data when sparse + * page I/O is enabled * @param data_page_mask Input data page mask from page-pruning step */ void prepare_data(read_mode mode, @@ -451,8 +482,9 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { * @brief Ratchet the pass/subpass/chunk process forward. * * @param mode Value indicating if the data sources are read all at once or chunk by chunk - * @param column_chunk_data Device spans of buffers containing column chunk data - * @param data_page_mask Input data page mask from page-pruning step for the current pass + * @param column_chunk_data Device spans containing column chunk data, or page data when sparse + * page I/O is enabled + * @param data_page_mask Input data page mask for the current pass */ void handle_chunking(read_mode mode, std::span const> column_chunk_data, @@ -464,9 +496,12 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { * A 'pass' is defined as a subset of row groups read out of the globally * requested set of all row groups. * - * @param column_chunk_data Device spans of buffers containing column chunk data + * @param column_chunk_data Device spans containing column chunk data, or page data when sparse + * page I/O is enabled + * @param data_page_mask Input data page mask for the current pass */ - void setup_next_pass(std::span const> column_chunk_data); + void setup_next_pass(std::span const> column_chunk_data, + std::span data_page_mask); /** * @brief Setup pointers to columns chunks to be processed for this pass. @@ -485,6 +520,13 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { */ void setup_compressed_data(std::span const> column_chunk_data); + /** + * @brief Setup sparse (page-level) data and decode page headers for the current pass. + * + * @param page_data Span of device spans of sparse page data + */ + void setup_sparse_compressed_data(std::span const> page_data); + /** * @brief Reset the internal state of the reader. */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index 96be4bf890f0..259435401aab 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -148,6 +148,17 @@ hybrid_scan_multifile::payload_column_chunks_byte_ranges( return _impl->payload_column_chunks_byte_ranges(row_group_indices, options); } +std::pair, std::vector> +hybrid_scan_multifile::payload_pages_byte_ranges( + cudf::host_span const> row_group_indices, + cudf::column_view const& row_mask, + parquet_reader_options const& options, + cuda::stream_ref stream) const +{ + CUDF_FUNC_RANGE(); + return _impl->payload_pages_byte_ranges(row_group_indices, row_mask, options, stream); +} + table_with_metadata hybrid_scan_multifile::materialize_payload_columns( cudf::host_span const> row_group_indices, cudf::host_span const> column_chunk_data, @@ -235,6 +246,21 @@ void hybrid_scan_multifile::setup_chunking_for_payload_columns( mr); } +void hybrid_scan_multifile::setup_chunking_for_payload_columns( + std::size_t chunk_read_limit, + std::size_t pass_read_limit, + cudf::host_span const> row_group_indices, + cudf::column_view const& row_mask, + cudf::host_span const> page_data, + parquet_reader_options const& options, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr) const +{ + CUDF_FUNC_RANGE(); + _impl->setup_chunking_for_payload_columns( + chunk_read_limit, pass_read_limit, row_group_indices, row_mask, page_data, options, stream, mr); +} + table_with_metadata hybrid_scan_multifile::materialize_payload_columns_chunk( cudf::column_view const& row_mask) const { diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_preprocess.cu b/cpp/src/io/parquet/experimental/hybrid_scan_preprocess.cu index 994176176a71..f2c56007b49b 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_preprocess.cu +++ b/cpp/src/io/parquet/experimental/hybrid_scan_preprocess.cu @@ -211,6 +211,42 @@ void hybrid_scan_reader_impl::setup_compressed_data( "Encountered page_offsets / num_columns mismatch"); } +void hybrid_scan_reader_impl::setup_sparse_compressed_data( + std::span const> page_data) +{ + auto& pass = *_pass_itm_data; + + // This function should never be called if `num_rows == 0`. + CUDF_EXPECTS(_pass_itm_data->num_rows > 0, "Number of reading rows must not be zero."); + CUDF_EXPECTS(_has_offset_index, "Sparse page I/O requires complete offset indexes"); + + auto& chunks = pass.chunks; + auto const total_pages = count_page_headers_with_pgidx(chunks, _stream); + CUDF_EXPECTS(total_pages == page_data.size(), + "Sparse page span count does not match page-index metadata"); + if (total_pages == 0) { return; } + + pass.has_compressed_data = false; + std::size_t page_idx = 0; + + for (auto const& chunk : chunks) { + auto const num_pages = chunk.num_data_pages + chunk.num_dict_pages; + auto const has_resident_page = std::any_of(page_data.begin() + page_idx, + page_data.begin() + page_idx + num_pages, + [](auto const& page) { return not page.empty(); }); + pass.has_compressed_data |= chunk.codec != Compression::UNCOMPRESSED and has_resident_page; + page_idx += num_pages; + } + + // `decode_page_headers` may not write every byte of each PageInfo, and `sort_pages` copies + // PageInfo as whole objects. + auto unsorted_pages = cudf::detail::make_zeroed_device_uvector_async( + total_pages, _stream, cudf::get_current_device_resource_ref()); + parquet::detail::decode_page_headers(pass, unsorted_pages, page_data, _stream); + CUDF_EXPECTS(pass.page_offsets.size() - 1 == static_cast(_input_columns.size()), + "Encountered page_offsets / num_columns mismatch"); +} + std::tuple, cudf::detail::hostdevice_vector> diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu index 1e88e996363c..7ec4aa859f0c 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter.cu +++ b/cpp/src/io/parquet/experimental/page_index_filter.cu @@ -1170,7 +1170,8 @@ thrust::host_vector aggregate_reader_metadata::compute_data_page_mask( // Copy over search results to host auto host_results = cudf::detail::make_pinned_vector_async(device_data_page_mask, stream); auto const total_pages = pinned_page_offsets.size() - num_columns; - auto data_page_mask = thrust::host_vector(total_pages); + auto data_page_mask = thrust::host_vector{}; + data_page_mask.reserve(total_pages); auto host_results_iter = host_results.begin(); stream.sync(); diff --git a/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp b/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp index a28f00c9aca1..cbd7ad66506d 100644 --- a/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp +++ b/cpp/src/io/parquet/io_utils/parquet_io_utils.cpp @@ -341,7 +341,8 @@ fetch_byte_ranges_to_device_async_impl( column_chunk_data.reserve(byte_ranges.size()); std::ignore = std::accumulate( byte_ranges.begin(), byte_ranges.end(), std::size_t{0}, [&](auto acc, auto const& range) { - column_chunk_data.emplace_back(buffer_data + acc, static_cast(range.size())); + auto const data = buffer_data == nullptr ? nullptr : buffer_data + acc; + column_chunk_data.emplace_back(data, static_cast(range.size())); return acc + range.size(); }); diff --git a/cpp/src/io/parquet/reader_impl.hpp b/cpp/src/io/parquet/reader_impl.hpp index 6552d268e79c..4c8cf0003b52 100644 --- a/cpp/src/io/parquet/reader_impl.hpp +++ b/cpp/src/io/parquet/reader_impl.hpp @@ -605,6 +605,9 @@ class reader_impl { // are offset indexes available for selected row groups bool _has_offset_index = false; + // whether sparse page I/O is enabled + bool _sparse_page_io = false; + std::optional> _reader_column_schema; // chunked reading happens in 2 parts: diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index fc35c234140f..3b3e7b4c7e0f 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -105,7 +105,8 @@ void reader_impl::build_string_dict_indices() rmm::exec_policy_nosync(_stream, cudf::get_current_device_resource_ref()), iter, iter + pass.chunks.size(), - set_str_dict_index_ptr{pass.str_dict_index.data(), str_dict_index_offsets, pass.chunks}); + set_str_dict_index_ptr{ + pass.str_dict_index.data(), str_dict_index_offsets, pass.chunks, _sparse_page_io}); // compute the indices kernel_error error_code(_stream); diff --git a/cpp/src/io/parquet/reader_impl_preprocess_utils.cu b/cpp/src/io/parquet/reader_impl_preprocess_utils.cu index cb6aa2fae662..aaba5c12219e 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess_utils.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess_utils.cu @@ -428,13 +428,13 @@ enum class page_data_source_type : uint8_t { * * @param pass Struct containing pass information * @param unsorted_pages Device span of page information to decode - * @param page_data Host span of page data spans (only used for PAGE_SPANS source) + * @param page_data Span of page data spans (only used for PAGE_SPANS source) * @param stream Stream to use */ template void decode_page_headers_impl(pass_intermediate_data& pass, device_span unsorted_pages, - host_span const> page_data, + std::span const> page_data, cuda::stream_ref stream) { CUDF_FUNC_RANGE(); @@ -640,7 +640,7 @@ void decode_page_headers(pass_intermediate_data& pass, void decode_page_headers(pass_intermediate_data& pass, device_span unsorted_pages, - host_span const> page_data, + std::span const> page_data, cuda::stream_ref stream) { decode_page_headers_impl( diff --git a/cpp/src/io/parquet/reader_impl_preprocess_utils.cuh b/cpp/src/io/parquet/reader_impl_preprocess_utils.cuh index af7c96f8316a..e140b64b2dd6 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess_utils.cuh +++ b/cpp/src/io/parquet/reader_impl_preprocess_utils.cuh @@ -14,6 +14,7 @@ #include #include +#include #include namespace cudf::io::parquet::detail { @@ -147,7 +148,7 @@ void decode_page_headers(pass_intermediate_data& pass, */ void decode_page_headers(pass_intermediate_data& pass, device_span unsorted_pages, - host_span const> page_data, + std::span const> page_data, cuda::stream_ref stream); /** @@ -224,10 +225,15 @@ struct set_str_dict_index_ptr { string_index_pair* const base; device_span str_dict_index_offsets; device_span chunks; + bool const sparse_page_io; __device__ constexpr inline void operator()(size_t i) { auto& chunk = chunks[i]; + // In Sparse I/O case, the dictionary page may be null if all data pages were pruned. + if (sparse_page_io and (chunk.dict_page == nullptr or chunk.dict_page->page_data == nullptr)) { + return; + } if (chunk.num_dict_pages > 0 and is_string_chunk(chunk)) { chunk.str_dict_index = base + str_dict_index_offsets[i]; } diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_composer.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_composer.cpp index a3a5a9337644..de22feb0da8f 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_composer.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_composer.cpp @@ -158,6 +158,72 @@ chunked_hybrid_scan_multifile(cudf::io::source_info const& source_info, concatenate_tables(std::move(payload_tables), stream, mr)}; } +std::tuple, std::unique_ptr> +chunked_sparse_hybrid_scan_multifile( + cudf::io::source_info const& source_info, + cudf::ast::operation const& filter_expression, + std::optional> const& payload_column_names, + bool case_sensitive_names, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr) +{ + auto options = cudf::io::parquet_reader_options::builder() + .filter(filter_expression) + .case_sensitive_names(case_sensitive_names) + .build(); + if (payload_column_names.has_value()) { options.set_column_names(payload_column_names.value()); } + + auto inputs = multifile_inputs(source_info); + auto reader = + cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options}; + setup_page_indexes(reader, inputs); + + auto const input_row_groups = reader.all_row_groups(options); + auto const row_groups = reader.filter_row_groups_with_stats(input_row_groups, options, stream); + auto row_mask = reader.build_row_mask_with_page_index_stats(row_groups, options, stream, mr); + + auto constexpr chunk_read_limit = std::size_t{256 * 1024}; + auto constexpr pass_read_limit = std::size_t{1024 * 1024}; + + auto filter_tables = std::vector>{}; + auto payload_tables = std::vector>{}; + + auto filter_column_chunks = fetch_multisource_device_data( + inputs, reader.filter_column_chunks_byte_ranges(row_groups, options), stream, mr); + auto row_mask_view = row_mask->mutable_view(); + reader.setup_chunking_for_filter_columns(chunk_read_limit, + pass_read_limit, + row_groups, + row_mask_view, + use_data_page_mask::YES, + filter_column_chunks.flat_spans, + options, + stream, + mr); + while (reader.has_next_table_chunk()) { + filter_tables.push_back(reader.materialize_filter_columns_chunk(row_mask_view).tbl); + } + + auto const payload_page_ranges = + reader.payload_pages_byte_ranges(row_groups, row_mask->view(), options, stream); + auto payload_page_data = fetch_multisource_device_data(inputs, payload_page_ranges, stream, mr); + + reader.setup_chunking_for_payload_columns(chunk_read_limit, + pass_read_limit, + row_groups, + row_mask->view(), + payload_page_data.flat_spans, + options, + stream, + mr); + while (reader.has_next_table_chunk()) { + payload_tables.push_back(reader.materialize_payload_columns_chunk(row_mask->view()).tbl); + } + + return std::tuple{concatenate_tables(std::move(filter_tables), stream, mr), + concatenate_tables(std::move(payload_tables), stream, mr)}; +} + std::unique_ptr chunked_hybrid_scan_multifile_single_step( cudf::io::source_info const& source_info, cudf::ast::operation const& filter_expression, diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_composer.hpp b/cpp/tests/io/experimental/hybrid_scan_multifile_composer.hpp index 95f530de442e..90bf42405a1e 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_composer.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_composer.hpp @@ -79,6 +79,30 @@ chunked_hybrid_scan_multifile(cudf::io::source_info const& source_info, cuda::stream_ref stream, rmm::device_async_resource_ref mr); +/** + * @brief Read parquet sources using chunked materialization and page-level payload I/O + * + * Filter columns continue to use the full-column-chunk path. Payload ranges are planned after + * filter materialization updates the row mask, then only requested pages are fetched. + * + * @param source_info Input source info containing one or more Parquet sources + * @param filter_expression Filter expression + * @param payload_column_names List of paths of select payload column names, if any + * @param case_sensitive_names Whether column names are case sensitive + * @param stream CUDA stream for hybrid scan reader + * @param mr Device memory resource + * + * @return Tuple of filter and payload tables + */ +std::tuple, std::unique_ptr> +chunked_sparse_hybrid_scan_multifile( + cudf::io::source_info const& source_info, + cudf::ast::operation const& filter_expression, + std::optional> const& payload_column_names, + bool case_sensitive_names, + cuda::stream_ref stream, + rmm::device_async_resource_ref mr); + /** * @brief Read parquet sources with the hybrid scan multifile reader in a single step using chunked * materialization 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 9070272add8b..6e7cdad91bde 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -587,6 +587,28 @@ TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask) test_all_true_row_mask(row_group_indices); } +TEST_F(HybridScanMultifileFiltersTest, SparsePayloadPagesWithoutOffsetIndexes) +{ + using T = uint32_t; + + auto file_buffers = std::vector>{}; + file_buffers.emplace_back(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().column_names({"col1"}).build(); + auto reader = + cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options}; + + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + auto const row_groups = reader.all_row_groups(options); + auto const row_mask = reader.build_all_true_row_mask(row_groups, stream, mr); + + EXPECT_THROW( + std::ignore = reader.payload_pages_byte_ranges(row_groups, row_mask->view(), options, stream), + cudf::logic_error); +} + template struct HybridScanMultifilePageIndexRowMaskTest : public HybridScanMultifileFiltersTest {}; diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp index 9a4d9daa2c97..47c8164468b6 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp @@ -12,17 +12,16 @@ #include #include +#include #include #include #include -#include -#include #include +#include #include #include #include #include -#include #include @@ -35,12 +34,14 @@ namespace { +using cudf::io::parquet::experimental::use_data_page_mask; + /** * @brief Helper to test multifile hybrid scan single-shot materialization * * Writes the input table to multiple parquet sources and compares filter, payload, and all-column - * materialization output with the regular multi-source parquet reader. The filter expression used - * is `col0 >= 100`. + * materialization output with the regular multi-source parquet reader. The filter expression is + * `col0 >= literal_value`. * * @note The first column in the input table must be constructed with * `cudf::test::ascending()` @@ -92,11 +93,16 @@ void test_hybrid_scan_multifile(std::vector const& columns, auto const [chunked_filter_table, chunked_payload_table] = chunked_hybrid_scan_multifile( source_info, filter_expression, {}, case_sensitive_names, stream, mr); + auto const [sparse_filter_table, sparse_payload_table] = chunked_sparse_hybrid_scan_multifile( + source_info, filter_expression, {}, case_sensitive_names, stream, mr); + auto const chunked_all_table = chunked_hybrid_scan_multifile_single_step( source_info, filter_expression, {}, case_sensitive_names, stream, mr); CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select({0}), filter_table->view()); CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select({0}), chunked_filter_table->view()); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select({0}), sparse_filter_table->view()); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(chunked_filter_table->view(), sparse_filter_table->view()); auto payload_column_indices = std::vector(columns.size() - 1); std::iota(payload_column_indices.begin(), payload_column_indices.end(), 1); @@ -104,6 +110,9 @@ void test_hybrid_scan_multifile(std::vector const& columns, payload_table->view()); CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select(payload_column_indices), chunked_payload_table->view()); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select(payload_column_indices), + sparse_payload_table->view()); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(chunked_payload_table->view(), sparse_payload_table->view()); CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->view(), all_table->view()); CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->view(), chunked_all_table->view()); } @@ -122,7 +131,7 @@ TEST_F(HybridScanMultifileTest, EmptyResult) auto col3 = make_list_str_column(gen, true, false); auto col4 = make_list_str_column(gen, true, true); - auto constexpr literal_value = uint32_t(num_ordered_rows); + auto constexpr literal_value = static_cast(num_ordered_rows); test_hybrid_scan_multifile({col0, *col1, *col2, *col3, *col4}, false, literal_value); } @@ -308,3 +317,158 @@ TEST_F(HybridScanMultifileTest, MaterializeStructs) test_hybrid_scan_multifile({col0, *col1, *col2}); } + +TEST_F(HybridScanMultifileTest, SparseDictionaryEncodedPages) +{ + auto constexpr num_sources = 2; + auto const [table, parquet_buffer] = create_parquet_with_stats(); + auto const payload_table = table->view().select({2}); + auto parquet_buffers = std::vector>(num_sources, parquet_buffer); + + auto const source_info = build_source_info(parquet_buffers); + auto const reader_options = + cudf::io::parquet_reader_options::builder().column_names({"col2"}).build(); + auto inputs = multifile_inputs(source_info); + auto reader = cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, + reader_options}; + setup_page_indexes(reader, inputs); + + auto const row_groups = reader.all_row_groups(reader_options); + auto row_mask_values = cudf::detail::make_counting_transform_iterator( + cudf::size_type{0}, [](auto i) { return (i / page_size_for_ordered_tests) % 2 == 0; }); + auto row_mask = cudf::test::fixed_width_column_wrapper( + row_mask_values, row_mask_values + reader.total_rows_in_row_groups(row_groups)); + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + + auto const page_ranges = + reader.payload_pages_byte_ranges(row_groups, row_mask, reader_options, stream); + auto page_data = fetch_multisource_device_data(inputs, page_ranges, stream, mr); + reader.setup_chunking_for_payload_columns( + 0, 0, row_groups, row_mask, page_data.flat_spans, reader_options, stream, mr); + + ASSERT_TRUE(reader.has_next_table_chunk()); + auto const result = reader.materialize_payload_columns_chunk(row_mask); + EXPECT_FALSE(reader.has_next_table_chunk()); + + auto const input = + cudf::concatenate(std::vector(num_sources, payload_table), stream, mr); + auto const expected = cudf::apply_boolean_mask(input->view(), row_mask, stream, mr); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->view(), result.tbl->view()); +} + +TEST_F(HybridScanMultifileTest, SparsePayloadWithAsymmetricRowGroupOrdering) +{ + using T = uint32_t; + + auto parquet_buffers = std::vector>{}; + parquet_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + // Name the descending column `col0` so this source retains earlier row groups. + parquet_buffers.emplace_back(std::get<1>(create_parquet_with_stats( + 100, cudf::io::compression_type::AUTO, {"col0", "col1", "col2"}, {1, 0, 2}))); + + auto const source_info = build_source_info(parquet_buffers); + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + auto scalar = cudf::numeric_scalar(75); + auto literal = cudf::ast::literal(scalar); + auto col_ref = cudf::ast::column_name_reference("col0"); + auto filter_expression = + cudf::ast::operation(cudf::ast::ast_operator::GREATER_EQUAL, col_ref, literal); + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + + auto inputs = multifile_inputs(source_info); + auto reader = + cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options}; + auto const row_groups = + reader.filter_row_groups_with_stats(reader.all_row_groups(options), options, stream); + EXPECT_EQ(row_groups, (std::vector>{{1, 2, 3}, {0, 1, 2}})); + + auto const expected = cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder(source_info).filter(filter_expression), stream, mr); + auto const [filter_table, payload_table] = + chunked_sparse_hybrid_scan_multifile(source_info, filter_expression, {}, true, stream, mr); + + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select({0}), filter_table->view()); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected.tbl->select({1, 2}), payload_table->view()); +} + +TEST_F(HybridScanMultifileTest, SparsePayloadEmptyAndAllPrunedPageData) +{ + using T = uint32_t; + + // Create two sources with page indexes for sparse payload materialization. + auto file_buffers = std::vector>{}; + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats())); + file_buffers.emplace_back(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().column_names({"col1"}).build(); + auto const stream = cudf::get_default_stream(); + auto const mr = cudf::get_current_device_resource_ref(); + + // Empty row-group selection accepts an empty outer page-data span. + { + auto reader = std::make_unique( + inputs.footer_byte_spans, options); + auto const row_groups = + std::vector>(inputs.footer_byte_spans.size()); + auto const empty_page_data = std::vector>{}; + auto false_scalar = cudf::numeric_scalar{false}; + auto row_mask = cudf::make_column_from_scalar(false_scalar, 0); + + EXPECT_NO_THROW(reader->setup_chunking_for_payload_columns( + 0, 0, row_groups, row_mask->view(), empty_page_data, options, stream, mr)); + ASSERT_TRUE(reader->has_next_table_chunk()); + auto const result = reader->materialize_payload_columns_chunk(row_mask->view()); + EXPECT_EQ(result.tbl->num_rows(), 0); + EXPECT_EQ(result.metadata.num_input_row_groups, 0); + EXPECT_FALSE(reader->has_next_table_chunk()); + } + + // An empty outer span is invalid when the selected row groups contain pages. + { + auto reader = std::make_unique( + inputs.footer_byte_spans, options); + setup_page_indexes(*reader, inputs); + + auto const row_groups = reader->all_row_groups(options); + auto const row_mask = reader->build_all_true_row_mask(row_groups, stream, mr); + auto const empty_page_data = std::vector>{}; + + EXPECT_THROW(reader->setup_chunking_for_payload_columns( + 0, 0, row_groups, row_mask->view(), empty_page_data, options, stream, mr), + cudf::logic_error); + } + + // An all-false row mask prunes every page and yields an empty payload table. + { + auto reader = std::make_unique( + inputs.footer_byte_spans, options); + setup_page_indexes(*reader, inputs); + + auto const row_groups = reader->all_row_groups(options); + auto false_scalar = cudf::numeric_scalar{false}; + auto row_mask = + cudf::make_column_from_scalar(false_scalar, reader->total_rows_in_row_groups(row_groups)); + auto const page_ranges = + reader->payload_pages_byte_ranges(row_groups, row_mask->view(), options, stream); + + ASSERT_FALSE(page_ranges.first.empty()); + EXPECT_TRUE(std::all_of(page_ranges.first.begin(), + page_ranges.first.end(), + [](auto const& range) { return range.is_empty(); })); + auto const all_pruned_page_data = + std::vector>(page_ranges.first.size()); + reader->setup_chunking_for_payload_columns( + 0, 0, row_groups, row_mask->view(), all_pruned_page_data, options, stream, mr); + ASSERT_TRUE(reader->has_next_table_chunk()); + auto const result = reader->materialize_payload_columns_chunk(row_mask->view()); + EXPECT_EQ(result.tbl->num_rows(), 0); + EXPECT_EQ(result.tbl->num_columns(), 1); + // Two sources with four row groups each + EXPECT_EQ(result.metadata.num_input_row_groups, 8); + EXPECT_FALSE(reader->has_next_table_chunk()); + } +}