This is an automated email from the ASF dual-hosted git repository.
alamb 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 4ceed4d86a fix(parquet): keep DELTA_BYTE_ARRAY dedup for values larger
than the page size limit (#10505)
4ceed4d86a is described below
commit 4ceed4d86af4314ddf61ef6267655addba6a0f19
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Mon Aug 24 16:31:18 2026 -0500
fix(parquet): keep DELTA_BYTE_ARRAY dedup for values larger than the page
size limit (#10505)
- closes https://github.com/apache/arrow-rs/issues/10489
- closes https://github.com/apache/arrow-rs/pull/10504
## The cause is the page flush, not the mini-batch splitting
The issue attributes the regression to #9972's byte-budget sub-batching.
It's actually the page flush; the sub-batching only exposes it.
`should_add_data_page` fires when `estimated_data_page_size() >=
data_page_size_limit`. For `DELTA_BYTE_ARRAY` that estimate is the real
encoded size, and the first value on a page is stored in full — so a
single 8 MiB value against a 1 MiB limit puts the page over the limit by
itself and triggers a flush. Flushing clears the encoder's `previous`,
so the next value gets `prefix_length = 0`, and so on. The encoding
degenerates to exactly `PLAIN`.
Before #9972 a 1024-row mini-batch meant the check simply didn't run
until 1024 values had been written, so the dedup survived. That was an
accident of `write_batch_size`, not a designed property — the same
column at 2000 rows already lost the prefix at the 1024-value boundary.
This matters for choosing a fix: **making the byte budget encoding-aware
(suggestion 2 in the issue) does not fix the reported bug.** However
precise the budget, the first value alone still exceeds the limit and
still triggers the flush.
## The fix
Parquet requires at least one value per data page, so a value larger
than the limit cannot be split out — the limit is unsatisfiable for it.
Counting those bytes against the limit is what forces the pathological
one-value-per-page cut.
Record that first value's encoded size in `PageMetrics::page_size_floor`
and apply the limit to what follows it, i.e. the bytes that *can* still
go on another page.
The exemption is gated on a new
`ColumnValueEncoder::compresses_against_previous_value` (default
`false`, `true` only for `DELTA_BYTE_ARRAY` in both the generic and
arrow encoders). `PLAIN` and `DELTA_LENGTH_BYTE_ARRAY` cost the same
wherever a value lands, so there is nothing to preserve by keeping
values together and they keep their existing tighter one-value page
bound. All three regression tests from #9972 pass unmodified.
## Why not "don't split when the split cannot help"
Suggestion 1 in the issue restores the dedup, but by removing the bound
that #9972 added, for exactly the workload it was added for. Measured
with `Some(1) => chunk_size` in `byte_budget_chunker.rs`, 2048 rows ×
128 KiB distinct values against a 64 KiB page limit — the same regime as
a 10 MB value against the 1 MiB default, scaled to fit in RAM:
| | pages | max page |
|---|---|---|
| 59.1.0 | 2048 | 128 KiB |
| suggestion 1 | 2 | **128 MiB** |
| this PR | 1024 | 256 KiB |
Page size under suggestion 1 is `write_batch_size × value_size`. At 1024
× 10 MB that is a 10 GB page, which is the failure #9972 fixed.
## Results
The reporter's table, reproduced verbatim (10 identical 8 MiB values,
raw input 80 MiB):
| `data_page_size_limit` | 58.4.0 | 59.1.0 | this PR |
| --- | --- | --- | --- |
| default (1 MiB) | 8.00 MiB | 80.00 MiB | **8.00 MiB** |
| 4 MiB | 8.00 MiB | 80.00 MiB | **8.00 MiB** |
| 8 MiB | 8.00 MiB | 80.00 MiB | **8.00 MiB** |
| 16 MiB | 8.00 MiB | 8.00 MiB | **8.00 MiB** |
Values that merely share a long prefix behave the same. Values sharing
*no* prefix are unaffected in file size and stay bounded at up to two
values per page — the exempt one plus the one that trips the budget.
## Tests
-
`test_column_writer_delta_byte_array_dedups_large_shared_prefix_values`
— 16 identical 64 KiB values, 16 KiB page limit. Fails on `main` with 1
MiB across 16 pages (byte for byte what `PLAIN` produces); passes here
at ~one value's worth.
-
`test_column_writer_delta_byte_array_bounds_pages_without_shared_prefix`
— the same column with values differing from byte 0, asserting pages
stay bounded by two values. Guards against fixing this by dropping the
bound.
- `test_large_string_delta_byte_array_shared_prefix` — the `ArrowWriter`
path from the report, as an exact page-layout assertion.
## Notes
One known limitation, pinned by
`test_column_writer_delta_byte_array_nullable_shared_prefix_partial_dedup`:
the exemption triggers when a page opens with a mini-batch holding
exactly one value. That is guaranteed for chunks without nulls (the
byte-budget chunker gives an over-limit value its own mini-batch), but a
null in a chunk makes the chunker convert values to levels by ratio and
round up, so pages can open with a two-value mini-batch, miss the
exemption, and store their first value in full. Dedup on such columns is
partial rather than absent, and never worse than `main`, which stores
every value in full. The clean fix is for the granular path to cut
windows after an exact value count instead of ratio-scaling; that is
independent of this change and tracked in #10538. The pinning test
documents the layout to expect once it lands.
A follow-up worth considering separately: the byte budget in
`count_within_budget_*` still measures raw payload length, so
`DELTA_BYTE_ARRAY` columns sub-batch more eagerly than the encoded size
warrants. That is a throughput question rather than a correctness one,
and it is not what caused this regression.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 <[email protected]>
---
parquet/src/arrow/arrow_writer/byte_array.rs | 8 ++
parquet/src/column/writer/encoder.rs | 27 ++++
parquet/src/column/writer/mod.rs | 198 ++++++++++++++++++++++++++-
parquet/tests/arrow_writer/layout.rs | 40 ++++++
4 files changed, 272 insertions(+), 1 deletion(-)
diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs
b/parquet/src/arrow/arrow_writer/byte_array.rs
index 5dbf70cbd8..6346199b16 100644
--- a/parquet/src/arrow/arrow_writer/byte_array.rs
+++ b/parquet/src/arrow/arrow_writer/byte_array.rs
@@ -567,6 +567,14 @@ impl ColumnValueEncoder for ByteArrayEncoder {
self.dict_encoder.is_some()
}
+ fn compresses_against_previous_value(&self) -> bool {
+ // While dictionary encoding is active the data page holds RLE
+ // indices, which carry no cross-value state; only the DELTA_BYTE_ARRAY
+ // fallback shares prefixes with the preceding value.
+ self.dict_encoder.is_none()
+ && matches!(self.fallback.encoder, FallbackEncoderImpl::Delta { ..
})
+ }
+
fn estimated_memory_size(&self) -> usize {
let encoder_size = match &self.dict_encoder {
Some(encoder) => encoder.estimated_memory_size(),
diff --git a/parquet/src/column/writer/encoder.rs
b/parquet/src/column/writer/encoder.rs
index e7b548f0dc..fe644d72e0 100644
--- a/parquet/src/column/writer/encoder.rs
+++ b/parquet/src/column/writer/encoder.rs
@@ -132,6 +132,27 @@ pub trait ColumnValueEncoder {
/// Returns true if this encoder has a dictionary page
fn has_dictionary(&self) -> bool;
+ /// Returns true if the encoder compresses each value against the value
+ /// immediately before it, within the current page.
+ ///
+ /// For such encodings a page boundary is not free: flushing discards the
+ /// previous value, so the first value of the next page is stored in full.
+ /// [`GenericColumnWriter::should_add_data_page`] uses this to exempt a
+ /// page's mandatory first value from the data page byte limit.
+ ///
+ /// Per encoding:
+ /// * `DELTA_BYTE_ARRAY`: true. Each value is stored as the length of the
+ /// prefix it shares with its predecessor plus the remaining suffix.
+ /// * Everything else: false, the default. `PLAIN` and
+ /// `DELTA_LENGTH_BYTE_ARRAY` store a value at the same cost wherever it
+ /// lands, and a dictionary outlives the pages that index into it, so no
+ /// page boundary makes a value more expensive.
+ ///
+ /// [`GenericColumnWriter::should_add_data_page`]:
crate::column::writer::GenericColumnWriter::should_add_data_page
+ fn compresses_against_previous_value(&self) -> bool {
+ false
+ }
+
/// Returns the estimated total memory usage of the encoder
///
fn estimated_memory_size(&self) -> usize;
@@ -329,6 +350,12 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
self.dict_encoder.is_some()
}
+ fn compresses_against_previous_value(&self) -> bool {
+ // While dictionary encoding is active `self.encoder` is unused: the
+ // data page holds RLE indices, which carry no cross-value state.
+ self.dict_encoder.is_none() && self.encoder.encoding() ==
Encoding::DELTA_BYTE_ARRAY
+ }
+
fn estimated_memory_size(&self) -> usize {
let encoder_size = self.encoder.estimated_memory_size();
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 6bda82d646..6ef18419f2 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -256,6 +256,12 @@ impl ColumnCloseResult {
struct PageMetrics {
num_buffered_values: u32,
num_buffered_rows: u32,
+ /// Encoded bytes that the data page byte limit does not apply to,
+ /// because they belong to the page's mandatory first value and cannot be
+ /// moved elsewhere. Zero unless that value alone exceeded the limit
+ /// *and* the encoding compresses against the preceding value; see
+ /// [`ColumnValueEncoder::compresses_against_previous_value`].
+ page_size_exemption: usize,
num_page_nulls: u64,
num_page_nans: Option<u64>,
repetition_level_histogram: Option<LevelHistogram>,
@@ -284,6 +290,7 @@ impl PageMetrics {
fn new_page(&mut self) {
self.num_buffered_values = 0;
self.num_buffered_rows = 0;
+ self.page_size_exemption = 0;
self.num_page_nulls = 0;
self.num_page_nans = None;
self.repetition_level_histogram
@@ -1033,8 +1040,13 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
None => self.encoder.write(values, values_offset,
values_to_write)?,
}
+ let page_was_empty = self.page_metrics.num_buffered_values == 0;
self.page_metrics.num_buffered_values += num_levels as u32;
+ if page_was_empty && values_to_write == 1 {
+ self.set_page_size_exemption();
+ }
+
if self.should_add_data_page() {
self.add_data_page()?;
}
@@ -1062,6 +1074,43 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
}
}
+ /// Exempt a page's mandatory first value from the data page byte limit,
+ /// when that value alone already exceeds it.
+ ///
+ /// Parquet requires every data page to hold at least one value, so such a
+ /// value cannot be split out no matter how the limit is set. Counting it
+ /// against the limit makes the limit unsatisfiable, and
+ /// [`Self::should_add_data_page`] then cuts a page after every single
+ /// value.
+ ///
+ /// For `DELTA_BYTE_ARRAY` that costs more than the extra pages. A value is
+ /// stored as a suffix of the value before it, and a page boundary resets
+ /// what "the value before it" refers to, so one value per page means every
+ /// value is stored in full: a column of large values sharing long prefixes
+ /// writes exactly the bytes `PLAIN` would
+ /// ([#10489](https://github.com/apache/arrow-rs/issues/10489)).
+ ///
+ /// Only encodings that compress against the preceding value opt in, so
+ /// `PLAIN` and `DELTA_LENGTH_BYTE_ARRAY` keep their tighter one-value page
+ /// bound.
+ ///
+ /// Known limitation: the caller's trigger keys on a page-opening
+ /// mini-batch holding exactly one value. Nulls in a chunk make the
+ /// byte-budget chunker emit multi-level mini-batches, so on nullable
+ /// columns pages that open with a two-value mini-batch miss the
+ /// exemption and dedup is only partial; see
+ ///
`test_column_writer_delta_byte_array_nullable_shared_prefix_partial_dedup`.
+ #[cold]
+ fn set_page_size_exemption(&mut self) {
+ if !self.encoder.compresses_against_previous_value() {
+ return;
+ }
+ let size = self.encoder.estimated_data_page_size();
+ if size >= self.props.column_data_page_size_limit(self.descr.path()) {
+ self.page_metrics.page_size_exemption = size;
+ }
+ }
+
/// Returns true if there is enough data for a data page, false otherwise.
#[inline]
fn should_add_data_page(&self) -> bool {
@@ -1074,7 +1123,10 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
}
self.page_metrics.num_buffered_rows as usize >=
self.props.data_page_row_count_limit()
- || self.encoder.estimated_data_page_size()
+ || self
+ .encoder
+ .estimated_data_page_size()
+ .saturating_sub(self.page_metrics.page_size_exemption)
>= self.props.column_data_page_size_limit(self.descr.path())
}
@@ -2986,6 +3038,150 @@ mod tests {
}
}
+ #[test]
+ fn test_column_writer_delta_byte_array_dedups_large_shared_prefix_values()
{
+ // Regression for https://github.com/apache/arrow-rs/issues/10489.
+ // 16 identical 64 KiB values against a 16 KiB page limit: every value
+ // is over the limit on its own, and `DELTA_BYTE_ARRAY` should still
+ // dedup them down to about one value's worth of bytes in total.
+ let value_size = 64 * 1024; // 64 KiB per value, > the page limit
+ let page_byte_limit = 16 * 1024;
+ let num_rows = 16;
+
+ let props = WriterProperties::builder()
+ .set_writer_version(WriterVersion::PARQUET_1_0)
+ .set_dictionary_enabled(false)
+ .set_encoding(Encoding::DELTA_BYTE_ARRAY)
+ .set_data_page_size_limit(page_byte_limit)
+ .set_statistics_enabled(EnabledStatistics::None)
+ .build();
+
+ // Identical values: one full value plus `num_rows - 1` zero-length
+ // suffixes is all this column should cost.
+ let data: Vec<_> = (0..num_rows)
+ .map(|_| ByteArray::from(vec![b'a'; value_size]))
+ .collect();
+ let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0,
&data, None, None);
+
+ // Every value must still end up somewhere.
+ let total_values: u32 = pages.data_pages.iter().map(|(_, n)| n).sum();
+ assert_eq!(total_values as usize, num_rows);
+
+ // Before the fix this was `num_rows * value_size` — byte for byte
+ // what PLAIN produces, i.e. the encoding doing no work at all.
+ let total_bytes: usize = pages.data_pages.iter().map(|(size, _)|
size).sum();
+ assert!(
+ total_bytes < 2 * value_size,
+ "expected under 2x a single value ({}B) for {num_rows} identical \
+ values, got {total_bytes}B across pages {:?}",
+ 2 * value_size,
+ pages.data_pages,
+ );
+ }
+
+ #[test]
+ fn
test_column_writer_delta_byte_array_bounds_pages_without_shared_prefix() {
+ // Companion to the test above: same shape, but the values share no
+ // prefix, so there is nothing to dedup and pages must stay bounded
+ // by the value size. This is why the exemption covers one value
+ // rather than dropping the byte budget altogether.
+ let value_size = 64 * 1024;
+ let page_byte_limit = 16 * 1024;
+ let num_rows = 16;
+
+ let props = WriterProperties::builder()
+ .set_writer_version(WriterVersion::PARQUET_1_0)
+ .set_dictionary_enabled(false)
+ .set_encoding(Encoding::DELTA_BYTE_ARRAY)
+ .set_data_page_size_limit(page_byte_limit)
+ .set_statistics_enabled(EnabledStatistics::None)
+ .build();
+
+ // No two values share a prefix: they differ at the first byte.
+ let data: Vec<_> = (0..num_rows)
+ .map(|i| ByteArray::from(vec![i as u8; value_size]))
+ .collect();
+ let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0,
&data, None, None);
+
+ let total_values: u32 = pages.data_pages.iter().map(|(_, n)| n).sum();
+ assert_eq!(total_values as usize, num_rows);
+
+ // Expect at most two values per page: the exempted first value plus
+ // one more that trips the budget.
+ let upper_bound = 2 * value_size + 64;
+ for (size, n_values) in &pages.data_pages {
+ assert!(
+ *size <= upper_bound,
+ "page size {size} exceeds two-value bound ({upper_bound}B);
pages {:?}",
+ pages.data_pages,
+ );
+ assert!(
+ *n_values <= 2,
+ "page holds {n_values} values, expected at most 2; pages {:?}",
+ pages.data_pages,
+ );
+ }
+ }
+
+ #[test]
+ fn
test_column_writer_delta_byte_array_nullable_shared_prefix_partial_dedup() {
+ // Documents the *current* behavior of the first-value exemption on a
+ // nullable column; this pins a known limitation, not an ideal.
+ //
+ // The exemption fires when a page's first mini-batch contains exactly
+ // one value. For a non-nullable column the byte-budget chunker gives
+ // an over-limit value a one-level mini-batch, so that always holds.
+ // One null in the chunk changes the level:value ratio to 17:16, the
+ // chunker rounds up to two-level mini-batches, and a page whose first
+ // mini-batch carries two values misses the exemption: it is cut after
+ // those two values, and its first value is stored in full.
+ //
+ // The one mini-batch that pairs the null with a value has a single
+ // value, so the page it opens does get the exemption and accumulates
+ // every remaining suffix. The result for 16 identical values with a
+ // null at index 8 is four two-value pages (each storing one value in
+ // full), then one exempt page holding the rest:
+ //
+ // values per page: [2, 2, 2, 2, 9] (counts include the null level)
+ // total bytes: ~5 full values, vs ~1 ideally and 16 for PLAIN
+ //
+ // If the exemption trigger is ever keyed on values written to the
+ // page (0 -> 1) instead of mini-batch shape, this test should fail
+ // with fewer, larger pages — update it to pin the improved layout.
+ let value_size = 64 * 1024;
+ let page_byte_limit = 16 * 1024;
+ let num_values = 16;
+
+ let props = WriterProperties::builder()
+ .set_writer_version(WriterVersion::PARQUET_1_0)
+ .set_dictionary_enabled(false)
+ .set_encoding(Encoding::DELTA_BYTE_ARRAY)
+ .set_data_page_size_limit(page_byte_limit)
+ .set_statistics_enabled(EnabledStatistics::None)
+ .build();
+
+ let data: Vec<_> = (0..num_values)
+ .map(|_| ByteArray::from(vec![b'a'; value_size]))
+ .collect();
+ // 17 levels: a null at index 8, values everywhere else.
+ let def_levels: Vec<i16> = (0..num_values as i16 + 1)
+ .map(|i| i16::from(i != 8))
+ .collect();
+ let pages =
+ write_and_collect_pages::<ByteArrayType>(props, 1, 0, &data,
Some(&def_levels), None);
+
+ let per_page_values: Vec<u32> = pages.data_pages.iter().map(|(_, n)|
*n).collect();
+ assert_eq!(per_page_values, vec![2, 2, 2, 2, 9]);
+
+ let total_bytes: usize = pages.data_pages.iter().map(|(size, _)|
size).sum();
+ assert!(
+ total_bytes > 4 * value_size && total_bytes < 6 * value_size,
+ "expected ~5 full values' worth of bytes (partial dedup), \
+ got {total_bytes}B across pages {:?}",
+ pages.data_pages,
+ );
+ }
+
#[test]
fn test_column_writer_caps_page_size_for_large_values_in_list() {
// Coverage for the Materialized-rep branch of
diff --git a/parquet/tests/arrow_writer/layout.rs
b/parquet/tests/arrow_writer/layout.rs
index 55a489272f..04f0a9cbed 100644
--- a/parquet/tests/arrow_writer/layout.rs
+++ b/parquet/tests/arrow_writer/layout.rs
@@ -755,6 +755,46 @@ fn test_large_string() {
});
}
+#[test]
+fn test_large_string_delta_byte_array_shared_prefix() {
+ // Regression for https://github.com/apache/arrow-rs/issues/10489, at the
+ // `ArrowWriter` level the report used.
+ //
+ // Same shape as `test_large_string` — 64 KiB values against a 16 KiB
+ // page limit — but `DELTA_BYTE_ARRAY` and 32 identical values. Expect a
+ // single page holding all 32 rows and about one value's worth of bytes,
+ // rather than the 32 pages and ~2 MiB `PLAIN` produces.
+ let value_size = 64 * 1024;
+ let strings: Vec<String> = (0..32).map(|_|
"x".repeat(value_size)).collect();
+ let array = Arc::new(StringArray::from(strings)) as _;
+ let batch = RecordBatch::try_from_iter([("col", array)]).unwrap();
+ let props = WriterProperties::builder()
+ .set_dictionary_enabled(false)
+ .set_encoding(Encoding::DELTA_BYTE_ARRAY)
+ .set_data_page_size_limit(16 * 1024)
+ .set_statistics_enabled(EnabledStatistics::None)
+ .build();
+
+ do_test(LayoutTest {
+ props,
+ batches: vec![batch],
+ layout: Layout {
+ row_groups: vec![RowGroup {
+ columns: vec![ColumnChunk {
+ pages: vec![Page {
+ rows: 32,
+ page_header_size: 21,
+ compressed_size: 65696,
+ encoding: Encoding::DELTA_BYTE_ARRAY,
+ page_type: PageType::DATA_PAGE,
+ }],
+ dictionary_page: None,
+ }],
+ }],
+ },
+ });
+}
+
#[test]
fn test_large_string_view() {
// Same bytes and expected layout as `test_large_string`, but the input
