diff --git a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc b/cpp/src/parquet/arrow/arrow_reader_writer_test.cc index 873a41f24e2d..3b29dbce59a5 100644 --- a/cpp/src/parquet/arrow/arrow_reader_writer_test.cc +++ b/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 writer_properties = + WriterProperties::Builder() + .data_page_version(this->GetParquetDataPageVersion()) + ->build(); + std::unique_ptr 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::max())); + ASSERT_OK(writer->Close()); + ASSERT_OK(out_stream->Close()); + + auto buffer_reader = std::make_shared<::arrow::io::BufferReader>(serialized_data); + std::unique_ptr parquet_reader = + ParquetFileReader::Open(std::move(buffer_reader)); + + std::shared_ptr 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 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 writer_properties = + WriterProperties::Builder() + .data_page_version(this->GetParquetDataPageVersion()) + ->build(); + std::unique_ptr 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::max())); + ASSERT_OK(writer->Close()); + ASSERT_OK(out_stream->Close()); + + auto buffer_reader = std::make_shared<::arrow::io::BufferReader>(serialized_data); + std::unique_ptr parquet_reader = + ParquetFileReader::Open(std::move(buffer_reader)); + + std::shared_ptr 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(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 diff --git a/cpp/src/parquet/column_writer.cc b/cpp/src/parquet/column_writer.cc index c899f5a9847a..5eee3c910e1e 100644 --- a/cpp/src/parquet/column_writer.cc +++ b/cpp/src/parquet/column_writer.cc @@ -2024,7 +2024,8 @@ Status TypedColumnWriterImpl::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::WriteArrowDictionary( AddIfNotNull(rep_levels, offset)); std::shared_ptr 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(); diff --git a/cpp/src/parquet/column_writer_test.cc b/cpp/src/parquet/column_writer_test.cc index b1f8b4e3be38..cd553ffe35dc 100644 --- a/cpp/src/parquet/column_writer_test.cc +++ b/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++; } diff --git a/cpp/src/parquet/file_serialize_test.cc b/cpp/src/parquet/file_serialize_test.cc index f287e493a908..e92f509a4473 100644 --- a/cpp/src/parquet/file_serialize_test.cc +++ b/cpp/src/parquet/file_serialize_test.cc @@ -148,8 +148,15 @@ class TestSerialize : public PrimitiveTypedTest { 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; diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h index 251448213896..dc0cc2e3a4e3 100644 --- a/cpp/src/parquet/properties.h +++ b/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; diff --git a/cpp/src/parquet/properties_test.cc b/cpp/src/parquet/properties_test.cc index 0dfebaf7a4f5..003bc825d76a 100644 --- a/cpp/src/parquet/properties_test.cc +++ b/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 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(); diff --git a/docs/source/cpp/parquet.rst b/docs/source/cpp/parquet.rst index 528984878c4a..4f4f4fc994b3 100644 --- a/docs/source/cpp/parquet.rst +++ b/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 diff --git a/docs/source/python/parquet/parquet.rst b/docs/source/python/parquet/parquet.rst index cd1667f74017..7cee5889303b 100644 --- a/docs/source/python/parquet/parquet.rst +++ b/docs/source/python/parquet/parquet.rst @@ -305,7 +305,7 @@ such as the row groups and column chunk metadata and statistics: num_columns: 4 num_rows: 3 - total_byte_size: 290 + total_byte_size: 299 sorting_columns: () >>> metadata.row_group(0).column(0) @@ -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 diff --git a/python/pyarrow/_dataset_parquet.pyx b/python/pyarrow/_dataset_parquet.pyx index b3553708fb6a..3dcc782eaffb 100644 --- a/python/pyarrow/_dataset_parquet.pyx +++ b/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, diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx index afe6ee994507..cdb72f5017a8 100644 --- a/python/pyarrow/_parquet.pyx +++ b/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) diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py index ab5eafe51718..e20b5c2fcd18 100644 --- a/python/pyarrow/parquet/core.py +++ b/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 `_. -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, diff --git a/python/pyarrow/tests/parquet/test_metadata.py b/python/pyarrow/tests/parquet/test_metadata.py index 817566125feb..34a2f4fcd2c4 100644 --- a/python/pyarrow/tests/parquet/test_metadata.py +++ b/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 diff --git a/python/pyarrow/tests/parquet/test_parquet_writer.py b/python/pyarrow/tests/parquet/test_parquet_writer.py index a411c4e54a73..264392bae109 100644 --- a/python/pyarrow/tests/parquet/test_parquet_writer.py +++ b/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