GH-47752: [C++][Python][Parquet] Add minimum compression savings for V2 pages

cpp/src/parquet/column_writer.cc

@@ -1057,9 +1057,12 @@ 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_space_savings();
+    page_is_compressed =
+        compressed < values->size() &&
+        (!minimum || 1.0 - static_cast<double>(compressed) / values->size() >= *minimum);
   }
   std::shared_ptr<Buffer> compressed_values =
       (page_is_compressed ? compressor_temp_buffer_ : values);

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>
 
@@ -112,7 +114,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,
+      std::optional<double> min_space_savings = {}) {
     sink_ = CreateOutputStream();
     WriterProperties::Builder wp_builder;
     wp_builder.version(version)->data_page_version(data_page_version);
@@ -130,6 +133,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_space_savings(min_space_savings);
     writer_properties_ = wp_builder.build();
 
     metadata_ = ColumnChunkMetaDataBuilder::Make(writer_properties_, this->descr_);
@@ -2084,6 +2088,54 @@ TEST_F(TestValuesWriterInt32Type, AvoidCompressedInDataPageV2) {
     verify_only_one_uncompressed_page(/*total_num_values=*/1);
   }
 }
+
+// Tests if compression works when:
+// - min_space_savings is unset
+// - min_space_savings = 0.1
+// - min_space_savings = 1
+TEST_F(TestValuesWriterInt32Type, MinSpaceSavingsDataPageV2) {
+  // 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);
+
+  ColumnProperties column_properties;
+  column_properties.set_compression(Compression::ZSTD);
+
+  const std::vector<std::pair<std::optional<double>, bool>> cases = {
+      {std::nullopt, true},
+      {0.0, true},
+      {0.1, true},
+      {1.0, 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

cpp/src/parquet/properties.h

@@ -403,6 +403,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_space_savings_(properties.min_space_savings()),
           created_by_(properties.created_by()),
           store_decimal_as_integer_(properties.store_decimal_as_integer()),
           page_checksum_enabled_(properties.page_checksum_enabled()),
@@ -529,6 +530,17 @@ class PARQUET_EXPORT WriterProperties {
       return this;
     }
 
+    /// Used to set minimum threshold for V2 page data values compression
+    /// If min_space_savings unset, keep compression if compressed value is smaller.
+    Builder* min_space_savings(std::optional<double> min_space_savings) {
+      if (min_space_savings &&
+          !(*min_space_savings >= 0.0 && *min_space_savings <= 1.0)) {
+        throw ParquetException("min_space_savings must be in range [0, 1]");
+      }
+      min_space_savings_ = min_space_savings;
+      return this;
+    }
+
     /// Specify the Parquet file version.
     /// Default PARQUET_2_6.
     Builder* version(ParquetVersion::type version) {
@@ -883,6 +895,9 @@ 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_space_savings_ && data_page_version_ == ParquetDataPageVersion::V1) {
+        throw ParquetException("min_space_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,7 +933,7 @@ 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_),
+          min_space_savings_, store_decimal_as_integer_, std::move(sorting_columns_),
           content_defined_chunking_enabled_, content_defined_chunking_options_));
     }
 
@@ -933,6 +948,7 @@ class PARQUET_EXPORT WriterProperties {
     int64_t max_rows_per_page_;
     ParquetVersion::type version_;
     ParquetDataPageVersion data_page_version_;
+    std::optional<double> min_space_savings_ = {};
     std::string created_by_;
     bool store_decimal_as_integer_;
     bool page_checksum_enabled_;
@@ -973,6 +989,8 @@ class PARQUET_EXPORT WriterProperties {
     return parquet_data_page_version_;
   }
 
+  std::optional<double> min_space_savings() const { return min_space_savings_; }
+
   inline ParquetVersion::type version() const { return parquet_version_; }
 
   inline std::string created_by() const { return parquet_created_by_; }
@@ -1100,16 +1118,17 @@ 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, std::optional<double> min_space_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),
         max_row_group_length_(max_row_group_length),
         pagesize_(pagesize),
         max_rows_per_page_(max_rows_per_page),
         parquet_data_page_version_(data_page_version),
+        min_space_savings_(min_space_savings),
         parquet_version_(version),
         parquet_created_by_(created_by),
         store_decimal_as_integer_(store_short_decimal_as_integer),
@@ -1129,6 +1148,7 @@ class PARQUET_EXPORT WriterProperties {
   int64_t pagesize_;
   int64_t max_rows_per_page_;
   ParquetDataPageVersion parquet_data_page_version_;
+  std::optional<double> min_space_savings_;
   ParquetVersion::type parquet_version_;
   std::string parquet_created_by_;
   bool store_decimal_as_integer_;

cpp/src/parquet/properties_test.cc

@@ -80,6 +80,51 @@ TEST(TestWriterProperties, DefaultCompression) {
             ::arrow::util::kUseDefaultCompressionLevel);
 }
 
+// checks if WriterProperties correctly store, copy, and clear min_space_savings
+TEST(TestWriterProperties, MinSpaceSavings) {
+  WriterProperties::Builder builder;
+  ASSERT_FALSE(builder.build()->min_space_savings().has_value());
+
+  builder.data_page_version(ParquetDataPageVersion::V2);
+  for (double savings : {0.0, 0.1, 1.0}) {
+    auto properties = builder.min_space_savings(savings)->build();
+    ASSERT_EQ(std::optional<double>(savings), properties->min_space_savings());
+
+    auto copied = WriterProperties::Builder(*properties).build();
+    ASSERT_EQ(properties->min_space_savings(), copied->min_space_savings());
+  }
+
+  auto properties = builder.min_space_savings(std::nullopt)->build();
+  ASSERT_FALSE(properties->min_space_savings().has_value());
+}
+
+// Tests for invalid settings
+TEST(TestWriterProperties, InvalidMinSpaceSavings) {
+  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_space_savings(savings),
+                 ParquetException);
+  }
+}
+
+TEST(TestWriterProperties, MinSpaceSavingsRequiresDataPageV2) {
+  for (double savings : {0.0, 0.1, 1.0}) {
+    WriterProperties::Builder builder;
+    builder.min_space_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_space_savings(std::nullopt);
+    EXPECT_NO_THROW(builder.build());
+  }
+}
+
 TEST(TestWriterProperties, AdvancedHandling) {
   WriterProperties::Builder builder;
   builder.compression("gzip", Compression::GZIP);

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_space_savings`` to set how much space compression must
+save. Use a value from 0.0 to 1.0. For example, ``min_space_savings(0.1)`` requires
+at least 10% savings. If the space savings do not meet the threshold, it is not
+compressed. If the ``min_space_savings`` option is unset, compression is kept as
+long as compressed values are smaller than original values. Setting a number
+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

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_space_savings=self._properties["min_space_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_space_savings=None,
             use_deprecated_int96_timestamps=False,
             coerce_timestamps=None,
             allow_truncated_timestamps=False,

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_space_savings=*
 ) except *
 
 

python/pyarrow/_parquet.pyx

@@ -2076,13 +2076,23 @@ 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_space_savings=None) except *:
 
     """General writer properties"""
     cdef:
         shared_ptr[WriterProperties] properties
         WriterProperties.Builder props
         CdcOptions cdc_options
+        double savings
+
+    if min_space_savings is not None:
+        savings = min_space_savings
+        if not (0.0 <= savings <= 1.0):
+            raise ValueError("min_space_savings must be in range [0, 1]")
+        if data_page_version in (None, "1.0"):
+            raise ValueError("min_space_savings requires V2 data pages")
+        props.min_space_savings(optional[double](savings))
 
     # data_page_version
 
@@ -2405,7 +2415,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_space_savings=None):
         cdef:
             shared_ptr[WriterProperties] properties
             shared_ptr[ArrowWriterProperties] arrow_properties
@@ -2442,7 +2453,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_space_savings=min_space_savings,
         )
         arrow_properties = _create_arrow_writer_properties(
             use_deprecated_int96_timestamps=use_deprecated_int96_timestamps,

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_space_savings(optional[double] min_space_savings) except +
             Builder* version(ParquetVersion version)
             Builder* compression(ParquetCompression codec)
             Builder* compression(const c_string& path,

python/pyarrow/parquet/core.py

@@ -840,6 +840,14 @@ def _sanitize_table(table, new_schema, flavor):
     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
     Parquet type casting behavior; for that use the "version" option.
+min_space_savings : float or None, default None
+    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. Compression is kept only if it makes the values smaller.
+    If None, no extra minimum savings is required.
+    Setting a number requires ``data_page_version="2.0"``. Using V1 or a number
+    outside the allowed range raises ``ValueError``.
 use_compliant_nested_type : bool, default True
     Whether to write compliant Parquet nested type (lists) as defined
     `here <https://github.com/apache/parquet-format/blob/master/
@@ -1090,6 +1098,7 @@ def __init__(self, where, schema, filesystem=None,
                  max_rows_per_page=None,
                  bloom_filter_options=None,
                  use_content_defined_chunking=False,
+                 min_space_savings=None,
                  **options):
         if use_deprecated_int96_timestamps is None:
             # Use int96 timestamps for Spark
@@ -1147,6 +1156,7 @@ def __init__(self, where, schema, filesystem=None,
             max_rows_per_page=max_rows_per_page,
             bloom_filter_options=bloom_filter_options,
             use_content_defined_chunking=use_content_defined_chunking,
+            min_space_savings=min_space_savings,
             **options)
         self.is_open = True
 
@@ -2039,6 +2049,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_space_savings=None,
                 **kwargs):
     # Implementor's note: when adding keywords here / updating defaults, also
     # update it in write_to_dataset and _dataset_parquet.pyx ParquetFileWriteOptions
@@ -2074,6 +2085,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_space_savings=min_space_savings,
                 **kwargs) as writer:
             writer.write_table(table, row_group_size=row_group_size)
     except Exception: