Improve Parquet writer struct test

cpp/src/io/parquet/chunk_dict.cu

@@ -133,8 +133,8 @@ struct map_insert_fn {
         size_type uniq_elem_size = 0;
 
         // Check if this index is valid.
-        auto const is_valid =
-          val_idx < end_value_idx and val_idx < data_col.size() and data_col.is_valid(val_idx);
+        auto const row      = frag->start_row + val_idx - start_value_idx;
+        auto const is_valid = val_idx < end_value_idx and is_valid_data(*col, row, val_idx);
 
         // Insert fragment index to hash map using a single thread (for best performance for now)
         // and count successful insertions.
@@ -215,7 +215,8 @@ struct map_find_fn {
   template <typename T>
   __device__ void operator()(size_type const start_value_idx,
                              size_type const end_value_idx,
-                             size_type const ck_start_val_idx)
+                             size_type const ck_start_val_idx,
+                             size_type const start_row)
   {
     if constexpr (column_device_view::has_element_accessor<T>()) {
       auto const col       = chunk->col_desc;
@@ -242,7 +243,8 @@ struct map_find_fn {
       // Note: Adjust the following loop to use `cg::tiles<map_cg_size>` if needed in the future.
       for (key_type val_idx = start_value_idx + t; val_idx < end_value_idx; val_idx += block_size) {
         // Find the key using a single thread for best performance for now.
-        if (data_col.is_valid(val_idx)) {
+        auto const row = start_row + val_idx - start_value_idx;
+        if (is_valid_data(*col, row, val_idx)) {
           auto const found_slot = map_find_ref.find(val_idx);
           // Fail if we didn't find the previously inserted key.
           cudf_assert(found_slot != map_find_ref.end() &&
@@ -414,7 +416,8 @@ CUDF_KERNEL void __launch_bounds__(block_size)
                   map_find_fn<block_size>{storage_ref, chunk},
                   start_value_idx,
                   end_value_idx,
-                  ck_start_val_idx);
+                  ck_start_val_idx,
+                  start_row);
 }
 
 /**
@@ -463,7 +466,8 @@ CUDF_KERNEL void __launch_bounds__(DEFAULT_BLOCK_SIZE)
     auto const val_idx = chunk_start_val + i;
     // Null rows leave `dict_index` undefined; gate the read with the column's validity bitmap to
     // avoid pulling garbage bits into the max.
-    if (val_idx < leaf_size && leaf_col.is_valid(val_idx)) {
+    auto const row = page.start_row + val_idx - page_start_val;
+    if (val_idx < leaf_size && is_valid_data(*col, row, val_idx)) {
       lane_max = cuda::std::max(lane_max, dict_index[i]);
     }
   }

cpp/src/io/parquet/page_enc.cu

@@ -164,8 +164,8 @@ void __device__ calculate_frag_size(frag_init_state_s* const s, int t)
   size_t len       = 0;
   for (uint32_t i = 0; i < nvals; i += block_size) {
     auto const val_idx  = start_value_idx + i + t;
-    auto const is_valid = i + t < nvals && val_idx < s->col.leaf_column->size() &&
-                          s->col.leaf_column->is_valid(val_idx);
+    auto const row      = s->frag.start_row + i + t;
+    auto const is_valid = i + t < nvals && is_valid_data(s->col, row, val_idx);
     if (is_valid) {
       num_valid++;
       len += dtype_len;
@@ -337,10 +337,8 @@ __device__ uint8_t const* delta_encode(page_enc_state_s<0>* s, uint64_t* buffer,
     size_type const val_idx_in_block = cur_val_idx + t;
     size_type const val_idx          = s->page_start_val + val_idx_in_block;
 
-    bool const is_valid =
-      (val_idx < s->col.leaf_column->size() && val_idx_in_block < s->page.num_leaf_values)
-        ? s->col.leaf_column->is_valid(val_idx)
-        : false;
+    bool const is_valid = val_idx_in_block < s->page.num_leaf_values &&
+                          is_valid_data(s->col, s->page.start_row + val_idx_in_block, val_idx);
 
     cur_val_idx += nvals;
 
@@ -1677,11 +1675,9 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
       } else {
         size_type const val_idx_in_leaf_col = s->page_start_val + val_idx_in_block;
 
-        is_valid = (val_idx_in_leaf_col < s->col.leaf_column->size() &&
-                    val_idx_in_block < s->page.num_leaf_values)
-                     ? s->col.leaf_column->is_valid(val_idx_in_leaf_col)
-                     : 0;
-        val_idx  = val_idx_in_leaf_col;
+        is_valid = val_idx_in_block < s->page.num_leaf_values &&
+                   is_valid_data(s->col, s->page.start_row + val_idx_in_block, val_idx_in_leaf_col);
+        val_idx = val_idx_in_leaf_col;
       }
       return cuda::std::make_tuple(is_valid, val_idx);
     }();
@@ -1916,10 +1912,9 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
       size_type const val_idx_in_block    = cur_val_idx + t;
       size_type const val_idx_in_leaf_col = s->page_start_val + val_idx_in_block;
 
-      uint32_t const is_valid = (val_idx_in_leaf_col < s->col.leaf_column->size() &&
-                                 val_idx_in_block < s->page.num_leaf_values)
-                                  ? s->col.leaf_column->is_valid(val_idx_in_leaf_col)
-                                  : 0;
+      uint32_t const is_valid =
+        val_idx_in_block < s->page.num_leaf_values &&
+        is_valid_data(s->col, s->page.start_row + val_idx_in_block, val_idx_in_leaf_col);
       // need to test for use_dictionary because it might be boolean
       uint32_t const val_idx =
         (s->ck.use_dictionary) ? val_idx_in_leaf_col - s->chunk_start_val : val_idx_in_leaf_col;
@@ -2136,7 +2131,7 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
     if (s->page.num_valid != 0) {
       for (uint32_t idx = 0; idx < s->page.num_leaf_values; idx++) {
         size_type const idx_in_col = s->page_start_val + idx;
-        if (s->col.leaf_column->is_valid(idx_in_col)) {
+        if (is_valid_data(s->col, s->page.start_row + idx, idx_in_col)) {
           if (type_id == type_id::STRING) {
             first_string = reinterpret_cast<uint8_t const*>(
               s->col.leaf_column->element<string_view>(idx_in_col).data());
@@ -2158,10 +2153,8 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
     size_type const val_idx_in_block = cur_val_idx + t;
     size_type const val_idx          = s->page_start_val + val_idx_in_block;
 
-    bool const is_valid =
-      (val_idx < s->col.leaf_column->size() && val_idx_in_block < s->page.num_leaf_values)
-        ? s->col.leaf_column->is_valid(val_idx)
-        : false;
+    bool const is_valid = val_idx_in_block < s->page.num_leaf_values &&
+                          is_valid_data(s->col, s->page.start_row + val_idx_in_block, val_idx);
 
     cur_val_idx += nvals;
 
@@ -2303,9 +2296,8 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
     // create the validity array
     for (int idx = t; idx < s->page.num_leaf_values; idx += block_size) {
       size_type const idx_in_col = s->page_start_val + idx;
-      bool const is_valid =
-        idx_in_col < s->col.leaf_column->size() and s->col.leaf_column->is_valid(idx_in_col);
-      forward_map[idx] = is_valid ? 1 : 0;
+      bool const is_valid        = is_valid_data(s->col, s->page.start_row + idx, idx_in_col);
+      forward_map[idx]           = is_valid ? 1 : 0;
     }
     __syncthreads();
 
@@ -2315,8 +2307,7 @@ CUDF_KERNEL void __launch_bounds__(block_size, 8)
     // now reverse map to get valid_idx -> leaf_idx mapping
     for (int idx = t; idx < s->page.num_leaf_values; idx += block_size) {
       size_type const idx_in_col = s->page_start_val + idx;
-      bool const is_valid =
-        idx_in_col < s->col.leaf_column->size() and s->col.leaf_column->is_valid(idx_in_col);
+      bool const is_valid        = is_valid_data(s->col, s->page.start_row + idx, idx_in_col);
       if (is_valid) { offsets_map[forward_map[idx]] = idx; }
     }
     __syncthreads();

cpp/src/io/parquet/parquet_gpu.cuh

@@ -84,6 +84,34 @@ inline size_type __device__ row_to_value_idx(size_type idx,
   return idx;
 }
 
+/**
+ * @brief Check whether a leaf value is defined by both its own validity and its struct ancestors.
+ */
+inline __device__ bool is_valid_data(parquet_column_device_view const& parquet_col,
+                                     size_type row,
+                                     size_type value_idx)
+{
+  auto const& leaf = *parquet_col.leaf_column;
+  if (value_idx >= leaf.size() or not leaf.is_valid(value_idx)) { return false; }
+
+  // List definition levels are precomputed by `get_dremel_data`. Their leaves may not map
+  // one-to-one to their ancestors so retain the existing leaf-validity behavior for this path.
+  if (parquet_col.level_offsets != nullptr) { return true; }
+
+  auto col        = *parquet_col.parent_column;
+  size_type level = 0;
+
+  // Walk down struct hierarchy and return false if any ancestor is null
+  while (col.type().id() == type_id::STRUCT) {
+    if (parquet_col.nullability[level] and not col.is_valid(row)) { return false; }
+    row += col.offset();  // rebase onto the sliced child
+    col = col.child(0);
+    ++level;
+  }
+
+  return true;
+}
+
 /**
  * @brief Insert chunk values into their respective hash maps
  *

cpp/tests/io/parquet_writer_test.cpp

@@ -335,6 +335,54 @@ TEST_F(ParquetWriterTest, Struct)
   cudf::io::read_parquet(read_args);
 }
 
+TEST_F(ParquetWriterTest, StructWithNonEmptyNulls)
+{
+  // Build a struct column with a required child and an optional child
+  constexpr cudf::size_type num_rows = 16;
+  auto const values                  = cuda::counting_iterator<int32_t>{100};
+  auto const validity =
+    cudf::detail::make_counting_transform_iterator(0, [] __device__(auto i) { return i % 4 != 0; });
+
+  auto required_child = cudf::test::fixed_width_column_wrapper<int32_t>(values, values + num_rows);
+  auto optional_child =
+    cudf::test::fixed_width_column_wrapper<int32_t>(values, values + num_rows, no_nulls());
+  std::vector<std::unique_ptr<cudf::column>> children;
+  children.push_back(required_child.release());
+  children.push_back(optional_child.release());
+  auto [mask, null_count] = cudf::test::detail::make_null_mask(validity, validity + num_rows);
+  auto struct_col =
+    cudf::create_structs_hierarchy(num_rows, std::move(children), null_count, std::move(mask));
+
+  // Write parquet with non-empty nulls
+  auto const input    = table_view({*struct_col});
+  auto const filepath = temp_env->get_temp_filepath("StructWithNonEmptyNulls.parquet");
+  {
+    cudf::io::table_input_metadata metadata(input);
+    metadata.column_metadata[0].child(0).set_nullability(false);
+    auto const write_args =
+      cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, input)
+        .metadata(std::move(metadata))
+        .dictionary_policy(cudf::io::dictionary_policy::NEVER)
+        .build();
+    cudf::io::write_parquet(write_args);
+  }
+
+  // Build expected table with propagated nulls into children columns
+  auto exp_child0 =
+    cudf::test::fixed_width_column_wrapper<int32_t>(values, values + num_rows, validity);
+  auto exp_child1 =
+    cudf::test::fixed_width_column_wrapper<int32_t>(values, values + num_rows, validity);
+  auto expected_col   = cudf::test::structs_column_wrapper({exp_child0, exp_child1}, validity);
+  auto const expected = table_view({expected_col});
+
+  // Read the written parquet file
+  auto const result = cudf::io::read_parquet(
+    cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)));
+
+  // Compare
+  CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
+}
+
 // custom data sink that supports device writes. uses plain file io.
 class custom_test_data_sink : public cudf::io::data_sink {
  public: