Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions cpp/src/parquet/column_writer.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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_compression_savings();
page_is_compressed =
compressed < values->size() &&
1.0 - static_cast<double>(compressed) / values->size() >= minimum;
Comment on lines +1063 to +1065

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the compressed < values->size() has become superfluous now. If compressed is >= values->size() then the compression savings check will fail.

Suggested change
page_is_compressed =
compressed < values->size() &&
1.0 - static_cast<double>(compressed) / values->size() >= minimum;
page_is_compressed =
(1.0 - static_cast<double>(compressed) / values->size()) >= minimum;

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @pitrou, thank you so much for the suggestion! I was thinking that since we set the default minimum at 0.0, if the compressed size equals the original size, the savings would be 0.0, so the condition 0.0 >= 0.0 would pass.

Would it make sense to keep compressed < values->size() so we avoid decompression with no space savings? Please let me know if I am overlooking something.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we change the default of min_compression_savings to 0.01? or some epsilon-like value? @pitrou

}
std::shared_ptr<Buffer> compressed_values =
(page_is_compressed ? compressor_temp_buffer_ : values);
Expand Down
66 changes: 65 additions & 1 deletion cpp/src/parquet/column_writer_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.

#include <cmath>
#include <memory>
#include <optional>
#include <utility>
#include <vector>

Expand All @@ -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"

Expand Down Expand Up @@ -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 = 0.0) {
sink_ = CreateOutputStream();
WriterProperties::Builder wp_builder;
wp_builder.version(version)->data_page_version(data_page_version);
Expand All @@ -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_);
Expand Down Expand Up @@ -2084,6 +2089,65 @@ 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 = {
{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
Expand Down
31 changes: 26 additions & 5 deletions cpp/src/parquet/properties.h
Original file line number Diff line number Diff line change
Expand Up @@ -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_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()),
Expand Down Expand Up @@ -529,6 +530,16 @@ class PARQUET_EXPORT WriterProperties {
return this;
}

/// Used to set minimum compression ratio threshold for V2 page data values
/// compression. The default is 0.0: keep compression if values become smaller.
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) {
Expand Down Expand Up @@ -883,6 +894,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_ != 0.0 &&
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);
Expand Down Expand Up @@ -918,8 +933,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:
Expand All @@ -933,6 +949,7 @@ class PARQUET_EXPORT WriterProperties {
int64_t max_rows_per_page_;
ParquetVersion::type version_;
ParquetDataPageVersion data_page_version_;
double min_compression_savings_ = 0.0;
std::string created_by_;
bool store_decimal_as_integer_;
bool page_checksum_enabled_;
Expand Down Expand Up @@ -973,6 +990,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_; }
Expand Down Expand Up @@ -1100,16 +1119,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, 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),
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_compression_savings_(min_compression_savings),
parquet_version_(version),
parquet_created_by_(created_by),
store_decimal_as_integer_(store_short_decimal_as_integer),
Expand All @@ -1129,6 +1149,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_;
Expand Down
45 changes: 45 additions & 0 deletions cpp/src/parquet/properties_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,51 @@ TEST(TestWriterProperties, DefaultCompression) {
::arrow::util::kUseDefaultCompressionLevel);
}

// Checks the default, stored, and copied min_compression_savings.
TEST(TestWriterProperties, MinCompressionSavings) {
WriterProperties::Builder builder;
ASSERT_EQ(0.0, builder.build()->min_compression_savings());

builder.data_page_version(ParquetDataPageVersion::V2);
for (double savings : {0.0, 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 : {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());
}
}

TEST(TestWriterProperties, AdvancedHandling) {
WriterProperties::Builder builder;
builder.compression("gzip", Compression::GZIP);
Expand Down
7 changes: 7 additions & 0 deletions docs/source/cpp/parquet.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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. With the default ``0.0``, compression is kept as long as
compressed values are smaller than original values. Setting a nonzero value 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
Expand Down
2 changes: 2 additions & 0 deletions python/pyarrow/_dataset_parquet.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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.0,
use_deprecated_int96_timestamps=False,
coerce_timestamps=None,
allow_truncated_timestamps=False,
Expand Down
3 changes: 2 additions & 1 deletion python/pyarrow/_parquet.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -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 *


Expand Down
17 changes: 14 additions & 3 deletions python/pyarrow/_parquet.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -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.0) 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.0 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

Expand Down Expand Up @@ -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.0):
cdef:
shared_ptr[WriterProperties] properties
shared_ptr[ArrowWriterProperties] arrow_properties
Expand Down Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions python/pyarrow/includes/libparquet.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading
Loading