io: zero parquet buffers that can hold inherited nulls

cpp/src/io/parquet/reader_impl_helpers.cpp

@@ -2011,6 +2011,13 @@ aggregate_reader_metadata::select_columns(
 
       cudf::io::detail::inline_column_buffer output_col(
         dtype, schema_elem.repetition_type == FieldRepetitionType::OPTIONAL);
+      // A required leaf under an optional or repeated ancestor can be null by inheritance even
+      // though it has no validity of its own. Non-leaf buffers either hold no data (structs) or
+      // have their offsets fully written during decode (lists), so only leaves need the fill.
+      if (schema_elem.repetition_type != FieldRepetitionType::OPTIONAL and
+          schema_elem.max_definition_level > 0 and schema_elem.num_children == 0) {
+        output_col.set_may_have_inherited_nulls();
+      }
       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()));

cpp/src/io/parquet/reader_impl_preprocess.cu

@@ -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);
         nullmask_bufs.emplace_back(
           out_buf.null_mask(),
           cudf::util::round_up_safe(out_buf.null_mask_size(), sizeof(cudf::bitmask_type)) /
@@ -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)) /

cpp/src/io/utilities/column_buffer.cpp

@@ -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
@@ -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);

cpp/src/io/utilities/column_buffer.hpp

@@ -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()
   {
@@ -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:

cpp/tests/io/parquet_reader_test.cpp

@@ -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>
@@ -36,6 +38,7 @@
 #include <cstring>
 #include <limits>
 #include <memory>
+#include <numeric>
 #include <optional>
 #include <stdexcept>
 #include <utility>
@@ -1759,6 +1762,154 @@ 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});
+  for (int read = 0; read < 2; ++read) {
+    // churn the pool with garbage so an unwritten slot holds stale bytes instead of a fresh
+    // zeroed page, which would let the test pass without the fix
+    for (auto size : {size_t{64}, size_t{4096}, size_t{1} << 16}) {
+      void* poison = nullptr;
+      CUDF_CUDA_TRY(cudaMalloc(&poison, size));
+      CUDF_CUDA_TRY(cudaMemset(poison, 0xEE, size));
+      CUDF_CUDA_TRY(cudaFree(poison));
+    }
+    auto result = cudf::io::read_parquet(in_opts);
+    CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
+  }
+}
+
+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});
+  for (int read = 0; read < 2; ++read) {
+    // churn the pool with garbage so an unwritten slot holds stale bytes instead of a fresh
+    // zeroed page, which would let the test pass without the fix
+    for (auto size : {size_t{64}, size_t{4096}, size_t{1} << 16}) {
+      void* poison = nullptr;
+      CUDF_CUDA_TRY(cudaMalloc(&poison, size));
+      CUDF_CUDA_TRY(cudaMemset(poison, 0xEE, size));
+      CUDF_CUDA_TRY(cudaFree(poison));
+    }
+    auto result = cudf::io::read_parquet(in_opts);
+    CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
+  }
+}
+
 TEST_F(ParquetReaderTest, NestingOptimizationTest)
 {
   // test nesting levels > cudf::io::parquet::detail::max_cacheable_nesting_decode_info deep.