This is an automated email from the ASF dual-hosted git repository.
Jefffrey pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 5fa7c8cc10 perf(parquet): resolve per-column writer properties once
per column (#10880)
5fa7c8cc10 is described below
commit 5fa7c8cc10cd821b77e0ea9d9444352b28942c8a
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Mon Aug 31 01:58:17 2026 -0500
perf(parquet): resolve per-column writer properties once per column (#10880)
# Which issue does this PR close?
- Contributes to #9722.
- Builds on #10878, which adds the `repeated_batches` benchmark cases
used below.
# Rationale for this change
`WriterProperties` keeps per-column overrides in a
`HashMap<ColumnPath, ColumnProperties>`. Every accessor that takes a
`&ColumnPath` — `compression`, `encoding`, `dictionary_enabled`,
`statistics_enabled`, `write_page_header_statistics`,
`column_data_page_size_limit`, `column_dictionary_page_size_limit`,
`column_data_page_v2_compression_ratio_threshold` and
`bloom_filter_properties` — hashes that path and searches the map.
`ColumnPath`
is a `Vec<String>`, so each search hashes every part of the path and
then
compares strings on a hit.
The column writers call those accessors repeatedly for the same leaf
column:
- 6 times while a column writer is constructed
(`GenericColumnWriter::new` plus
`ColumnValueEncoder::try_new`);
- 2 more on every `write`, in `ByteBudgetChunker::new`;
- 1 per mini-batch in `should_add_data_page`, and another in
`should_dict_fallback` when a dictionary encoder is active;
- 1 per page in `add_data_page`, plus one more for Data Page v2.
`WriterProperties` is immutable once built, so all of that recomputes an
answer
that was already available when the column writer was created.
Counting the searches directly (temporary instrumentation on the map
access) over
the `writer_overhead` shapes:
| benchmark case | searches before | searches after |
| --- | --- | --- |
| `writer_overhead/1000_cols` | 10,000 | 1,000 |
| `writer_overhead/5000_cols` | 50,000 | 5,000 |
| `writer_overhead/10000_cols` | 100,000 | 10,000 |
| `writer_overhead/1000_cols/repeated_batches` | 103,000 | 1,000 |
| `writer_overhead/5000_cols/repeated_batches` | 515,000 | 5,000 |
After the change it is exactly one search per leaf column per row group.
# What changes are included in this PR?
- Adds `WriterProperties::resolve_column_properties`, which searches the
per-column map once and returns a `ResolvedColumnProperties` holding
every
per-column setting already resolved against the file-wide defaults. The
struct
is `pub(crate)`; no public API is added.
- `GenericColumnWriter::new` calls it once and stores the result. The
per-batch
and per-page paths, `ByteBudgetChunker::new`, and both
`ColumnValueEncoder`
implementations read from it instead of searching the map again.
`ColumnValueEncoder::try_new` and `create_bloom_filter` now take the
resolved
settings; that trait is not nameable outside the crate.
- The existing per-column accessors keep their behaviour and are
reimplemented on
top of the same resolution helpers, so the "column override, else file
default,
else constant" rule has one definition rather than one per accessor.
# Are these changes tested?
Yes.
- The existing suite passes unchanged, with and without `encryption`:
`cargo test -p parquet --features "arrow async encryption test_common
experimental"`.
- New `test_resolve_column_properties_matches_individual_accessors`
asserts the
one-pass resolution agrees with every individual accessor, for a column
with
overrides, a column that inherits the file defaults, and properties left
entirely at their defaults.
- To check the writer is byte-for-byte unchanged I wrote the same file
from a
build before and after this change, with a workload covering the
settings this
PR touches: per-column compression, encoding, dictionary on and off,
page and
dictionary page size limits, page-header statistics, a bloom filter,
dictionary
fallback on multi-KB values, and both writer versions. The two files are
identical.
Benchmark:
cargo bench -p parquet --bench writer_overhead
I did not have an idle machine, and criterion's wall-clock intervals
were too
wide there to be usable. I measured CPU time (user + sys) per iteration
instead,
which is far less sensitive to competing load: the minimum over 40
alternating
runs of the two builds, with a separately measured ~13.3 ms of fixed
benchmark
setup subtracted.
| benchmark case | before | after | |
| --- | --- | --- | --- |
| `writer_overhead/1000_cols` | 2.7 ms | 2.1 ms | see note |
| `writer_overhead/5000_cols` | 20.2 ms | 18.5 ms | −8 % |
| `writer_overhead/10000_cols` | 38.8 ms | 34.2 ms | −12 % |
| `writer_overhead/1000_cols/repeated_batches` | 12.9 ms | 8.6 ms | −34
% |
| `writer_overhead/5000_cols/repeated_batches` | 88.0 ms | 70.1 ms | −20
% |
Note: at 1,000 columns with a single batch the remaining figure is close
enough
to the spread of the setup subtraction that I would not read a
percentage into
it. The other four imply about 40 ns per map search, consistently across
all four
shapes, which matches the search counts above. A run on an idle machine
would be
worth having, and the benchmark from #10878 makes that easy.
`ColumnWriterImpl<Int32Type>` grows from 1312 to 1376 bytes to hold the
resolved
settings. Peak RSS on the 10,000 column benchmark was unchanged.
# Are there any user-facing changes?
No. No public API is added, removed or changed, and the bytes written
are
unchanged.
# AI usage
This PR was written with Claude Code and reviewed by a human. The
byte-for-byte
output comparison and the map-search counts above are the checks run to
confirm
the behaviour is unchanged.
---------
Co-authored-by: Claude <[email protected]>
---
parquet/src/arrow/arrow_writer/byte_array.rs | 38 +--
parquet/src/column/writer/byte_budget_chunker.rs | 8 +-
parquet/src/column/writer/encoder.rs | 40 ++-
parquet/src/column/writer/mod.rs | 39 ++-
parquet/src/file/properties.rs | 302 +++++++++++++++++++----
5 files changed, 326 insertions(+), 101 deletions(-)
diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs
b/parquet/src/arrow/arrow_writer/byte_array.rs
index 85f37d6da6..e6f843f959 100644
--- a/parquet/src/arrow/arrow_writer/byte_array.rs
+++ b/parquet/src/arrow/arrow_writer/byte_array.rs
@@ -24,7 +24,9 @@ use crate::data_type::{AsBytes, ByteArray, Int32Type};
use crate::encodings::encoding::{DeltaBitPackEncoder, Encoder};
use crate::encodings::rle::RleEncoder;
use crate::errors::{ParquetError, Result};
-use crate::file::properties::{EnabledStatistics, WriterProperties,
WriterVersion};
+use crate::file::properties::{
+ EnabledStatistics, ResolvedColumnProperties, WriterProperties,
WriterVersion,
+};
use crate::geospatial::accumulator::{GeoStatsAccumulator,
try_new_geo_stats_accumulator};
use crate::geospatial::statistics::GeospatialStatistics;
use crate::schema::types::ColumnDescPtr;
@@ -135,16 +137,16 @@ enum FallbackEncoderImpl {
}
impl FallbackEncoder {
- /// Create the fallback encoder for the given [`ColumnDescPtr`] and
[`WriterProperties`]
- fn new(descr: &ColumnDescPtr, props: &WriterProperties) -> Result<Self> {
+ /// Create the fallback encoder for the given [`WriterProperties`] and the
+ /// column settings already resolved from them
+ fn new(props: &WriterProperties, column_props: &ResolvedColumnProperties)
-> Result<Self> {
// Set either main encoder or fallback encoder.
- let encoding =
- props
- .encoding(descr.path())
- .unwrap_or_else(|| match props.writer_version() {
- WriterVersion::PARQUET_1_0 => Encoding::PLAIN,
- WriterVersion::PARQUET_2_0 => Encoding::DELTA_BYTE_ARRAY,
- });
+ let encoding = column_props
+ .encoding
+ .unwrap_or_else(|| match props.writer_version() {
+ WriterVersion::PARQUET_1_0 => Encoding::PLAIN,
+ WriterVersion::PARQUET_2_0 => Encoding::DELTA_BYTE_ARRAY,
+ });
let encoder = match encoding {
Encoding::PLAIN => FallbackEncoderImpl::Plain { buffer: vec![] },
@@ -441,19 +443,21 @@ impl ColumnValueEncoder for ByteArrayEncoder {
Some(sbbf)
}
- fn try_new(descr: &ColumnDescPtr, props: &WriterProperties) -> Result<Self>
+ fn try_new(
+ descr: &ColumnDescPtr,
+ props: &WriterProperties,
+ column_props: &ResolvedColumnProperties,
+ ) -> Result<Self>
where
Self: Sized,
{
- let dictionary = props
- .dictionary_enabled(descr.path())
- .then(DictEncoder::default);
+ let dictionary =
column_props.dictionary_enabled.then(DictEncoder::default);
- let fallback = FallbackEncoder::new(descr, props)?;
+ let fallback = FallbackEncoder::new(props, column_props)?;
- let (bloom_filter, bloom_filter_target_fpp) =
create_bloom_filter(props, descr)?;
+ let (bloom_filter, bloom_filter_target_fpp) =
create_bloom_filter(column_props)?;
- let statistics_enabled = props.statistics_enabled(descr.path());
+ let statistics_enabled = column_props.statistics_enabled;
let geo_stats_accumulator = try_new_geo_stats_accumulator(descr);
diff --git a/parquet/src/column/writer/byte_budget_chunker.rs
b/parquet/src/column/writer/byte_budget_chunker.rs
index 56c7b4c6f8..0d8eeb85ab 100644
--- a/parquet/src/column/writer/byte_budget_chunker.rs
+++ b/parquet/src/column/writer/byte_budget_chunker.rs
@@ -20,7 +20,7 @@
use crate::basic::Type;
use crate::column::writer::LevelDataRef;
use crate::column::writer::encoder::ColumnValueEncoder;
-use crate::file::properties::WriterProperties;
+use crate::file::properties::ResolvedColumnProperties;
use crate::schema::types::ColumnDescriptor;
/// Picks byte-budget-aware mini-batch sizes for one column.
@@ -61,11 +61,11 @@ impl ByteBudgetChunker {
#[inline]
pub(crate) fn new(
descr: &ColumnDescriptor,
- props: &WriterProperties,
+ column_props: &ResolvedColumnProperties,
base_batch_size: usize,
) -> Self {
- let page_byte_limit = props.column_data_page_size_limit(descr.path());
- let dict_page_byte_limit =
props.column_dictionary_page_size_limit(descr.path());
+ let page_byte_limit = column_props.data_page_size_limit;
+ let dict_page_byte_limit = column_props.dictionary_page_size_limit;
let static_bytes_per_value = match descr.physical_type() {
Type::BOOLEAN => Some(1),
Type::INT32 | Type::FLOAT => Some(std::mem::size_of::<i32>()),
diff --git a/parquet/src/column/writer/encoder.rs
b/parquet/src/column/writer/encoder.rs
index fe644d72e0..467c1510b5 100644
--- a/parquet/src/column/writer/encoder.rs
+++ b/parquet/src/column/writer/encoder.rs
@@ -26,7 +26,7 @@ use crate::data_type::DataType;
use crate::data_type::private::ParquetValueType;
use crate::encodings::encoding::{DictEncoder, Encoder, get_encoder};
use crate::errors::{ParquetError, Result};
-use crate::file::properties::{EnabledStatistics, WriterProperties};
+use crate::file::properties::{EnabledStatistics, ResolvedColumnProperties,
WriterProperties};
use crate::geospatial::accumulator::{GeoStatsAccumulator,
try_new_geo_stats_accumulator};
use crate::geospatial::statistics::GeospatialStatistics;
use crate::schema::types::{BasicTypeInfo, ColumnDescPtr};
@@ -80,7 +80,18 @@ pub trait ColumnValueEncoder {
type Values: ColumnValues + ?Sized;
/// Create a new [`ColumnValueEncoder`]
- fn try_new(descr: &ColumnDescPtr, props: &WriterProperties) -> Result<Self>
+ ///
+ /// `column_props` holds the settings in `props` that apply specifically to
+ /// `descr`, already resolved by the caller.
+ #[expect(
+ private_interfaces,
+ reason = "this trait is not nameable outside the crate"
+ )]
+ fn try_new(
+ descr: &ColumnDescPtr,
+ props: &WriterProperties,
+ column_props: &ResolvedColumnProperties,
+ ) -> Result<Self>
where
Self: Sized;
@@ -254,22 +265,30 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
Some(sbbf)
}
- fn try_new(descr: &ColumnDescPtr, props: &WriterProperties) ->
Result<Self> {
- let dict_supported = props.dictionary_enabled(descr.path())
+ #[expect(
+ private_interfaces,
+ reason = "this trait is not nameable outside the crate"
+ )]
+ fn try_new(
+ descr: &ColumnDescPtr,
+ props: &WriterProperties,
+ column_props: &ResolvedColumnProperties,
+ ) -> Result<Self> {
+ let dict_supported = column_props.dictionary_enabled
&& has_dictionary_support(T::get_physical_type(), props);
let dict_encoder = dict_supported.then(||
DictEncoder::new(descr.clone()));
// Set either main encoder or fallback encoder.
let encoder = get_encoder(
- props
- .encoding(descr.path())
+ column_props
+ .encoding
.unwrap_or_else(|| fallback_encoding(T::get_physical_type(),
props)),
descr,
)?;
- let statistics_enabled = props.statistics_enabled(descr.path());
+ let statistics_enabled = column_props.statistics_enabled;
- let (bloom_filter, bloom_filter_target_fpp) =
create_bloom_filter(props, descr)?;
+ let (bloom_filter, bloom_filter_target_fpp) =
create_bloom_filter(column_props)?;
let geo_stats_accumulator = try_new_geo_stats_accumulator(descr);
@@ -477,10 +496,9 @@ where
/// Creates a bloom filter sized for the column's configured NDV, returning
the filter
/// and the target FPP for folding.
pub(crate) fn create_bloom_filter(
- props: &WriterProperties,
- descr: &ColumnDescPtr,
+ column_props: &ResolvedColumnProperties,
) -> Result<(Option<Sbbf>, f64)> {
- match props.bloom_filter_properties(descr.path()) {
+ match column_props.bloom_filter_properties.as_ref() {
Some(bf_props) => Ok((
Some(Sbbf::new_with_ndv_fpp(bf_props.ndv(), bf_props.fpp())?),
bf_props.fpp(),
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index c0b7ee04ad..a91d2af53d 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -45,7 +45,8 @@ use crate::file::metadata::{
OffsetIndexBuilder, PageEncodingStats,
};
use crate::file::properties::{
- EnabledStatistics, WriterProperties, WriterPropertiesPtr, WriterVersion,
+ EnabledStatistics, ResolvedColumnProperties, WriterProperties,
WriterPropertiesPtr,
+ WriterVersion,
};
use crate::file::statistics::{Statistics, ValueStatistics};
use crate::schema::types::{BasicTypeInfo, ColumnDescPtr, ColumnDescriptor};
@@ -457,7 +458,10 @@ pub struct GenericColumnWriter<'a, E: ColumnValueEncoder> {
// Column writer properties
descr: ColumnDescPtr,
props: WriterPropertiesPtr,
- statistics_enabled: EnabledStatistics,
+ /// Per-column settings for [`Self::descr`], resolved once here so that the
+ /// per-batch and per-page write paths never search the per-column override
+ /// map in `props` again.
+ column_props: ResolvedColumnProperties,
page_writer: Box<dyn PageWriter + 'a>,
codec: Compression,
@@ -499,12 +503,13 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
props: WriterPropertiesPtr,
page_writer: Box<dyn PageWriter + 'a>,
) -> Self {
- let codec = props.compression(descr.path());
+ let column_props = props.resolve_column_properties(descr.path());
+ let codec = column_props.compression;
let codec_options = CodecOptionsBuilder::default().build();
let compressor = create_codec(codec, &codec_options).unwrap();
- let encoder = E::try_new(&descr, props.as_ref()).unwrap();
+ let encoder = E::try_new(&descr, props.as_ref(),
&column_props).unwrap();
- let statistics_enabled = props.statistics_enabled(descr.path());
+ let statistics_enabled = column_props.statistics_enabled;
let mut encodings = BTreeSet::new();
// Used for level information
@@ -540,7 +545,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
rep_levels_encoder:
Self::create_level_encoder(descr.max_rep_level(), &props),
descr,
props,
- statistics_enabled,
+ column_props,
page_writer,
codec,
compressor,
@@ -636,7 +641,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
};
debug_assert!(base_batch_size > 0);
- let chunker = ByteBudgetChunker::new(&self.descr, &self.props,
base_batch_size);
+ let chunker = ByteBudgetChunker::new(&self.descr, &self.column_props,
base_batch_size);
while levels_offset < num_levels {
let mut end_offset = num_levels.min(levels_offset +
base_batch_size);
@@ -1065,11 +1070,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
#[inline]
fn should_dict_fallback(&self) -> bool {
match self.encoder.estimated_dict_page_size() {
- Some(size) => {
- size >= self
- .props
- .column_dictionary_page_size_limit(self.descr.path())
- }
+ Some(size) => size >= self.column_props.dictionary_page_size_limit,
None => false,
}
}
@@ -1106,7 +1107,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
return;
}
let size = self.encoder.estimated_data_page_size();
- if size >= self.props.column_data_page_size_limit(self.descr.path()) {
+ if size >= self.column_props.data_page_size_limit {
self.page_metrics.page_size_exemption = size;
}
}
@@ -1127,7 +1128,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
.encoder
.estimated_data_page_size()
.saturating_sub(self.page_metrics.page_size_exemption)
- >= self.props.column_data_page_size_limit(self.descr.path())
+ >= self.column_props.data_page_size_limit
}
/// Performs dictionary fallback.
@@ -1417,7 +1418,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
update_min(&self.descr, &min, &mut
self.column_metrics.min_column_value);
update_max(&self.descr, &max, &mut
self.column_metrics.max_column_value);
- (self.statistics_enabled == EnabledStatistics::Page).then_some(
+ (self.column_props.statistics_enabled ==
EnabledStatistics::Page).then_some(
ValueStatistics::new(
Some(min),
Some(max),
@@ -1445,7 +1446,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
// From here on, we only need page statistics if they will be written
to the page header.
let page_statistics = page_statistics
- .filter(|_|
self.props.write_page_header_statistics(self.descr.path()))
+ .filter(|_| self.column_props.write_page_header_statistics)
.map(|stats| self.truncate_statistics(Statistics::from(stats)));
let compressed_page = match self.props.writer_version() {
@@ -1509,9 +1510,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
let buffer_len = buffer.len();
cmpr.compress(&values_data.buf, &mut buffer)?;
let compressed_values_size = buffer.len() - buffer_len;
- let threshold = self
- .props
-
.column_data_page_v2_compression_ratio_threshold(self.descr.path());
+ let threshold =
self.column_props.data_page_v2_compression_ratio_threshold;
if (compressed_values_size as f64) >=
(uncompressed_size as f64) * threshold
{
buffer.truncate(buffer_len);
@@ -1599,7 +1598,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
.set_data_page_offset(data_page_offset)
.set_dictionary_page_offset(dict_page_offset);
- if self.statistics_enabled != EnabledStatistics::None {
+ if self.column_props.statistics_enabled != EnabledStatistics::None {
let backwards_compatible_min_max =
self.descr.sort_order().is_signed();
let distinct_count = self
diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs
index b6369f1ace..8e3895c5a5 100644
--- a/parquet/src/file/properties.rs
+++ b/parquet/src/file/properties.rs
@@ -306,11 +306,7 @@ impl WriterProperties {
///
/// Note: this is a best effort limit based on the write batch size.
pub fn column_data_page_size_limit(&self, col: &ColumnPath) -> usize {
- self.column_properties
- .get(col)
- .and_then(|c| c.data_page_size_limit())
- .or_else(|| self.default_column_properties.data_page_size_limit())
- .unwrap_or(DEFAULT_PAGE_SIZE)
+ resolve_data_page_size_limit(self.column_override(col),
&self.default_column_properties)
}
/// Returns dictionary page size limit.
@@ -326,11 +322,10 @@ impl WriterProperties {
/// Returns dictionary page size limit for a specific column.
pub fn column_dictionary_page_size_limit(&self, col: &ColumnPath) -> usize
{
- self.column_properties
- .get(col)
- .and_then(|c| c.dictionary_page_size_limit())
- .or_else(||
self.default_column_properties.dictionary_page_size_limit())
- .unwrap_or(DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT)
+ resolve_dictionary_page_size_limit(
+ self.column_override(col),
+ &self.default_column_properties,
+ )
}
/// Returns the maximum page row count
@@ -473,14 +468,10 @@ impl WriterProperties {
///
/// Takes precedence over
[`Self::data_page_v2_compression_ratio_threshold`].
pub fn column_data_page_v2_compression_ratio_threshold(&self, col:
&ColumnPath) -> f64 {
- self.column_properties
- .get(col)
- .and_then(|c| c.data_page_v2_compression_ratio_threshold())
- .or_else(|| {
- self.default_column_properties
- .data_page_v2_compression_ratio_threshold()
- })
- .unwrap_or(DEFAULT_DATA_PAGE_V2_COMPRESSION_RATIO_THRESHOLD)
+ resolve_data_page_v2_compression_ratio_threshold(
+ self.column_override(col),
+ &self.default_column_properties,
+ )
}
/// Returns encoding for a data page, when dictionary encoding is enabled.
@@ -510,43 +501,28 @@ impl WriterProperties {
/// If encoding is not set, then column writer will choose the best
encoding
/// based on the column type.
pub fn encoding(&self, col: &ColumnPath) -> Option<Encoding> {
- self.column_properties
- .get(col)
- .and_then(|c| c.encoding())
- .or_else(|| self.default_column_properties.encoding())
+ resolve_encoding(self.column_override(col),
&self.default_column_properties)
}
/// Returns compression codec for a column.
///
/// For more details see
[`WriterPropertiesBuilder::set_column_compression`]
pub fn compression(&self, col: &ColumnPath) -> Compression {
- self.column_properties
- .get(col)
- .and_then(|c| c.compression())
- .or_else(|| self.default_column_properties.compression())
- .unwrap_or(DEFAULT_COMPRESSION)
+ resolve_compression(self.column_override(col),
&self.default_column_properties)
}
/// Returns `true` if dictionary encoding is enabled for a column.
///
/// For more details see
[`WriterPropertiesBuilder::set_dictionary_enabled`]
pub fn dictionary_enabled(&self, col: &ColumnPath) -> bool {
- self.column_properties
- .get(col)
- .and_then(|c| c.dictionary_enabled())
- .or_else(|| self.default_column_properties.dictionary_enabled())
- .unwrap_or(DEFAULT_DICTIONARY_ENABLED)
+ resolve_dictionary_enabled(self.column_override(col),
&self.default_column_properties)
}
/// Returns which statistics are written for a column.
///
/// For more details see
[`WriterPropertiesBuilder::set_statistics_enabled`]
pub fn statistics_enabled(&self, col: &ColumnPath) -> EnabledStatistics {
- self.column_properties
- .get(col)
- .and_then(|c| c.statistics_enabled())
- .or_else(|| self.default_column_properties.statistics_enabled())
- .unwrap_or(DEFAULT_STATISTICS_ENABLED)
+ resolve_statistics_enabled(self.column_override(col),
&self.default_column_properties)
}
/// Returns `true` if [`Statistics`] are to be written to the page header
for a column.
@@ -555,14 +531,10 @@ impl WriterProperties {
///
/// [`Statistics`]: crate::file::statistics::Statistics
pub fn write_page_header_statistics(&self, col: &ColumnPath) -> bool {
- self.column_properties
- .get(col)
- .and_then(|c| c.write_page_header_statistics())
- .or_else(|| {
- self.default_column_properties
- .write_page_header_statistics()
- })
- .unwrap_or(DEFAULT_WRITE_PAGE_HEADER_STATISTICS)
+ resolve_write_page_header_statistics(
+ self.column_override(col),
+ &self.default_column_properties,
+ )
}
/// Returns the [`BloomFilterProperties`] for the given column
@@ -571,10 +543,41 @@ impl WriterProperties {
///
/// For more details see
[`WriterPropertiesBuilder::set_column_bloom_filter_enabled`]
pub fn bloom_filter_properties(&self, col: &ColumnPath) ->
Option<&BloomFilterProperties> {
- self.column_properties
- .get(col)
- .and_then(|c| c.bloom_filter_properties())
- .or_else(||
self.default_column_properties.bloom_filter_properties())
+ resolve_bloom_filter_properties(self.column_override(col),
&self.default_column_properties)
+ }
+
+ /// Returns the per-column override entry for `col`, if any.
+ ///
+ /// This is the only place the per-column map is searched. Searching it
hashes
+ /// `col`, which is a `Vec<String>`, so callers that need more than one
setting
+ /// should go through [`Self::resolve_column_properties`] rather than call
+ /// several single-setting accessors.
+ #[inline]
+ fn column_override(&self, col: &ColumnPath) -> Option<&ColumnProperties> {
+ self.column_properties.get(col)
+ }
+
+ /// Resolves every per-column writer setting for `col` with a single
search of
+ /// the per-column override map.
+ ///
+ /// A column writer needs most of these settings, and needs some of them
again
+ /// on every batch and every page, so it resolves them once when it is
created
+ /// and reads the result from then on.
+ pub(crate) fn resolve_column_properties(&self, col: &ColumnPath) ->
ResolvedColumnProperties {
+ let column = self.column_override(col);
+ let default = &self.default_column_properties;
+ ResolvedColumnProperties {
+ encoding: resolve_encoding(column, default),
+ compression: resolve_compression(column, default),
+ dictionary_enabled: resolve_dictionary_enabled(column, default),
+ statistics_enabled: resolve_statistics_enabled(column, default),
+ write_page_header_statistics:
resolve_write_page_header_statistics(column, default),
+ data_page_size_limit: resolve_data_page_size_limit(column,
default),
+ dictionary_page_size_limit:
resolve_dictionary_page_size_limit(column, default),
+ data_page_v2_compression_ratio_threshold:
+ resolve_data_page_v2_compression_ratio_threshold(column,
default),
+ bloom_filter_properties: resolve_bloom_filter_properties(column,
default).cloned(),
+ }
}
/// Return file encryption properties
@@ -1805,6 +1808,128 @@ impl ColumnProperties {
}
}
+/// Every per-column writer setting for one leaf column, resolved against the
+/// per-column overrides and the file-wide defaults.
+///
+/// Built by [`WriterProperties::resolve_column_properties`].
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct ResolvedColumnProperties {
+ /// See [`WriterProperties::encoding`].
+ pub(crate) encoding: Option<Encoding>,
+ /// See [`WriterProperties::compression`].
+ pub(crate) compression: Compression,
+ /// See [`WriterProperties::dictionary_enabled`].
+ pub(crate) dictionary_enabled: bool,
+ /// See [`WriterProperties::statistics_enabled`].
+ pub(crate) statistics_enabled: EnabledStatistics,
+ /// See [`WriterProperties::write_page_header_statistics`].
+ pub(crate) write_page_header_statistics: bool,
+ /// See [`WriterProperties::column_data_page_size_limit`].
+ pub(crate) data_page_size_limit: usize,
+ /// See [`WriterProperties::column_dictionary_page_size_limit`].
+ pub(crate) dictionary_page_size_limit: usize,
+ /// See
[`WriterProperties::column_data_page_v2_compression_ratio_threshold`].
+ pub(crate) data_page_v2_compression_ratio_threshold: f64,
+ /// See [`WriterProperties::bloom_filter_properties`].
+ pub(crate) bloom_filter_properties: Option<BloomFilterProperties>,
+}
+
+/// Returns the setting read by `get` for `column` if it sets one, otherwise
the
+/// setting on `default`.
+///
+/// `column` is the per-column override entry, if the column has one.
+#[inline]
+fn column_or_default<T>(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+ get: impl Fn(&ColumnProperties) -> Option<T>,
+) -> Option<T> {
+ column.and_then(&get).or_else(|| get(default))
+}
+
+fn resolve_encoding(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> Option<Encoding> {
+ column_or_default(column, default, ColumnProperties::encoding)
+}
+
+fn resolve_compression(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> Compression {
+ column_or_default(column, default,
ColumnProperties::compression).unwrap_or(DEFAULT_COMPRESSION)
+}
+
+fn resolve_dictionary_enabled(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> bool {
+ column_or_default(column, default, ColumnProperties::dictionary_enabled)
+ .unwrap_or(DEFAULT_DICTIONARY_ENABLED)
+}
+
+fn resolve_statistics_enabled(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> EnabledStatistics {
+ column_or_default(column, default, ColumnProperties::statistics_enabled)
+ .unwrap_or(DEFAULT_STATISTICS_ENABLED)
+}
+
+fn resolve_write_page_header_statistics(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> bool {
+ column_or_default(
+ column,
+ default,
+ ColumnProperties::write_page_header_statistics,
+ )
+ .unwrap_or(DEFAULT_WRITE_PAGE_HEADER_STATISTICS)
+}
+
+fn resolve_data_page_size_limit(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> usize {
+ column_or_default(column, default, ColumnProperties::data_page_size_limit)
+ .unwrap_or(DEFAULT_PAGE_SIZE)
+}
+
+fn resolve_dictionary_page_size_limit(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> usize {
+ column_or_default(
+ column,
+ default,
+ ColumnProperties::dictionary_page_size_limit,
+ )
+ .unwrap_or(DEFAULT_DICTIONARY_PAGE_SIZE_LIMIT)
+}
+
+fn resolve_data_page_v2_compression_ratio_threshold(
+ column: Option<&ColumnProperties>,
+ default: &ColumnProperties,
+) -> f64 {
+ column_or_default(
+ column,
+ default,
+ ColumnProperties::data_page_v2_compression_ratio_threshold,
+ )
+ .unwrap_or(DEFAULT_DATA_PAGE_V2_COMPRESSION_RATIO_THRESHOLD)
+}
+
+fn resolve_bloom_filter_properties<'a>(
+ column: Option<&'a ColumnProperties>,
+ default: &'a ColumnProperties,
+) -> Option<&'a BloomFilterProperties> {
+ column
+ .and_then(ColumnProperties::bloom_filter_properties)
+ .or_else(|| default.bloom_filter_properties())
+}
+
/// Reference counted reader properties.
pub type ReaderPropertiesPtr = Arc<ReaderProperties>;
@@ -1936,6 +2061,85 @@ mod tests {
assert_eq!(WriterVersion::PARQUET_2_0.as_num(), 2);
}
+ /// Every setting resolved in one pass must equal what the individual
+ /// per-column accessors return, for a column that overrides settings, a
+ /// column that inherits them, and settings left at their defaults.
+ #[test]
+ fn test_resolve_column_properties_matches_individual_accessors() {
+ let overridden = ColumnPath::from("overridden");
+ let inherited = ColumnPath::from("inherited");
+
+ let props = WriterProperties::builder()
+ .set_encoding(Encoding::DELTA_BINARY_PACKED)
+ .set_compression(Compression::SNAPPY)
+ .set_dictionary_enabled(false)
+ .set_statistics_enabled(EnabledStatistics::Chunk)
+ .set_write_page_header_statistics(false)
+ .set_data_page_size_limit(1111)
+ .set_dictionary_page_size_limit(2222)
+ .set_data_page_v2_compression_ratio_threshold(0.25)
+ .set_bloom_filter_enabled(true)
+ .set_column_encoding(overridden.clone(), Encoding::PLAIN)
+ .set_column_compression(overridden.clone(),
Compression::UNCOMPRESSED)
+ .set_column_dictionary_enabled(overridden.clone(), true)
+ .set_column_statistics_enabled(overridden.clone(),
EnabledStatistics::Page)
+ .set_column_write_page_header_statistics(overridden.clone(), true)
+ .set_column_data_page_size_limit(overridden.clone(), 3333)
+ .set_column_dictionary_page_size_limit(overridden.clone(), 4444)
+
.set_column_data_page_v2_compression_ratio_threshold(overridden.clone(), 0.75)
+ .set_column_bloom_filter_fpp(overridden.clone(), 0.5)
+ .build();
+
+ // A column with no overrides at all, on properties that are themselves
+ // entirely default.
+ let bare = WriterProperties::builder().build();
+
+ for (props, col) in [
+ (&props, &overridden),
+ (&props, &inherited),
+ (&bare, &inherited),
+ ] {
+ let resolved = props.resolve_column_properties(col);
+ assert_eq!(resolved.encoding, props.encoding(col), "{col:?}");
+ assert_eq!(resolved.compression, props.compression(col),
"{col:?}");
+ assert_eq!(
+ resolved.dictionary_enabled,
+ props.dictionary_enabled(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.statistics_enabled,
+ props.statistics_enabled(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.write_page_header_statistics,
+ props.write_page_header_statistics(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.data_page_size_limit,
+ props.column_data_page_size_limit(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.dictionary_page_size_limit,
+ props.column_dictionary_page_size_limit(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.data_page_v2_compression_ratio_threshold,
+ props.column_data_page_v2_compression_ratio_threshold(col),
+ "{col:?}"
+ );
+ assert_eq!(
+ resolved.bloom_filter_properties.as_ref(),
+ props.bloom_filter_properties(col),
+ "{col:?}"
+ );
+ }
+ }
+
#[test]
fn test_writer_properties_default_settings() {
let props = WriterProperties::default();
