update

Author: ArnavBalyan

cpp/src/parquet/metadata.cc

@@ -307,6 +307,19 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
     possible_encoded_stats_ = nullptr;
     possible_geo_stats_ = nullptr;
     InitKeyValueMetadata();
+
+    // Per the Parquet specification, a column with max_definition_level == 0
+    // is required at every level and therefore cannot contain null values.
+    // Reject inconsisten metadata from writers violating this invariant.
+    if (descr_->max_definition_level() == 0 && column_metadata_->__isset.statistics &&
+        column_metadata_->statistics.__isset.null_count &&
+        column_metadata_->statistics.null_count > 0) {
+      std::stringstream ss;
+      ss << "Malformed Parquet file: column '" << descr_->path()->ToDotString()
+         << "' has max_definition_level == 0 but statistics report null_count="
+         << column_metadata_->statistics.null_count;
+      throw ParquetException(ss.str());
+    }
   }
 
   bool Equals(const ColumnChunkMetaDataImpl& other) const {

cpp/src/parquet/metadata_test.cc

@@ -17,6 +17,7 @@
 
 #include "parquet/metadata.h"
 
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "arrow/util/key_value_metadata.h"
@@ -427,6 +428,45 @@ TEST(Metadata, TestReadPageIndex) {
   }
 }
 
+// Regression test: a column with max_definition_level == 0 cannot encode
+// nulls, so any positive null_count in its statistics indicates a malformed
+// file. The reader should reject it when the metadata is loaded.
+TEST(Metadata, RejectsRequiredColumnWithNonZeroNullCount) {
+  schema::NodeVector fields;
+  fields.push_back(schema::Int32("required_col", Repetition::REQUIRED));
+  auto schema_node = std::static_pointer_cast<schema::GroupNode>(
+      schema::GroupNode::Make("schema", Repetition::REQUIRED, fields));
+
+  SchemaDescriptor schema_descr;
+  schema_descr.Init(schema_node);
+
+  format::ColumnChunk column_chunk;
+  format::ColumnMetaData& column_metadata = column_chunk.meta_data;
+  column_chunk.__isset.meta_data = true;
+
+  column_metadata.type = format::Type::INT32;
+  column_metadata.codec = format::CompressionCodec::UNCOMPRESSED;
+  column_metadata.num_values = 1000;
+  column_metadata.total_uncompressed_size = 4000;
+  column_metadata.total_compressed_size = 4000;
+  column_metadata.data_page_offset = 4;
+  column_metadata.path_in_schema.push_back("required_col");
+
+  column_metadata.statistics.null_count = 105;
+  column_metadata.statistics.__isset.null_count = true;
+  column_metadata.__isset.statistics = true;
+
+  EXPECT_THROW_THAT(
+      [&]() { ColumnChunkMetaData::Make(&column_chunk, schema_descr.Column(0)); },
+      ParquetException,
+      ::testing::Property(
+          &ParquetException::what,
+          ::testing::AllOf(::testing::HasSubstr("Malformed Parquet file"),
+                           ::testing::HasSubstr("required_col"),
+                           ::testing::HasSubstr("max_definition_level == 0"),
+                           ::testing::HasSubstr("null_count=105"))));
+}
+
 TEST(Metadata, TestSortingColumns) {
   schema::NodeVector fields;
   fields.push_back(schema::Int32("sort_col", Repetition::REQUIRED));