diff --git a/cpp/src/io/parquet/page_decode.cuh b/cpp/src/io/parquet/page_decode.cuh index 3c24912095a..0e6a4dbd8a5 100644 --- a/cpp/src/io/parquet/page_decode.cuh +++ b/cpp/src/io/parquet/page_decode.cuh @@ -1458,10 +1458,15 @@ inline __device__ bool setup_local_page_info(auto* const s, /** * @brief Zero-fill null positions in output data using parallel per-validity-block processing * - * This function processes the validity bitmap and zero-fills all positions in the output - * data that correspond to null values. It uses a parallel approach where each thread - * handles one 32-bit validity block at a time, looping only over the zero bits (null positions) - * within that block. + * Each warp handles one 32-bit validity block, with each lane zero-filling a single null position. + * Remaining blocks that exceed the warp count are handled `process_block_sequential`. + * + * @note This handles only nulls in a leaf's own bitmap. Nulls inherited from `optional` + * ancestors are zero-filled by `reader_impl::allocate_columns` because an ancestor's validity map + * may not be available to this leaf. + * + * Callers use this for structural outputs: nullable string lengths, list offsets, and dictionary + * indices. Fixed-width null values need no initialization because they are masked. * * @tparam block_size CUDA block size for the kernel * @param s Page state containing all necessary information @@ -1481,7 +1486,7 @@ __device__ void zero_fill_null_positions_shared( int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto const& ni = s->nesting.nesting_info[leaf_level_index]; - // Check if we have nulls to fill + // Check if this leaf has a validity map to zero out nulls if ((ni.valid_map == nullptr) || (num_values == 0)) { return; } auto const data_out = ni.data_out; diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 72425abc59d..0041e40bd44 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -483,11 +483,12 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) auto const& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; if (ni.valid_map != nullptr) { int const num_values = ni.valid_map_offset - init_valid_map_offset; - zero_fill_null_positions_shared(s, - s->output_cvt.dtype_len, - init_valid_map_offset, - num_values, - static_cast(block.thread_rank())); + zero_fill_null_positions_shared( + s, + s->output_cvt.dtype_len, + init_valid_map_offset, + num_values, + static_cast(block.thread_rank())); } } diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index 923e7ee0d23..adf97205f45 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -948,14 +948,33 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_ // Validity Buffer is a uint32_t pointer std::vector> nullmask_bufs; + // An optional ancestor leaves unwritten output slots until the next repeated level. So, for a + // non-nullable STRING (FIELD) with a nullable ancestor, the column is nullable and not all rows + // will be decoded. The decoder may not detect this because it may not have a validity map from + // the ancestor. To avoid this, zero-fill such STRING buffers here as their uninitialized lengths + // are converted to offsets. No handling needed here for nullable strings (zero-filled by decoder + // using their own validity bitmap), fixed-width (masked), LIST offsets (never have gaps), and + // dictionary indices (have no ancestors). + auto const compute_has_unwritten_slots = [](auto const& out_buf, bool has_nullable_ancestor) { + return has_nullable_ancestor and out_buf.type.id() == type_id::STRING and + not out_buf.is_nullable; + }; + auto unwritten_bufs = cudf::detail::make_empty_pinned_vector>( + _input_columns.size(), _stream); + for (auto const& input_col : _input_columns) { size_t const max_depth = input_col.nesting_depth(); - auto* cols = &_output_buffers; + auto* cols = &_output_buffers; + bool has_nullable_ancestor = false; for (size_t l_idx = 0; l_idx < max_depth; l_idx++) { auto& out_buf = (*cols)[input_col.nesting[l_idx]]; cols = &out_buf.children; + auto const has_unwritten_slots = compute_has_unwritten_slots(out_buf, has_nullable_ancestor); + has_nullable_ancestor = + out_buf.type.id() == type_id::LIST ? false : (has_nullable_ancestor or out_buf.is_nullable); + // if this has a list parent, we have to get column sizes from the // data computed during compute_page_sizes if (out_buf.user_data & PARQUET_COLUMN_BUFFER_FLAG_HAS_LIST_PARENT) { @@ -976,6 +995,10 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_ out_buf.null_mask(), cudf::util::round_up_safe(out_buf.null_mask_size(), sizeof(cudf::bitmask_type)) / sizeof(cudf::bitmask_type)); + if (has_unwritten_slots and out_buf.data() != nullptr) { + unwritten_bufs.push_back( + {static_cast(out_buf.data()), out_buf.data_size()}); + } } } } @@ -1068,10 +1091,19 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_ for (size_type idx = 0; idx < static_cast(_input_columns.size()); idx++) { auto const& input_col = _input_columns[idx]; auto* cols = &_output_buffers; + // See the identically named variable in the non-list allocation loop above + bool has_nullable_ancestor = false; for (size_type l_idx = 0; l_idx < static_cast(input_col.nesting_depth()); l_idx++) { auto& out_buf = (*cols)[input_col.nesting[l_idx]]; cols = &out_buf.children; + + auto const has_unwritten_slots = + compute_has_unwritten_slots(out_buf, has_nullable_ancestor); + has_nullable_ancestor = out_buf.type.id() == type_id::LIST + ? false + : (has_nullable_ancestor or out_buf.is_nullable); + // if this buffer is part of a list hierarchy, we need to determine it's // final size and allocate it here. // @@ -1095,6 +1127,10 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_ out_buf.null_mask(), cudf::util::round_up_safe(out_buf.null_mask_size(), sizeof(cudf::bitmask_type)) / sizeof(cudf::bitmask_type)); + if (has_unwritten_slots and out_buf.data() != nullptr) { + unwritten_bufs.push_back( + {static_cast(out_buf.data()), out_buf.data_size()}); + } } } } @@ -1105,6 +1141,14 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_ cudf::host_span const>{nullmask_bufs}, _stream); cudf::detail::batched_memset( pinned_nullmask_bufs, std::numeric_limits::max(), _stream); + + // Need to zero non-nullable string lengths with nullable ancestors + if (not unwritten_bufs.empty()) { + cudf::detail::batched_memset( + cudf::host_span const>{unwritten_bufs}, + static_cast(0), + _stream); + } } void reader_impl::fill_pruned_offsets(size_t skip_rows, diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index a77ad1fd1de..043218125dd 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -39,6 +39,7 @@ #include #include #include +#include #include using ParquetDecompressionTest = DecompressionTest; @@ -6337,3 +6338,206 @@ TEST_F(ParquetReaderTest, NestedMismatchedSchemaColumnValidation) EXPECT_THROW(cudf::io::read_parquet(opts), std::invalid_argument); } } +namespace { + +/** + * @brief Create an optional struct with required children + * + * @param children Child columns without null masks + * @param num_rows Number of rows + * @param should_propagate_nulls Whether to push the struct's nulls down into its children + * @return Optional struct column with every seventh row null + */ +std::unique_ptr make_optional_struct( + std::vector>&& children, + cudf::size_type num_rows, + bool should_propagate_nulls) +{ + auto const validity = + cudf::detail::make_counting_transform_iterator(0, [](auto i) { return (i % 7) != 0; }); + auto [null_mask, null_count] = cudf::test::detail::make_null_mask(validity, validity + num_rows); + return should_propagate_nulls + ? cudf::make_structs_column( + num_rows, std::move(children), null_count, std::move(null_mask)) + : cudf::create_structs_hierarchy( + num_rows, std::move(children), null_count, std::move(null_mask)); +} + +} // namespace + +TEST_F(ParquetReaderTest, TwoRequiredStringLeavesWithNullableAncestor) +{ + // Build a table with an optional struct { required string, required string } column + // to test the case with two required strings sharing an immediate nullable ancestor. + + constexpr cudf::size_type num_rows = 2000; + constexpr auto const value = std::string_view{"fixed_width_payload"}; + + auto const values = cuda::make_constant_iterator(value); + cudf::test::strings_column_wrapper a_col{values, values + num_rows}; + cudf::test::strings_column_wrapper b_col{values, values + num_rows}; + + std::vector> children; + children.push_back(a_col.release()); + children.push_back(b_col.release()); + auto struct_col = make_optional_struct(std::move(children), num_rows, false); + + auto const filepath = + temp_env->get_temp_filepath("TwoRequiredStringLeavesWithNullableAncestor.parquet"); + + // Write the table to Parquet + { + auto const written = table_view{{struct_col->view()}}; + cudf::io::table_input_metadata input_metadata(written); + input_metadata.column_metadata[0].set_name("s"); + input_metadata.column_metadata[0].child(0).set_name("a").set_nullability(false); + input_metadata.column_metadata[0].child(1).set_name("b").set_nullability(false); + + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, written) + .metadata(std::move(input_metadata)) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .compression(cudf::io::compression_type::NONE) + .build()); + } + + // Build expected table from written struct's children + auto const expected = + make_optional_struct(std::move(struct_col->release().children), num_rows, true); + + // Read the table from Parquet + auto const result = cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}).build()); + + // Compare + CUDF_TEST_EXPECT_TABLES_EQUAL(table_view{{expected->view()}}, result.tbl->view()); +} + +TEST_F(ParquetReaderTest, RequiredStringLeafWithSeparatedNullableAncestor) +{ + // Build a table with an optional struct { required struct { required string } } column + // to test the case with an nullable ancestor of a required string separated by a required + // struct. + + constexpr cudf::size_type num_rows = 2000; + constexpr auto const value = std::string_view{"fixed_width_payload"}; + + auto const values = cuda::make_constant_iterator(value); + cudf::test::strings_column_wrapper child_col{values, values + num_rows}; + + std::vector> inner_children; + inner_children.push_back(child_col.release()); + auto inner_struct = + cudf::create_structs_hierarchy(num_rows, std::move(inner_children), 0, rmm::device_buffer{}); + + std::vector> outer_children; + outer_children.push_back(std::move(inner_struct)); + auto outer_struct = make_optional_struct(std::move(outer_children), num_rows, false); + + auto const filepath = + temp_env->get_temp_filepath("RequiredStringLeafWithSeparatedNullableAncestor.parquet"); + + // Write the table to Parquet + { + auto const written = table_view{{outer_struct->view()}}; + cudf::io::table_input_metadata input_metadata(written); + input_metadata.column_metadata[0] + .set_name("outer") + .child(0) + .set_name("inner") + .set_nullability(false) + .child(0) + .set_name("value") + .set_nullability(false); + + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, written) + .metadata(std::move(input_metadata)) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .compression(cudf::io::compression_type::NONE) + .build()); + } + + // Build expected table from written struct's children + auto const expected = + make_optional_struct(std::move(outer_struct->release().children), num_rows, true); + // Read the table from Parquet + auto const result = cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}).build()); + + // Compare + CUDF_TEST_EXPECT_TABLES_EQUAL(table_view{{expected->view()}}, result.tbl->view()); +} + +TEST_F(ParquetReaderTest, RequiredStringLeafWithNullableAncestorUnderList) +{ + // Build a table with a required list column to test the + // case with a nullable ancestor of a required string inside a list + + constexpr cudf::size_type num_lists = 500; + constexpr cudf::size_type list_size = 4; + constexpr cudf::size_type num_elements = num_lists * list_size; + constexpr auto const value = std::string_view{"fixed_width_payload"}; + + auto const values = cuda::make_constant_iterator(value); + cudf::test::strings_column_wrapper child_col{values, values + num_elements}; + + std::vector> children; + children.push_back(child_col.release()); + auto struct_col = make_optional_struct(std::move(children), num_elements, false); + + auto offsets = cudf::detail::make_counting_transform_iterator( + 0, [](auto i) { return static_cast(i * list_size); }); + column_wrapper offsets_col(offsets, offsets + num_lists + 1); + + auto list_col = cudf::make_lists_column( + num_lists, offsets_col.release(), std::move(struct_col), 0, rmm::device_buffer{}); + + auto const filepath = + temp_env->get_temp_filepath("RequiredStringLeafWithNullableAncestorUnderList.parquet"); + + // Write the table to Parquet + { + auto const written = table_view{{list_col->view()}}; + cudf::io::table_input_metadata input_metadata(written); + input_metadata.column_metadata[0] + .set_name("outer") + .child(1) + .set_name("inner") + .child(0) + .set_name("value") + .set_nullability(false); + + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, written) + .metadata(std::move(input_metadata)) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .compression(cudf::io::compression_type::NONE) + .build()); + } + + // Build expected table from written offsets and the leaf strings + auto written_contents = list_col->release(); + auto exp_offsets = + std::move(written_contents.children[cudf::lists_column_view::offsets_column_index]); + auto const written_struct = + written_contents.children[cudf::lists_column_view::child_column_index]->view(); + + auto const struct_validity = cudf::is_valid(written_struct); + auto exp_leaf = + cudf::copy_if_else(written_struct.child(0), cudf::string_scalar{""}, struct_validity->view()); + + std::vector> exp_children; + exp_children.push_back(std::move(exp_leaf)); + auto exp_struct = make_optional_struct(std::move(exp_children), num_elements, false); + + auto const expected = cudf::make_lists_column( + num_lists, std::move(exp_offsets), std::move(exp_struct), 0, rmm::device_buffer{}); + + // Read the table from Parquet + auto const result = cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}).build()); + + // Compare + CUDF_TEST_EXPECT_TABLES_EQUAL(table_view{{expected->view()}}, result.tbl->view()); +}