GH-51097: Fix Parquet null counts for fixed-width leaves

cpp/src/parquet/arrow/arrow_statistics_test.cc

@@ -160,6 +160,51 @@ INSTANTIATE_TEST_SUITE_P(
             /*expected_min=*/"z",
             /*expected_max=*/"z"}));
 
+TEST(StatisticsTest, FixedWidthLeafUnderListStructNullCount) {
+  // Null counts for fixed-width leaves under list<struct<...>>
+  // must include null and empty list entries from the repeated ancestor.
+  auto schema = ::arrow::schema({::arrow::field(
+      "col", ::arrow::list(::arrow::struct_(
+                    {::arrow::field("s", ::arrow::utf8()),
+                     ::arrow::field("i32", ::arrow::int32())})))});
+
+  auto table = ::arrow::Table::Make(
+      schema,
+      {::arrow::ArrayFromJSON(
+          ::arrow::list(::arrow::struct_(
+              {::arrow::field("s", ::arrow::utf8()),
+               ::arrow::field("i32", ::arrow::int32())})),
+          R"([[{"s":"a","i32":1}],null,[],[{"s":null,"i32":null},{"s":"b","i32":2}]])")});
+
+  std::shared_ptr<::arrow::ResizableBuffer> serialized_data = AllocateBuffer();
+  auto out_stream =
+      std::make_shared<::arrow::io::BufferOutputStream>(serialized_data);
+
+  ASSERT_OK_AND_ASSIGN(
+      std::unique_ptr<FileWriter> writer,
+      FileWriter::Open(*schema, default_memory_pool(), out_stream,
+                       default_writer_properties(),
+                       default_arrow_writer_properties()));
+  ASSERT_OK(writer->WriteTable(*table));
+  ASSERT_OK(writer->Close());
+  ASSERT_OK(out_stream->Close());
+
+  auto buffer_reader = std::make_shared<::arrow::io::BufferReader>(serialized_data);
+  auto parquet_reader = ParquetFileReader::Open(std::move(buffer_reader));
+  auto metadata = parquet_reader->metadata();
+  auto row_group = metadata->RowGroup(0);
+
+  ASSERT_EQ(row_group->num_columns(), 2);
+
+  auto int32_stats = row_group->ColumnChunk(1)->statistics();
+  ASSERT_NE(int32_stats, nullptr);
+
+  // Fixed-width leaves must include nulls from repeated ancestors
+  // (e.g. null or empty lists) in the column statistics.
+  EXPECT_EQ(int32_stats->null_count(), 3);
+  EXPECT_EQ(int32_stats->num_values(), 2);
+}
+
 TEST(StatisticsTest, TruncateOnlyHalfMinMax) {
   // GH-43382: Tests when we only have min or max, the `HasMinMax` should be false.
   std::shared_ptr<::arrow::ResizableBuffer> serialized_data = AllocateBuffer();

cpp/src/parquet/column_writer.cc

@@ -1394,21 +1394,22 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
       MaybeCalculateValidityBits(AddIfNotNull(def_levels, offset), batch_size,
                                  &batch_num_values, &batch_num_spaced_values,
                                  &null_count);
+      const int64_t total_null_count = batch_size - batch_num_values;
 
       WriteLevelsSpaced(batch_size, AddIfNotNull(def_levels, offset),
                         AddIfNotNull(rep_levels, offset));
       if (bits_buffer_ != nullptr) {
         WriteValuesSpaced(AddIfNotNull(values, value_offset), batch_num_values,
                           batch_num_spaced_values, bits_buffer_->data(), /*offset=*/0,
-                          /*num_levels=*/batch_size, null_count);
+                          /*num_levels=*/batch_size, total_null_count);
       } else {
         WriteValuesSpaced(AddIfNotNull(values, value_offset), batch_num_values,
                           batch_num_spaced_values, valid_bits,
                           valid_bits_offset + value_offset, /*num_levels=*/batch_size,
-                          null_count);
+                          total_null_count);
       }
-      CommitWriteAndCheckPageLimit(batch_size, batch_num_spaced_values, null_count,
-                                   check_page);
+      CommitWriteAndCheckPageLimit(batch_size, batch_num_spaced_values,
+                                   total_null_count, check_page);
       value_offset += batch_num_spaced_values;
 
       // Dictionary size checked separately from data page size since we
@@ -1750,6 +1751,7 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     internal::DefLevelsToBitmap(def_levels, batch_size, level_info_, &io);
     *out_values_to_write = io.values_read - io.null_count;
     *out_spaced_values_to_write = io.values_read;
+    // io.null_count excludes nulls from repeated ancestors.
     *null_count = io.null_count;
   }
 
@@ -2038,6 +2040,7 @@ Status TypedColumnWriterImpl<ParquetType>::WriteArrowDictionary(
     // had so we need to recompute it from def levels.
     MaybeCalculateValidityBits(AddIfNotNull(def_levels, offset), batch_size,
                                &batch_num_values, &batch_num_spaced_values, &null_count);
+    const int64_t total_null_count = batch_size - batch_num_values;
     WriteLevelsSpaced(batch_size, AddIfNotNull(def_levels, offset),
                       AddIfNotNull(rep_levels, offset));
     std::shared_ptr<Array> writeable_indices =
@@ -2051,7 +2054,8 @@ Status TypedColumnWriterImpl<ParquetType>::WriteArrowDictionary(
     dict_encoder->PutIndices(*writeable_indices);
     // Update unencoded byte array data size to size statistics
     UpdateUnencodedDataBytes();
-    CommitWriteAndCheckPageLimit(batch_size, batch_num_values, null_count, check_page);
+    CommitWriteAndCheckPageLimit(batch_size, batch_num_values, total_null_count,
+                                 check_page);
     value_offset += batch_num_spaced_values;
   };