From d053bf0699fa7e81f61500222f801e898347fa06 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:05:07 -0500 Subject: [PATCH] fix(parquet): keep DELTA_BYTE_ARRAY dedup for values larger than the page limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `BYTE_ARRAY` value larger than `data_page_size_limit` exceeds the limit on its own, so the post-write `should_add_data_page` check cuts a page after every single value. Parquet requires at least one value per data page, so that value is unsplittable and the limit is simply unsatisfiable there. For `DELTA_BYTE_ARRAY` this is destructive rather than merely wasteful: each page boundary discards the encoder's previous value, so every value gets `prefix_length = 0` and the encoding degenerates to exactly `PLAIN`. Columns of large values that share long prefixes stop being deduplicated (#10489). Exempt a page's mandatory first value from the byte limit when that value alone already exceeds it, so the limit applies to what follows — the bytes we can still place on another page. The exemption is gated on `ColumnValueEncoder::compresses_against_previous_value`, so only `DELTA_BYTE_ARRAY` opts in; `PLAIN` and `DELTA_LENGTH_BYTE_ARRAY` cost the same wherever a value lands and keep their existing one-value page bound. Pages therefore stay bounded by the value size rather than growing with `write_batch_size`, preserving the fix from #9972. Co-Authored-By: Claude Opus 5 --- parquet/src/arrow/arrow_writer/byte_array.rs | 8 ++ parquet/src/column/writer/encoder.rs | 24 ++++ parquet/src/column/writer/mod.rs | 136 ++++++++++++++++++- parquet/tests/arrow_writer_layout.rs | 42 ++++++ 4 files changed, 209 insertions(+), 1 deletion(-) diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs b/parquet/src/arrow/arrow_writer/byte_array.rs index 145431c26465..6160c67e3fa0 100644 --- a/parquet/src/arrow/arrow_writer/byte_array.rs +++ b/parquet/src/arrow/arrow_writer/byte_array.rs @@ -572,6 +572,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 d9adacff4101..b555f6d88177 100644 --- a/parquet/src/column/writer/encoder.rs +++ b/parquet/src/column/writer/encoder.rs @@ -132,6 +132,24 @@ 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 + /// before it *within the current page* — today, `DELTA_BYTE_ARRAY` and + /// its shared-prefix lengths. + /// + /// 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. + /// A column of large values that share long prefixes therefore collapses + /// to `PLAIN` if every value is cut into its own page. The column writer + /// uses this to exempt a page's mandatory first value from the data page + /// byte limit; see `should_add_data_page`. + /// + /// Defaults to `false`: for `PLAIN` and `DELTA_LENGTH_BYTE_ARRAY` a + /// value costs the same wherever it lands, so there is nothing to + /// preserve by keeping values together. + fn compresses_against_previous_value(&self) -> bool { + false + } + /// Returns the estimated total memory usage of the encoder /// fn estimated_memory_size(&self) -> usize; @@ -324,6 +342,12 @@ impl ColumnValueEncoder for ColumnValueEncoderImpl { 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 aa9cef16c5ad..4389b2098bb3 100644 --- a/parquet/src/column/writer/mod.rs +++ b/parquet/src/column/writer/mod.rs @@ -245,6 +245,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_floor: usize, num_page_nulls: u64, repetition_level_histogram: Option, definition_level_histogram: Option, @@ -272,6 +278,7 @@ impl PageMetrics { fn new_page(&mut self) { self.num_buffered_values = 0; self.num_buffered_rows = 0; + self.page_size_floor = 0; self.num_page_nulls = 0; self.repetition_level_histogram .as_mut() @@ -1001,8 +1008,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_floor(); + } + if self.should_add_data_page() { self.add_data_page()?; } @@ -1030,6 +1042,32 @@ 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 + /// `should_add_data_page` then cuts a page after every single value. For + /// `DELTA_BYTE_ARRAY` that is destructive rather than merely wasteful: + /// each page boundary discards the previous value, so a column of large + /// values sharing long prefixes degenerates to `PLAIN` (#10489). + /// + /// Recording the value's encoded size here makes the limit apply to what + /// follows it — the bytes we *can* still place elsewhere. Only encodings + /// that compress against the preceding value opt in, so `PLAIN` and + /// `DELTA_LENGTH_BYTE_ARRAY` keep their tighter one-value page bound. + #[cold] + fn set_page_size_floor(&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_floor = size; + } + } + /// Returns true if there is enough data for a data page, false otherwise. #[inline] fn should_add_data_page(&self) -> bool { @@ -1042,7 +1080,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_floor) >= self.props.column_data_page_size_limit(self.descr.path()) } @@ -2890,6 +2931,99 @@ mod tests { } } + #[test] + fn test_column_writer_delta_byte_array_dedups_large_shared_prefix_values() { + // Regression for #10489. `DELTA_BYTE_ARRAY` stores each value as a + // prefix shared with its predecessor plus a suffix, but that context + // is per-page: flushing a page clears the encoder's `previous`, so + // the first value on every page pays full price. + // + // A value larger than `data_page_size_limit` blows the limit on its + // own, so the post-write `should_add_data_page` check cuts a page + // after every single value and the encoding degenerates to exactly + // PLAIN. Parquet requires at least one value per page, so that first + // value is unsplittable and must be exempt from the limit — the + // values after it then cost ~nothing. + 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::(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 ~one value's worth of bytes for {num_rows} identical values, \ + got {total_bytes}B across pages {:?}", + pages.data_pages, + ); + } + + #[test] + fn test_column_writer_delta_byte_array_bounds_pages_without_shared_prefix() { + // Companion to the test above, and the reason the first-value + // exemption is scoped to *one* value rather than dropping the byte + // budget altogether: when large values share no prefix there is + // nothing to dedup, and pages must stay bounded by the value size + // rather than growing with `write_batch_size`. + 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(); + + // Each value differs from byte 0, so every prefix length is 0. + let data: Vec<_> = (0..num_rows) + .map(|i| ByteArray::from(vec![i as u8; value_size])) + .collect(); + let pages = write_and_collect_pages::(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); + + // The exempted first value plus one more that trips the budget: at + // most two values' worth of payload on any page, never the whole + // 1024-row mini-batch. + 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_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 1c63a3144391..12c1355decb8 100644 --- a/parquet/tests/arrow_writer_layout.rs +++ b/parquet/tests/arrow_writer_layout.rs @@ -748,6 +748,48 @@ fn test_large_string() { }); } +#[test] +fn test_large_string_delta_byte_array_shared_prefix() { + // Regression for #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 identical values. Cutting one + // page per value would reset the encoder's shared-prefix state every + // time and reproduce `PLAIN` byte for byte (32 pages, ~2 MiB). Instead + // the page's mandatory first value is exempt from the byte limit, so all + // 32 values share one page: one value stored in full plus 31 empty + // suffixes. + let value_size = 64 * 1024; + let strings: Vec = (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