Skip to content
Closed
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions cpp/src/io/parquet/reader_impl_helpers.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2011,6 +2011,12 @@ aggregate_reader_metadata::select_columns(

cudf::io::detail::inline_column_buffer output_col(
dtype, schema_elem.repetition_type == FieldRepetitionType::OPTIONAL);
// A required element under an optional or repeated ancestor can be null by inheritance
// even though it has no validity of its own.
if (schema_elem.repetition_type != FieldRepetitionType::OPTIONAL and
schema_elem.max_definition_level > 0) {
output_col.set_may_have_inherited_nulls();
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
if (has_list_parent) { output_col.user_data |= PARQUET_COLUMN_BUFFER_FLAG_HAS_LIST_PARENT; }
// store the index of this element if inserted in out_col_array
nesting.push_back(static_cast<int>(out_col_array.size()));
Expand Down
14 changes: 10 additions & 4 deletions cpp/src/io/parquet/reader_impl_preprocess.cu
Original file line number Diff line number Diff line change
Expand Up @@ -970,8 +970,11 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_
CUDF_EXPECTS(out_buf_size <= std::numeric_limits<cudf::size_type>::max(),
"Number of rows exceeds cudf's column size limit",
std::overflow_error);
out_buf.create_with_mask(
out_buf_size, cudf::mask_state::UNINITIALIZED, false, _stream, _mr);
out_buf.create_with_mask(out_buf_size,
cudf::mask_state::UNINITIALIZED,
out_buf.may_have_inherited_nulls(),
_stream,
_mr);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
nullmask_bufs.emplace_back(
out_buf.null_mask(),
cudf::util::round_up_safe(out_buf.null_mask_size(), sizeof(cudf::bitmask_type)) /
Expand Down Expand Up @@ -1089,8 +1092,11 @@ void reader_impl::allocate_columns(read_mode mode, size_t skip_rows, size_t num_
std::overflow_error);
// allocate
// we're going to start null mask as all valid and then turn bits off if necessary
out_buf.create_with_mask(
buffer_size, cudf::mask_state::UNINITIALIZED, false, _stream, _mr);
out_buf.create_with_mask(buffer_size,
cudf::mask_state::UNINITIALIZED,
out_buf.may_have_inherited_nulls(),
_stream,
_mr);
nullmask_bufs.emplace_back(
out_buf.null_mask(),
cudf::util::round_up_safe(out_buf.null_mask_size(), sizeof(cudf::bitmask_type)) /
Expand Down
5 changes: 3 additions & 2 deletions cpp/src/io/utilities/column_buffer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,8 @@ void cudf::io::detail::inline_column_buffer::create_string_data(size_t num_bytes
namespace {

/**
* @brief Recursively copy `name`, `user_data`, and `string_as_binary` fields of one buffer to
* another.
* @brief Recursively copy `name`, `user_data`, `string_as_binary`, and the inherited-nulls flag
* of one buffer to another.
*
* @param buff The old output buffer
* @param new_buff The new output buffer
Expand All @@ -79,6 +79,7 @@ void copy_buffer_data(string_policy const& buff, string_policy& new_buff)
new_buff.name = buff.name;
new_buff.user_data = buff.user_data;
new_buff.string_as_binary = buff.string_as_binary;
if (buff.may_have_inherited_nulls()) { new_buff.set_may_have_inherited_nulls(); }
for (auto const& child : buff.children) {
auto& new_child = new_buff.children.emplace_back(string_policy(child.type, child.is_nullable));
copy_buffer_data(child, new_child);
Expand Down
9 changes: 8 additions & 1 deletion cpp/src/io/utilities/column_buffer.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,17 @@ class column_buffer_base {
rmm::device_async_resource_ref mr);

// Create a new column_buffer that has empty data but with the same basic information as the
// input column, including same type, nullability, name, and user_data.
// input column, including same type, nullability, name, user_data, and inherited-nulls flag.
static string_policy empty_like(string_policy const& input);

void set_null_mask(rmm::device_buffer&& mask) { _null_mask = std::move(mask); }

// A non-nullable buffer whose ancestors are nullable can still hold rows that are null via
// inheritance. Decode only writes the slots of present values, so the data array of such a
// buffer is zeroed at allocation to keep unwritten slots from exposing stale memory.
void set_may_have_inherited_nulls() { _may_have_inherited_nulls = true; }
[[nodiscard]] bool may_have_inherited_nulls() const { return _may_have_inherited_nulls; }

template <typename T = uint32_t>
auto null_mask()
{
Expand All @@ -154,6 +160,7 @@ class column_buffer_base {
rmm::device_buffer _data{};
rmm::device_buffer _null_mask{};
size_type _null_count{0};
bool _may_have_inherited_nulls{false};
rmm::device_async_resource_ref _mr{cudf::get_current_device_resource_ref()};

public:
Expand Down
133 changes: 133 additions & 0 deletions cpp/tests/io/parquet_reader_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@
#include <cudf_test/table_utilities.hpp>

#include <cudf/column/column.hpp>
#include <cudf/column/column_factories.hpp>
#include <cudf/copying.hpp>
#include <cudf/io/parquet.hpp>
#include <cudf/io/parquet_metadata.hpp>
#include <cudf/null_mask.hpp>
#include <cudf/reshape.hpp>
#include <cudf/stream_compaction.hpp>
#include <cudf/table/table.hpp>
Expand All @@ -36,6 +38,7 @@
#include <cstring>
#include <limits>
#include <memory>
#include <numeric>
#include <optional>
#include <stdexcept>
#include <utility>
Expand Down Expand Up @@ -1759,6 +1762,136 @@ TEST_F(ParquetReaderTest, StructByteArray)
CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
}

TEST_F(ParquetReaderTest, RequiredBinaryUnderNullStruct)
{
// A required BYTE_ARRAY leaf under an optional struct has no leaf validity of its own; rows
// where the struct is null are absent entirely. The reader must not leave the corresponding
// string length slots uninitialized before scanning them into offsets.
constexpr auto num_rows = 10;
constexpr auto null_row = 5;
constexpr auto row_chars = 4;

std::vector<uint8_t> payload_values(num_rows * row_chars);
for (cudf::size_type row = 0; row < num_rows; ++row) {
std::fill_n(
payload_values.data() + row * row_chars, row_chars, static_cast<uint8_t>('a' + row % 26));
}
auto const offsets_iter = cudf::detail::make_counting_transform_iterator(
0, [](cudf::size_type i) { return i * row_chars; });
cudf::test::fixed_width_column_wrapper<cudf::size_type> offsets(offsets_iter,
offsets_iter + num_rows + 1);
auto payload_values_col =
cudf::test::fixed_width_column_wrapper<uint8_t>(payload_values.begin(), payload_values.end());
auto payload =
cudf::make_lists_column(num_rows, offsets.release(), payload_values_col.release(), 0, {});
// the struct wrapper superimposes parent nulls onto children, so build the hierarchy directly
// to keep the payload child non-nullable, which is what makes it a required leaf in parquet
auto struct_mask = cudf::create_null_mask(num_rows, cudf::mask_state::ALL_VALID);
cudf::set_null_mask(
static_cast<cudf::bitmask_type*>(struct_mask.data()), null_row, null_row + 1, false);

std::vector<std::unique_ptr<cudf::column>> write_children;
write_children.push_back(std::move(payload));
auto write_struct =
cudf::create_structs_hierarchy(num_rows, std::move(write_children), 1, std::move(struct_mask));
auto const write_table = table_view{{write_struct->view()}};

cudf::io::table_input_metadata output_metadata(write_table);
output_metadata.column_metadata[0]
.set_name("s")
.child(0)
.set_name("payload")
.set_nullability(false)
.set_output_as_binary(true);

auto filepath = temp_env->get_temp_filepath("RequiredBinaryUnderNullStruct.parquet");
cudf::io::parquet_writer_options out_opts =
cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, write_table)
.metadata(std::move(output_metadata))
.dictionary_policy(cudf::io::dictionary_policy::NEVER)
.compression(cudf::io::compression_type::NONE);
cudf::io::write_parquet(out_opts);

// the leaf itself is required so it carries no validity of its own; only the struct is null
std::vector<std::string> expected_values(num_rows);
for (cudf::size_type row = 0; row < num_rows; ++row) {
expected_values[row] = std::string(row_chars, static_cast<char>('a' + row % 26));
}
expected_values[null_row].clear();
auto expected_strings =
cudf::test::strings_column_wrapper{expected_values.begin(), expected_values.end()};

auto expected_mask = cudf::create_null_mask(num_rows, cudf::mask_state::ALL_VALID);
cudf::set_null_mask(
static_cast<cudf::bitmask_type*>(expected_mask.data()), null_row, null_row + 1, false);
std::vector<std::unique_ptr<cudf::column>> expected_children;
expected_children.push_back(expected_strings.release());
auto expected_struct = cudf::create_structs_hierarchy(
num_rows, std::move(expected_children), 1, std::move(expected_mask));
auto const expected = table_view{{expected_struct->view()}};

cudf::io::parquet_reader_options in_opts =
cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath});
auto result = cudf::io::read_parquet(in_opts);

CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}

TEST_F(ParquetReaderTest, RequiredIntUnderNullStruct)
{
// same shape as RequiredBinaryUnderNullStruct with a fixed-width leaf: the value slot of an
// inherited-null row is never written by decode, so the buffer must come up zeroed
constexpr auto num_rows = 10;
constexpr auto null_row = 5;

auto const value_iter = cudf::detail::make_counting_transform_iterator(
0, [](cudf::size_type i) { return static_cast<int32_t>(i + 1); });
auto values = cudf::test::fixed_width_column_wrapper<int32_t>(value_iter, value_iter + num_rows);

auto struct_mask = cudf::create_null_mask(num_rows, cudf::mask_state::ALL_VALID);
cudf::set_null_mask(
static_cast<cudf::bitmask_type*>(struct_mask.data()), null_row, null_row + 1, false);
std::vector<std::unique_ptr<cudf::column>> write_children;
write_children.push_back(values.release());
auto write_struct =
cudf::create_structs_hierarchy(num_rows, std::move(write_children), 1, std::move(struct_mask));
auto const write_table = table_view{{write_struct->view()}};

cudf::io::table_input_metadata output_metadata(write_table);
output_metadata.column_metadata[0].set_name("s").child(0).set_name("n").set_nullability(false);

auto filepath = temp_env->get_temp_filepath("RequiredIntUnderNullStruct.parquet");
cudf::io::parquet_writer_options out_opts =
cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, write_table)
.metadata(std::move(output_metadata))
.dictionary_policy(cudf::io::dictionary_policy::NEVER)
.compression(cudf::io::compression_type::NONE);
cudf::io::write_parquet(out_opts);

// the leaf is required so the read-back child carries no validity; its slot at the
// inherited-null row is zero
std::vector<int32_t> expected_values(num_rows);
std::iota(expected_values.begin(), expected_values.end(), 1);
expected_values[null_row] = 0;
auto expected_values_col =
cudf::test::fixed_width_column_wrapper<int32_t>(expected_values.begin(), expected_values.end());

auto expected_mask = cudf::create_null_mask(num_rows, cudf::mask_state::ALL_VALID);
cudf::set_null_mask(
static_cast<cudf::bitmask_type*>(expected_mask.data()), null_row, null_row + 1, false);
std::vector<std::unique_ptr<cudf::column>> expected_children;
expected_children.push_back(expected_values_col.release());
auto expected_struct = cudf::create_structs_hierarchy(
num_rows, std::move(expected_children), 1, std::move(expected_mask));
auto const expected = table_view{{expected_struct->view()}};

cudf::io::parquet_reader_options in_opts =
cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath});
auto result = cudf::io::read_parquet(in_opts);

CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

TEST_F(ParquetReaderTest, NestingOptimizationTest)
{
// test nesting levels > cudf::io::parquet::detail::max_cacheable_nesting_decode_info deep.
Expand Down
Loading