This is an automated email from the ASF dual-hosted git repository.
rok pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new aae29bf6d4e GH-47752: [C++][Python][Parquet] Add minimum compression
savings for V2 pages (#52580)
aae29bf6d4e is described below
commit aae29bf6d4e85bafa7b9bd2bc45d8ab0b0b48373
Author: Mashrafee Aryan <[email protected]>
AuthorDate: Mon Nov 2 08:40:40 2026 -0600
GH-47752: [C++][Python][Parquet] Add minimum compression savings for V2
pages (#52580)
Fixes #47752.
### Rationale for this change
Parquet's V2 pages currently keep compression whenever the compressed
values are smaller than the original values. Even if it saves very little
space, it still does it. For instance, it can compress 100 KB to 99 KB and save
it. Even though that's 1% storage efficient, it might not be worth the
decompression overhead while reading the file.
### What changes are included in this PR?
In the PR, I added an optional `min_compression_savings` setting between
`0.0` and `1.0`. It gives users an option to select the minimum threshold to
compress V2 page data values. For instance, if the original file is `100 KB`
and the user has `min_compression_savings= 0.1`, it means the compressed file
needs to save at least 10% space. As a result, unless the file is `<= 90 KB`,
it won't get saved.
I keep the current behavior when the user does not set `min_space_savings`.
Added C++ and Python implementation.
### Are these changes tested?
- C++ implementation has been tested with unit tests.
- PyArrow implementation has been tested with unti tests.
### Are there any user-facing changes?
C++ and Python users can set `min_compression_savings` when writing V2
pages. For example, `0.1` requires at least 10% space savings before keeping
compression.
### Was AI used for this PR?
ChatGPT helped to set up the build for PyArrow.
Codex helped find a bug in dataset_writing which ignored min_space_savings.
**PR code and description written by:**
- [x] Human (I wrote it. Used ChatGPT to fix my grammar.)
- [ ] AI
**Reviewed before submission by:**
- [x] Human
- [ ] AI
- [ ] Not reviewed
* GitHub Issue: #47752
Lead-authored-by: Mashrfee Aryan <[email protected]>
Co-authored-by: Mashrafee Aryan
<[email protected]>
Co-authored-by: Antoine Pitrou <[email protected]>
Signed-off-by: Rok Mihevc <[email protected]>
---
cpp/src/parquet/column_writer.cc | 8 ++-
cpp/src/parquet/column_writer_test.cc | 67 +++++++++++++++++-
cpp/src/parquet/properties.h | 32 +++++++--
cpp/src/parquet/properties_test.cc | 47 +++++++++++++
docs/source/cpp/parquet.rst | 7 ++
python/pyarrow/_dataset_parquet.pyx | 2 +
python/pyarrow/_parquet.pxd | 3 +-
python/pyarrow/_parquet.pyx | 17 ++++-
python/pyarrow/includes/libparquet.pxd | 1 +
python/pyarrow/parquet/core.py | 11 +++
.../pyarrow/tests/parquet/test_parquet_writer.py | 80 ++++++++++++++++++++++
python/pyarrow/tests/test_dataset.py | 26 +++++++
12 files changed, 288 insertions(+), 13 deletions(-)
diff --git a/cpp/src/parquet/column_writer.cc b/cpp/src/parquet/column_writer.cc
index 23e501d9d67..c899f5a9847 100644
--- a/cpp/src/parquet/column_writer.cc
+++ b/cpp/src/parquet/column_writer.cc
@@ -1059,9 +1059,11 @@ void ColumnWriterImpl::BuildDataPageV2(int64_t
definition_levels_rle_size,
bool page_is_compressed = false;
if (pager_->has_compressor() && values->size() > 0) {
pager_->Compress(*values, compressor_temp_buffer_.get());
- if (compressor_temp_buffer_->size() < values->size()) {
- page_is_compressed = true;
- }
+
+ const int64_t compressed = compressor_temp_buffer_->size();
+ const auto minimum = properties_->min_compression_savings();
+ page_is_compressed =
+ (1.0 - static_cast<double>(compressed) / values->size()) >= minimum;
}
std::shared_ptr<Buffer> compressed_values =
(page_is_compressed ? compressor_temp_buffer_ : values);
diff --git a/cpp/src/parquet/column_writer_test.cc
b/cpp/src/parquet/column_writer_test.cc
index 778b7c9a845..b1f8b4e3be3 100644
--- a/cpp/src/parquet/column_writer_test.cc
+++ b/cpp/src/parquet/column_writer_test.cc
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
+#include <cmath>
#include <memory>
+#include <optional>
#include <utility>
#include <vector>
@@ -27,6 +29,7 @@
#include "arrow/testing/gtest_util.h"
#include "arrow/util/bit_util.h"
#include "arrow/util/bitmap_builders.h"
+#include "arrow/util/compression.h"
#include "arrow/util/config.h"
#include "arrow/util/key_value_metadata.h"
@@ -112,7 +115,8 @@ class TestPrimitiveWriter : public
PrimitiveTypedTest<TestType> {
const ParquetVersion::type version = ParquetVersion::PARQUET_1_0,
const ParquetDataPageVersion data_page_version =
ParquetDataPageVersion::V1,
bool enable_checksum = false, int64_t page_size = kDefaultDataPageSize,
- int64_t max_rows_per_page = kDefaultMaxRowsPerPage) {
+ int64_t max_rows_per_page = kDefaultMaxRowsPerPage,
+ double min_compression_savings = kDefaultMinCompressionSavings) {
sink_ = CreateOutputStream();
WriterProperties::Builder wp_builder;
wp_builder.version(version)->data_page_version(data_page_version);
@@ -130,6 +134,7 @@ class TestPrimitiveWriter : public
PrimitiveTypedTest<TestType> {
wp_builder.max_statistics_size(column_properties.max_statistics_size());
wp_builder.data_pagesize(page_size);
wp_builder.max_rows_per_page(max_rows_per_page);
+ wp_builder.min_compression_savings(min_compression_savings);
writer_properties_ = wp_builder.build();
metadata_ = ColumnChunkMetaDataBuilder::Make(writer_properties_,
this->descr_);
@@ -2118,6 +2123,66 @@ TEST_F(TestValuesWriterInt32Type,
AvoidCompressedInDataPageV2) {
verify_only_one_uncompressed_page(/*total_num_values=*/1);
}
}
+
+// Tests the default, fixed thresholds, and both sides of the actual savings.
+TEST_F(TestValuesWriterInt32Type, MinCompressionSavingsDataPageV2) {
+ // Used zeros because it compresses well
+ this->SetUpSchema(Repetition::OPTIONAL);
+ this->descr_ = this->schema_.Column(0);
+ this->GenerateData(SMALL_SIZE);
+ std::fill(this->values_.begin(), this->values_.end(), 0);
+ std::fill(this->def_levels_.begin(), this->def_levels_.end(), 1);
+
+ ColumnProperties column_properties;
+ column_properties.set_compression(Compression::ZSTD);
+
+ // PLAIN-encoded int32 zeros have the same bytes as the input values.
+ auto codec = GetCodec(Compression::ZSTD);
+ const int64_t values_size = SMALL_SIZE * sizeof(int32_t);
+ const auto* values_bytes = reinterpret_cast<const
uint8_t*>(this->values_ptr_);
+ std::vector<uint8_t> compressed(codec->MaxCompressedLen(values_size,
values_bytes));
+ ASSERT_OK_AND_ASSIGN(
+ auto compressed_size,
+ codec->Compress(values_size, values_bytes, compressed.size(),
compressed.data()));
+ const double savings = 1.0 - static_cast<double>(compressed_size) /
values_size;
+ ASSERT_GT(savings, 0.01);
+ ASSERT_LT(savings, 0.99);
+
+ const std::vector<std::pair<double, bool>> cases = {
+ {kDefaultMinCompressionSavings, true},
+ {0.0, true},
+ {0.1, true},
+ {1.0, false},
+ {savings - 0.01, true},
+ {savings + 0.01, false},
+ };
+
+ for (const auto& [minimum, expected_compressed] : cases) {
+ auto writer =
+ this->BuildWriter(SMALL_SIZE, column_properties,
ParquetVersion::PARQUET_2_LATEST,
+ ParquetDataPageVersion::V2, false,
kDefaultDataPageSize,
+ kDefaultMaxRowsPerPage, minimum);
+
+ writer->WriteBatch(SMALL_SIZE, this->def_levels_.data(), nullptr,
this->values_ptr_);
+ writer->Close();
+
+ ASSERT_OK_AND_ASSIGN(auto buffer, this->sink_->Finish());
+
+ auto page_reader =
+ PageReader::Open(std::make_shared<::arrow::io::BufferReader>(buffer),
SMALL_SIZE,
+ Compression::ZSTD, default_reader_properties(),
*this->descr_);
+
+ auto page = page_reader->NextPage();
+
+ ASSERT_NE(page, nullptr);
+ ASSERT_EQ(PageType::DATA_PAGE_V2, page->type());
+
+ auto data_page = std::static_pointer_cast<DataPageV2>(page);
+
+ ASSERT_EQ(expected_compressed, data_page->is_compressed());
+ }
+}
+
#endif
// Test writing and reading geometry columns
diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h
index 89f2be42bff..25144821389 100644
--- a/cpp/src/parquet/properties.h
+++ b/cpp/src/parquet/properties.h
@@ -174,6 +174,7 @@ ReaderProperties PARQUET_EXPORT default_reader_properties();
static constexpr int64_t kDefaultDataPageSize = 1024 * 1024;
static constexpr int64_t kDefaultMaxRowsPerPage = 20'000;
+static constexpr double kDefaultMinCompressionSavings = 0.01;
static constexpr bool DEFAULT_IS_DICTIONARY_ENABLED = true;
static constexpr int64_t DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT =
kDefaultDataPageSize;
static constexpr int64_t DEFAULT_WRITE_BATCH_SIZE = 1024;
@@ -403,6 +404,7 @@ class PARQUET_EXPORT WriterProperties {
max_rows_per_page_(properties.max_rows_per_page()),
version_(properties.version()),
data_page_version_(properties.data_page_version()),
+ min_compression_savings_(properties.min_compression_savings()),
created_by_(properties.created_by()),
store_decimal_as_integer_(properties.store_decimal_as_integer()),
page_checksum_enabled_(properties.page_checksum_enabled()),
@@ -529,6 +531,16 @@ class PARQUET_EXPORT WriterProperties {
return this;
}
+ /// Set minimum compression ratio threshold for V2 data pages.
+ /// The default is 0.01: keep compression if it saves at least 1%.
+ Builder* min_compression_savings(double min_compression_savings) {
+ if (!(min_compression_savings >= 0.0 && min_compression_savings <= 1.0))
{
+ throw ParquetException("min_compression_savings must be in range [0,
1]");
+ }
+ min_compression_savings_ = min_compression_savings;
+ return this;
+ }
+
/// Specify the Parquet file version.
/// Default PARQUET_2_6.
Builder* version(ParquetVersion::type version) {
@@ -883,6 +895,10 @@ class PARQUET_EXPORT WriterProperties {
/// \brief Build the WriterProperties with the builder parameters.
/// \return The WriterProperties defined by the builder.
std::shared_ptr<WriterProperties> build() {
+ if (min_compression_savings_ > kDefaultMinCompressionSavings &&
+ data_page_version_ == ParquetDataPageVersion::V1) {
+ throw ParquetException("min_compression_savings requires V2 data
pages");
+ }
std::unordered_map<std::string, ColumnProperties> column_properties;
auto get = [&](const std::string& key) -> ColumnProperties& {
auto it = column_properties.find(key);
@@ -918,8 +934,9 @@ class PARQUET_EXPORT WriterProperties {
pagesize_, max_rows_per_page_, version_, created_by_,
page_checksum_enabled_,
size_statistics_level_, std::move(file_encryption_properties_),
default_column_properties_, column_properties, data_page_version_,
- store_decimal_as_integer_, std::move(sorting_columns_),
- content_defined_chunking_enabled_,
content_defined_chunking_options_));
+ min_compression_savings_, store_decimal_as_integer_,
+ std::move(sorting_columns_), content_defined_chunking_enabled_,
+ content_defined_chunking_options_));
}
private:
@@ -933,6 +950,7 @@ class PARQUET_EXPORT WriterProperties {
int64_t max_rows_per_page_;
ParquetVersion::type version_;
ParquetDataPageVersion data_page_version_;
+ double min_compression_savings_ = kDefaultMinCompressionSavings;
std::string created_by_;
bool store_decimal_as_integer_;
bool page_checksum_enabled_;
@@ -973,6 +991,8 @@ class PARQUET_EXPORT WriterProperties {
return parquet_data_page_version_;
}
+ double min_compression_savings() const { return min_compression_savings_; }
+
inline ParquetVersion::type version() const { return parquet_version_; }
inline std::string created_by() const { return parquet_created_by_; }
@@ -1100,9 +1120,9 @@ class PARQUET_EXPORT WriterProperties {
std::shared_ptr<FileEncryptionProperties> file_encryption_properties,
const ColumnProperties& default_column_properties,
const std::unordered_map<std::string, ColumnProperties>&
column_properties,
- ParquetDataPageVersion data_page_version, bool
store_short_decimal_as_integer,
- std::vector<SortingColumn> sorting_columns, bool
content_defined_chunking_enabled,
- CdcOptions content_defined_chunking_options)
+ ParquetDataPageVersion data_page_version, double min_compression_savings,
+ bool store_short_decimal_as_integer, std::vector<SortingColumn>
sorting_columns,
+ bool content_defined_chunking_enabled, CdcOptions
content_defined_chunking_options)
: pool_(pool),
dictionary_pagesize_limit_(dictionary_pagesize_limit),
write_batch_size_(write_batch_size),
@@ -1110,6 +1130,7 @@ class PARQUET_EXPORT WriterProperties {
pagesize_(pagesize),
max_rows_per_page_(max_rows_per_page),
parquet_data_page_version_(data_page_version),
+ min_compression_savings_(min_compression_savings),
parquet_version_(version),
parquet_created_by_(created_by),
store_decimal_as_integer_(store_short_decimal_as_integer),
@@ -1129,6 +1150,7 @@ class PARQUET_EXPORT WriterProperties {
int64_t pagesize_;
int64_t max_rows_per_page_;
ParquetDataPageVersion parquet_data_page_version_;
+ double min_compression_savings_;
ParquetVersion::type parquet_version_;
std::string parquet_created_by_;
bool store_decimal_as_integer_;
diff --git a/cpp/src/parquet/properties_test.cc
b/cpp/src/parquet/properties_test.cc
index 2f22844f8bb..0dfebaf7a4f 100644
--- a/cpp/src/parquet/properties_test.cc
+++ b/cpp/src/parquet/properties_test.cc
@@ -80,6 +80,53 @@ TEST(TestWriterProperties, DefaultCompression) {
::arrow::util::kUseDefaultCompressionLevel);
}
+// Checks the default, stored, and copied min_compression_savings.
+TEST(TestWriterProperties, MinCompressionSavings) {
+ WriterProperties::Builder builder;
+ ASSERT_EQ(0.01, builder.build()->min_compression_savings());
+
+ builder.data_page_version(ParquetDataPageVersion::V2);
+ for (double savings : {0.0, kDefaultMinCompressionSavings, 0.1, 1.0}) {
+ auto properties = builder.min_compression_savings(savings)->build();
+ ASSERT_EQ(savings, properties->min_compression_savings());
+
+ auto copied = WriterProperties::Builder(*properties).build();
+ ASSERT_EQ(properties->min_compression_savings(),
copied->min_compression_savings());
+ }
+
+ auto properties = builder.min_compression_savings(0.0)->build();
+ ASSERT_EQ(0.0, properties->min_compression_savings());
+}
+
+// Tests for invalid settings
+TEST(TestWriterProperties, InvalidMinCompressionSavings) {
+ for (double savings : {-0.1, 1.1, std::numeric_limits<double>::infinity(),
+ -std::numeric_limits<double>::infinity(),
+ std::numeric_limits<double>::quiet_NaN()}) {
+ EXPECT_THROW(WriterProperties::Builder().min_compression_savings(savings),
+ ParquetException);
+ }
+}
+
+TEST(TestWriterProperties, MinCompressionSavingsRequiresDataPageV2) {
+ for (double savings : {2 * kDefaultMinCompressionSavings, 0.1, 1.0}) {
+ WriterProperties::Builder builder;
+ builder.min_compression_savings(savings);
+ 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);
+ EXPECT_NO_THROW(builder.build());
+ }
+}
+
TEST(TestWriterProperties, AdvancedHandling) {
WriterProperties::Builder builder;
builder.compression("gzip", Compression::GZIP);
diff --git a/docs/source/cpp/parquet.rst b/docs/source/cpp/parquet.rst
index 00999f1bd2c..528984878c4 100644
--- a/docs/source/cpp/parquet.rst
+++ b/docs/source/cpp/parquet.rst
@@ -271,6 +271,13 @@ 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.
+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
+at least 10% savings. If the space savings for a page do not meet the
threshold,
+it is not compressed. The default is ``0.01``, so compression must save at
least 1%.
+Setting a value above this default with V1 or using a number outside the
allowed
+range throws ``ParquetException``.
+
Compression is off by default, but to get the most out of Parquet, you should
also choose a compression codec. You can choose one for the whole file or
choose one for individual columns. If you choose a mix, the file-level option
diff --git a/python/pyarrow/_dataset_parquet.pyx
b/python/pyarrow/_dataset_parquet.pyx
index 43eca8957be..b3553708fb6 100644
--- a/python/pyarrow/_dataset_parquet.pyx
+++ b/python/pyarrow/_dataset_parquet.pyx
@@ -664,6 +664,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
store_decimal_as_integer=self._properties["store_decimal_as_integer"],
use_content_defined_chunking=self._properties["use_content_defined_chunking"],
bloom_filter_options=self._properties["bloom_filter_options"],
+
min_compression_savings=self._properties["min_compression_savings"],
)
def _set_arrow_properties(self):
@@ -707,6 +708,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
use_byte_stream_split=False,
column_encoding=None,
data_page_version="1.0",
+ min_compression_savings=0.01,
use_deprecated_int96_timestamps=False,
coerce_timestamps=None,
allow_truncated_timestamps=False,
diff --git a/python/pyarrow/_parquet.pxd b/python/pyarrow/_parquet.pxd
index 36fc2ccf2f3..9cdba15638e 100644
--- a/python/pyarrow/_parquet.pxd
+++ b/python/pyarrow/_parquet.pxd
@@ -57,7 +57,8 @@ cdef shared_ptr[WriterProperties] _create_writer_properties(
sorting_columns=*,
store_decimal_as_integer=*,
use_content_defined_chunking=*,
- bloom_filter_options=*
+ bloom_filter_options=*,
+ min_compression_savings=*
) except *
diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx
index 2621c3bd6d6..40c504ef3b0 100644
--- a/python/pyarrow/_parquet.pyx
+++ b/python/pyarrow/_parquet.pyx
@@ -2076,13 +2076,22 @@ cdef shared_ptr[WriterProperties]
_create_writer_properties(
sorting_columns=None,
store_decimal_as_integer=False,
use_content_defined_chunking=False,
- bloom_filter_options=None) except *:
+ bloom_filter_options=None,
+ min_compression_savings=0.01) except *:
"""General writer properties"""
cdef:
shared_ptr[WriterProperties] properties
WriterProperties.Builder props
CdcOptions cdc_options
+ double savings
+
+ 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"):
+ raise ValueError("min_compression_savings requires V2 data pages")
+ props.min_compression_savings(savings)
# data_page_version
@@ -2405,7 +2414,8 @@ cdef class ParquetWriter(_Weakrefable):
store_decimal_as_integer=False,
use_content_defined_chunking=False,
write_time_adjusted_to_utc=False,
- bloom_filter_options=None):
+ bloom_filter_options=None,
+ min_compression_savings=0.01):
cdef:
shared_ptr[WriterProperties] properties
shared_ptr[ArrowWriterProperties] arrow_properties
@@ -2442,7 +2452,8 @@ cdef class ParquetWriter(_Weakrefable):
sorting_columns=sorting_columns,
store_decimal_as_integer=store_decimal_as_integer,
use_content_defined_chunking=use_content_defined_chunking,
- bloom_filter_options=bloom_filter_options
+ bloom_filter_options=bloom_filter_options,
+ min_compression_savings=min_compression_savings,
)
arrow_properties = _create_arrow_writer_properties(
use_deprecated_int96_timestamps=use_deprecated_int96_timestamps,
diff --git a/python/pyarrow/includes/libparquet.pxd
b/python/pyarrow/includes/libparquet.pxd
index 3a5f91716c1..a7ad239feac 100644
--- a/python/pyarrow/includes/libparquet.pxd
+++ b/python/pyarrow/includes/libparquet.pxd
@@ -485,6 +485,7 @@ cdef extern from "parquet/api/writer.h" namespace "parquet"
nogil:
cdef cppclass WriterProperties:
cppclass Builder:
Builder* data_page_version(ParquetDataPageVersion version)
+ Builder* min_compression_savings(double min_compression_savings)
except +
Builder* version(ParquetVersion version)
Builder* compression(ParquetCompression codec)
Builder* compression(const c_string& path,
diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py
index fa3b897ad23..ab5eafe5171 100644
--- a/python/pyarrow/parquet/core.py
+++ b/python/pyarrow/parquet/core.py
@@ -994,6 +994,13 @@ use_content_defined_chunking : bool or dict, default False
balance between deduplication ratio and fragmentation. Use norm_level=1
or
norm_level=2 to reach a higher deduplication ratio at the expense of
fragmentation.
+min_compression_savings : float, default 0.01
+ Minimum space savings required to keep compression for each V2 data page.
+ 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``.
"""
_parquet_writer_example_doc = """\
@@ -1094,6 +1101,7 @@ Examples
max_rows_per_page=None,
bloom_filter_options=None,
use_content_defined_chunking=False,
+ min_compression_savings=0.01,
**options):
if use_deprecated_int96_timestamps is None:
# Use int96 timestamps for Spark
@@ -1151,6 +1159,7 @@ Examples
max_rows_per_page=max_rows_per_page,
bloom_filter_options=bloom_filter_options,
use_content_defined_chunking=use_content_defined_chunking,
+ min_compression_savings=min_compression_savings,
**options)
self.is_open = True
@@ -2055,6 +2064,7 @@ def write_table(table, where, row_group_size=None,
version='2.6',
max_rows_per_page=None,
bloom_filter_options=None,
use_content_defined_chunking=False,
+ min_compression_savings=0.01,
**kwargs):
# Implementor's note: when adding keywords here / updating defaults, also
# update it in write_to_dataset and _dataset_parquet.pyx
ParquetFileWriteOptions
@@ -2090,6 +2100,7 @@ def write_table(table, where, row_group_size=None,
version='2.6',
max_rows_per_page=max_rows_per_page,
bloom_filter_options=bloom_filter_options,
use_content_defined_chunking=use_content_defined_chunking,
+ min_compression_savings=min_compression_savings,
**kwargs) as writer:
writer.write_table(table, row_group_size=row_group_size)
except Exception:
diff --git a/python/pyarrow/tests/parquet/test_parquet_writer.py
b/python/pyarrow/tests/parquet/test_parquet_writer.py
index a49441f09f4..a411c4e54a7 100644
--- a/python/pyarrow/tests/parquet/test_parquet_writer.py
+++ b/python/pyarrow/tests/parquet/test_parquet_writer.py
@@ -41,6 +41,86 @@ except ImportError:
pytestmark = pytest.mark.parquet
[email protected]("page_version, minimum", [
+ ("1.0", None),
+ ("1.0", 0.0),
+ ("1.0", 0.01),
+ ("2.0", None),
+ ("2.0", 0.0),
+ ("2.0", 0.01),
+ ("2.0", 0.1),
+ ("2.0", 1.0),
+])
[email protected]("compression", ["none", "snappy"])
[email protected]("use_writer", [False, True])
+def test_min_compression_savings(
+ minimum, page_version, compression, use_writer):
+ table = pa.table({
+ "value": pa.array([0.0] * 1000, type=pa.float32())
+ })
+ sink = pa.BufferOutputStream()
+ options = dict(
+ compression=compression,
+ use_dictionary=False,
+ data_page_version=page_version,
+ )
+ if minimum is not None:
+ options["min_compression_savings"] = minimum
+
+ if use_writer:
+ with pq.ParquetWriter(sink, table.schema, **options) as writer:
+ writer.write_table(table)
+ else:
+ pq.write_table(table, sink, **options)
+
+ buf = sink.getvalue()
+ assert pq.read_table(buf).equals(table)
+
+ column = pq.ParquetFile(buf).metadata.row_group(0).column(0)
+ expect_uncompressed = (
+ compression == "none"
+ or (page_version == "2.0" and minimum == 1.0)
+ )
+ if expect_uncompressed:
+ assert column.total_compressed_size == column.total_uncompressed_size
+ else:
+ assert column.total_compressed_size < column.total_uncompressed_size
+
+
+# Tests invalid settings
[email protected](
+ "minimum",
+ [-0.1, 1.1, float("inf"), -float("inf"), float("nan")]
+)
[email protected]("use_writer", [False, True])
+def test_invalid_min_compression_savings(minimum, use_writer):
+ table = pa.table({"value": [1.0]})
+ sink = pa.BufferOutputStream()
+
+ with pytest.raises(ValueError, match="min_compression_savings"):
+ if use_writer:
+ pq.ParquetWriter(
+ sink, table.schema, data_page_version="2.0",
+ min_compression_savings=minimum)
+ else:
+ pq.write_table(
+ table, sink, data_page_version="2.0",
+ min_compression_savings=minimum)
+
+
[email protected]("minimum", [0.02, 0.1, 1.0])
[email protected]("use_writer", [False, True])
+def test_min_compression_savings_rejects_v1(minimum, use_writer):
+ table = pa.table({"value": [1.0]})
+ sink = pa.BufferOutputStream()
+
+ with pytest.raises(ValueError, match="min_compression_savings requires
V2"):
+ if use_writer:
+ pq.ParquetWriter(sink, table.schema,
min_compression_savings=minimum)
+ else:
+ pq.write_table(table, sink, min_compression_savings=minimum)
+
+
@pytest.mark.pandas
def test_parquet_incremental_file_build(tempdir):
df = _test_dataframe(100)
diff --git a/python/pyarrow/tests/test_dataset.py
b/python/pyarrow/tests/test_dataset.py
index 9ddfdfdc9c4..b50441a1ae0 100644
--- a/python/pyarrow/tests/test_dataset.py
+++ b/python/pyarrow/tests/test_dataset.py
@@ -5949,6 +5949,32 @@ def test_write_dataset_content_defined_chunking(tempdir,
cdc):
assert ds.dataset(tempdir, format=fmt).to_table().equals(expected_table)
[email protected]
+def test_parquet_min_compression_savings(tempdir):
+ import pyarrow.parquet as pq
+
+ table = pa.table({
+ "value": pa.array([0.0] * 1000, type=pa.float32())
+ })
+ sizes = []
+
+ for minimum in [0.0, 0.1, 1.0]:
+ directory = tempdir / str(minimum)
+ pq.write_to_dataset(
+ table,
+ directory,
+ compression="snappy",
+ use_dictionary=False,
+ data_page_version="2.0",
+ min_compression_savings=minimum,
+ )
+ result = ds.dataset(directory, format="parquet").to_table()
+ assert result.equals(table)
+ sizes.append(next(directory.iterdir()).stat().st_size)
+
+ assert sizes[-1] > max(sizes[:-1])
+
+
def test_make_write_options_error():
# GH-39440: calling make_write_options as a static class method
msg_1 = ("make_write_options() should be called on an "
