GH-55272: [C++][Parquet][Python] Default to V2 data pages

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -5225,6 +5225,116 @@ TEST_P(TestArrowWriteDictionary, StatisticsUnifiedDictionary) {
   ASSERT_EQ(rg1_page_reader->NextPage(), nullptr);
 }
 
+TEST_P(TestArrowWriteDictionary, StatisticsWithNullsMatchingDictLength) {
+  // Unique includes nulls, so ensure unreferenced dictionary values
+  // are not mistakenly added to statistics when unique count matches dict length.
+  auto dict_type = ::arrow::dictionary(::arrow::int32(), ::arrow::utf8());
+  auto schema = ::arrow::schema({::arrow::field("values", dict_type)});
+
+  auto dictionary = ArrayFromJSON(::arrow::utf8(), R"(["a", "b"])");
+  auto indices = ArrayFromJSON(::arrow::int32(), R"([0, null, 0])");
+  ASSERT_OK_AND_ASSIGN(auto dict_array, ::arrow::DictionaryArray::FromArrays(
+                                            dict_type, indices, dictionary));
+  auto table = ::arrow::Table::Make(schema, {dict_array});
+
+  std::shared_ptr<::arrow::ResizableBuffer> serialized_data = AllocateBuffer();
+  auto out_stream = std::make_shared<::arrow::io::BufferOutputStream>(serialized_data);
+
+  std::shared_ptr<WriterProperties> writer_properties =
+      WriterProperties::Builder()
+          .data_page_version(this->GetParquetDataPageVersion())
+          ->build();
+  std::unique_ptr<FileWriter> writer;
+  ASSERT_OK_AND_ASSIGN(
+      writer, FileWriter::Open(*schema, ::arrow::default_memory_pool(), out_stream,
+                               writer_properties, default_arrow_writer_properties()));
+  ASSERT_OK(writer->WriteTable(*table, std::numeric_limits<int64_t>::max()));
+  ASSERT_OK(writer->Close());
+  ASSERT_OK(out_stream->Close());
+
+  auto buffer_reader = std::make_shared<::arrow::io::BufferReader>(serialized_data);
+  std::unique_ptr<ParquetFileReader> parquet_reader =
+      ParquetFileReader::Open(std::move(buffer_reader));
+
+  std::shared_ptr<FileMetaData> metadata = parquet_reader->metadata();
+  ASSERT_EQ(metadata->num_row_groups(), 1);
+  auto stats = metadata->RowGroup(0)->ColumnChunk(0)->statistics();
+  ASSERT_NE(stats, nullptr);
+  EXPECT_EQ(stats->num_values(), 2);
+  EXPECT_EQ(stats->null_count(), 1);
+  EXPECT_EQ(stats->EncodeMin(), "a");
+  // "b" was never referenced in the data, so EncodeMax must be "a", not "b"
+  EXPECT_EQ(stats->EncodeMax(), "a");
+}
+
+TEST_P(TestArrowWriteDictionary, NestedDictionaryParentNullStatistics) {
+  // Ensure statistics accurately reflect parent nulls for nested dictionary columns.
+  auto dict_type = ::arrow::dictionary(::arrow::int32(), ::arrow::utf8());
+  auto inner_field = ::arrow::field("inner", dict_type, /*nullable=*/false);
+  auto struct_type = ::arrow::struct_({inner_field});
+  auto schema = ::arrow::schema({::arrow::field("root", struct_type, /*nullable=*/true)});
+
+  auto dictionary = ArrayFromJSON(::arrow::utf8(), R"(["a", "b"])");
+  auto indices = ArrayFromJSON(::arrow::int32(), R"([0, 0, 1])");
+  ASSERT_OK_AND_ASSIGN(auto inner_dict, ::arrow::DictionaryArray::FromArrays(
+                                            dict_type, indices, dictionary));
+
+  // Parent struct validity bitmap: [valid, null, valid]
+  ::arrow::TypedBufferBuilder<bool> bitmap_builder;
+  ASSERT_OK(bitmap_builder.Append(true));
+  ASSERT_OK(bitmap_builder.Append(false));
+  ASSERT_OK(bitmap_builder.Append(true));
+  ASSERT_OK_AND_ASSIGN(auto null_bitmap, bitmap_builder.Finish());
+
+  ASSERT_OK_AND_ASSIGN(auto struct_array, ::arrow::StructArray::Make(
+                                              {inner_dict}, {inner_field}, null_bitmap));
+  auto table = ::arrow::Table::Make(schema, {struct_array});
+
+  std::shared_ptr<::arrow::ResizableBuffer> serialized_data = AllocateBuffer();
+  auto out_stream = std::make_shared<::arrow::io::BufferOutputStream>(serialized_data);
+
+  std::shared_ptr<WriterProperties> writer_properties =
+      WriterProperties::Builder()
+          .data_page_version(this->GetParquetDataPageVersion())
+          ->build();
+  std::unique_ptr<FileWriter> writer;
+  ASSERT_OK_AND_ASSIGN(
+      writer, FileWriter::Open(*schema, ::arrow::default_memory_pool(), out_stream,
+                               writer_properties, default_arrow_writer_properties()));
+  ASSERT_OK(writer->WriteTable(*table, std::numeric_limits<int64_t>::max()));
+  ASSERT_OK(writer->Close());
+  ASSERT_OK(out_stream->Close());
+
+  auto buffer_reader = std::make_shared<::arrow::io::BufferReader>(serialized_data);
+  std::unique_ptr<ParquetFileReader> parquet_reader =
+      ParquetFileReader::Open(std::move(buffer_reader));
+
+  std::shared_ptr<FileMetaData> metadata = parquet_reader->metadata();
+  ASSERT_EQ(metadata->num_row_groups(), 1);
+  auto stats = metadata->RowGroup(0)->ColumnChunk(0)->statistics();
+  ASSERT_NE(stats, nullptr);
+  // There are 2 valid inner values and 1 null (from the parent struct)
+  EXPECT_EQ(stats->num_values(), 2);
+  EXPECT_EQ(stats->null_count(), 1);
+  EXPECT_EQ(stats->EncodeMin(), "a");
+  EXPECT_EQ(stats->EncodeMax(), "b");
+
+  // In V2, verify that the data page header num_nulls matches page statistics
+  if (this->GetParquetDataPageVersion() == ParquetDataPageVersion::V2) {
+    auto rg_reader = parquet_reader->RowGroup(0)->GetColumnPageReader(0);
+    auto dict_page = rg_reader->NextPage();
+    ASSERT_NE(dict_page, nullptr);
+    ASSERT_EQ(dict_page->type(), PageType::DICTIONARY_PAGE);
+    auto page = rg_reader->NextPage();
+    ASSERT_NE(page, nullptr);
+    ASSERT_EQ(page->type(), PageType::DATA_PAGE_V2);
+    auto data_page_v2 = std::static_pointer_cast<DataPageV2>(page);
+    EXPECT_EQ(data_page_v2->num_nulls(), 1);
+    EXPECT_EQ(data_page_v2->num_rows(), 3);
+    EXPECT_EQ(data_page_v2->num_values(), 3);
+  }
+}
+
 // ----------------------------------------------------------------------
 // Tests for directly reading DictionaryArray
 

cpp/src/parquet/column_writer.cc

@@ -2024,7 +2024,8 @@ Status TypedColumnWriterImpl<ParquetType>::WriteArrowDictionary(
                             ::arrow::compute::Unique(*chunk_indices, &exec_ctx));
 
     // On first run, we might be able to re-use the existing dictionary
-    if (referenced_indices.length() == dictionary->length()) {
+    if (chunk_indices->null_count() == 0 &&
+        referenced_indices.length() == dictionary->length()) {
       referenced_dictionary = dictionary;
     } else {
       PARQUET_ASSIGN_OR_THROW(
@@ -2064,12 +2065,12 @@ Status TypedColumnWriterImpl<ParquetType>::WriteArrowDictionary(
                       AddIfNotNull(rep_levels, offset));
     std::shared_ptr<Array> writeable_indices =
         indices->Slice(value_offset, batch_num_spaced_values);
-    if (page_statistics_ || bloom_filter_writer_) {
-      update_stats(/*num_chunk_levels=*/batch_size, writeable_indices);
-    }
     PARQUET_ASSIGN_OR_THROW(
         writeable_indices,
         MaybeReplaceValidity(writeable_indices, null_count, ctx->memory_pool));
+    if (page_statistics_ || bloom_filter_writer_) {
+      update_stats(/*num_chunk_levels=*/batch_size, writeable_indices);
+    }
     dict_encoder->PutIndices(*writeable_indices);
     // Update unencoded byte array data size to size statistics
     UpdateUnencodedDataBytes();

cpp/src/parquet/column_writer_test.cc

@@ -1134,7 +1134,7 @@ TEST(TestColumnWriter, LARGE_MEMORY_TEST(WriteLargeDictEncodedPage)) {
     if (page_count == 0) {
       ASSERT_EQ(page->type(), PageType::DICTIONARY_PAGE);
     } else {
-      ASSERT_EQ(page->type(), PageType::DATA_PAGE);
+      ASSERT_EQ(page->type(), PageType::DATA_PAGE_V2);
     }
     page_count++;
   }

cpp/src/parquet/file_serialize_test.cc

@@ -148,8 +148,15 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
       const int64_t total_compressed_size = rg_metadata->total_compressed_size();
       if (expected_codec_type == Compression::UNCOMPRESSED) {
         ASSERT_EQ(total_byte_size, total_compressed_size);
-      } else {
+      } else if (writer_properties->data_page_version() == ParquetDataPageVersion::V1) {
         ASSERT_NE(total_byte_size, total_compressed_size);
+      } else {
+        // In Parquet V2, selective compression may keep small or incompressible data
+        // pages uncompressed (total_compressed_size == total_byte_size), while
+        // compressed dictionary pages or pages with savings may be larger or smaller
+        // than uncompressed sizes.
+        ASSERT_GT(total_byte_size, 0);
+        ASSERT_GT(total_compressed_size, 0);
       }
 
       int64_t total_column_byte_size = 0;

cpp/src/parquet/properties.h

@@ -387,7 +387,7 @@ class PARQUET_EXPORT WriterProperties {
           pagesize_(kDefaultDataPageSize),
           max_rows_per_page_(kDefaultMaxRowsPerPage),
           version_(ParquetVersion::PARQUET_2_6),
-          data_page_version_(ParquetDataPageVersion::V1),
+          data_page_version_(ParquetDataPageVersion::V2),
           created_by_(DEFAULT_CREATED_BY),
           store_decimal_as_integer_(false),
           page_checksum_enabled_(false),
@@ -525,7 +525,7 @@ class PARQUET_EXPORT WriterProperties {
     }
 
     /// Specify the data page version.
-    /// Default V1.
+    /// Default V2.
     Builder* data_page_version(ParquetDataPageVersion data_page_version) {
       data_page_version_ = data_page_version;
       return this;

cpp/src/parquet/properties_test.cc

@@ -67,7 +67,7 @@ TEST(TestWriterProperties, Basics) {
   ASSERT_EQ(kDefaultDataPageSize, props->data_pagesize());
   ASSERT_EQ(DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT, props->dictionary_pagesize_limit());
   ASSERT_EQ(ParquetVersion::PARQUET_2_6, props->version());
-  ASSERT_EQ(ParquetDataPageVersion::V1, props->data_page_version());
+  ASSERT_EQ(ParquetDataPageVersion::V2, props->data_page_version());
   ASSERT_FALSE(props->page_checksum_enabled());
 }
 
@@ -112,14 +112,15 @@ TEST(TestWriterProperties, MinCompressionSavingsRequiresDataPageV2) {
   for (double savings : {2 * kDefaultMinCompressionSavings, 0.1, 1.0}) {
     WriterProperties::Builder builder;
     builder.min_compression_savings(savings);
+    EXPECT_NO_THROW(builder.build());
+
+    builder.data_page_version(ParquetDataPageVersion::V1);
     EXPECT_THROW(builder.build(), ParquetException);
 
     builder.data_page_version(ParquetDataPageVersion::V2);
     EXPECT_NO_THROW(builder.build());
 
     builder.data_page_version(ParquetDataPageVersion::V1);
-    EXPECT_THROW(builder.build(), ParquetException);
-
     builder.min_compression_savings(0.0);
     EXPECT_NO_THROW(builder.build());
     builder.min_compression_savings(kDefaultMinCompressionSavings);
@@ -418,7 +419,7 @@ std::vector<WriterPropertiesTestCase> writer_properties_test_cases() {
     builder.max_row_group_length(DEFAULT_MAX_ROW_GROUP_LENGTH - 1);
     builder.data_pagesize(kDefaultDataPageSize - 1);
     builder.max_rows_per_page(kDefaultMaxRowsPerPage - 1);
-    builder.data_page_version(ParquetDataPageVersion::V2);
+    builder.data_page_version(ParquetDataPageVersion::V1);
     builder.version(ParquetVersion::type::PARQUET_2_4);
     builder.enable_store_decimal_as_integer();
     builder.disable_write_page_index();

docs/source/cpp/parquet.rst

@@ -267,9 +267,10 @@ group that takes precedent over the ``chunk_size`` passed in the write methods.
 
 You can set the version of Parquet to write with ``version``, which determines
 which logical types are available. In addition, you can set the data page version
-with ``data_page_version``. It's V1 by default; setting to V2 will allow more
-optimal compression (skipping compressing pages where there isn't a space
-benefit), but not all readers support this data page version.
+with ``data_page_version``. It defaults to V2, which allows more optimal
+compression (skipping compressing pages where there isn't a space benefit) and
+page-level null count metadata. If compatibility with older readers is needed,
+it can be set to V1.
 
 For V2 pages, use ``min_compression_savings`` to set how much space compression must
 save. Use a value from 0.0 to 1.0. For example, ``min_compression_savings(0.1)`` requires

docs/source/python/parquet/parquet.rst

@@ -305,7 +305,7 @@ such as the row groups and column chunk metadata and statistics:
    <pyarrow._parquet.RowGroupMetaData object at ...>
      num_columns: 4
      num_rows: 3
-     total_byte_size: 290
+     total_byte_size: 299
      sorting_columns: ()
    >>> metadata.row_group(0).column(0)
    <pyarrow._parquet.ColumnChunkMetaData object at ...>
@@ -333,8 +333,8 @@ such as the row groups and column chunk metadata and statistics:
      has_dictionary_page: True
      dictionary_page_offset: 4
      data_page_offset: 36
-     total_compressed_size: 106
-     total_uncompressed_size: 102
+     total_compressed_size: 105
+     total_uncompressed_size: 103
      bloom_filter_offset: None
      bloom_filter_length: None
 

python/pyarrow/_dataset_parquet.pyx

@@ -707,7 +707,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
             compression_level=None,
             use_byte_stream_split=False,
             column_encoding=None,
-            data_page_version="1.0",
+            data_page_version="2.0",
             min_compression_savings=0.01,
             use_deprecated_int96_timestamps=False,
             coerce_timestamps=None,

python/pyarrow/_parquet.pyx

@@ -2089,7 +2089,7 @@ cdef shared_ptr[WriterProperties] _create_writer_properties(
     savings = min_compression_savings
     if not (0.0 <= savings <= 1.0):
         raise ValueError("min_compression_savings must be in range [0, 1]")
-    if savings > 0.01 and data_page_version in (None, "1.0"):
+    if savings > 0.01 and data_page_version == "1.0":
         raise ValueError("min_compression_savings requires V2 data pages")
     props.min_compression_savings(savings)
 

python/pyarrow/parquet/core.py

@@ -840,9 +840,9 @@ def _sanitize_table(table, new_schema, flavor):
     Certain encodings are only compatible with certain data types.
     Please refer to the encodings section of `Reading and writing Parquet
     files <https://arrow.apache.org/docs/cpp/parquet.html#encodings>`_.
-data_page_version : {"1.0", "2.0"}, default "1.0"
+data_page_version : {"1.0", "2.0"}, default "2.0"
     The serialized Parquet data page format version to write, defaults to
-    1.0. This does not impact the file schema logical types and Arrow to
+    2.0. This does not impact the file schema logical types and Arrow to
     Parquet type casting behavior; for that use the "version" option.
 use_compliant_nested_type : bool, default True
     Whether to write compliant Parquet nested type (lists) as defined
@@ -999,8 +999,8 @@ def _sanitize_table(table, new_schema, flavor):
     Use a value from 0.0 to 1.0. For example, 0.1 requires at least 10% savings.
     Savings are measured on encoded values, excluding repetition and definition
     levels. The default is 0.01, so compression must save at least 1%.
-    Values above this default require ``data_page_version="2.0"``. Using
-    a larger value with V1 or a number outside the allowed range raises ``ValueError``.
+    Using a larger value with V1 or a number outside the allowed range
+    raises ``ValueError``.
 """
 
 _parquet_writer_example_doc = """\
@@ -1087,7 +1087,7 @@ def __init__(self, where, schema, filesystem=None,
                  use_byte_stream_split=False,
                  column_encoding=None,
                  writer_engine_version=None,
-                 data_page_version='1.0',
+                 data_page_version='2.0',
                  use_compliant_nested_type=True,
                  encryption_properties=None,
                  write_batch_size=None,
@@ -2050,7 +2050,7 @@ def write_table(table, where, row_group_size=None, version='2.6',
                 compression_level=None,
                 use_byte_stream_split=False,
                 column_encoding=None,
-                data_page_version='1.0',
+                data_page_version='2.0',
                 use_compliant_nested_type=True,
                 encryption_properties=None,
                 write_batch_size=None,

python/pyarrow/tests/parquet/test_metadata.py

@@ -133,7 +133,7 @@ def test_parquet_metadata_api():
     assert col_meta.is_stats_set is True
     assert isinstance(col_meta.statistics, pq.Statistics)
     assert col_meta.compression == 'SNAPPY'
-    assert set(col_meta.encodings) == {'PLAIN', 'RLE'}
+    assert set(col_meta.encodings) == {'RLE'}
     assert col_meta.has_dictionary_page is False
     assert col_meta.dictionary_page_offset is None
     assert col_meta.data_page_offset > 0

python/pyarrow/tests/parquet/test_parquet_writer.py

@@ -116,9 +116,24 @@ def test_min_compression_savings_rejects_v1(minimum, use_writer):
 
     with pytest.raises(ValueError, match="min_compression_savings requires V2"):
         if use_writer:
-            pq.ParquetWriter(sink, table.schema, min_compression_savings=minimum)
+            pq.ParquetWriter(sink, table.schema, data_page_version="1.0",
+                             min_compression_savings=minimum)
         else:
-            pq.write_table(table, sink, min_compression_savings=minimum)
+            pq.write_table(table, sink, data_page_version="1.0",
+                           min_compression_savings=minimum)
+
+
+@pytest.mark.parametrize("use_writer", [False, True])
+def test_min_compression_savings_default_v2(use_writer):
+    table = pa.table({"value": [1.0]})
+    sink = pa.BufferOutputStream()
+
+    if use_writer:
+        with pq.ParquetWriter(
+                sink, table.schema, min_compression_savings=0.1) as writer:
+            writer.write_table(table)
+    else:
+        pq.write_table(table, sink, min_compression_savings=0.1)
 
 
 @pytest.mark.pandas