fix(parquet): keep DELTA_BYTE_ARRAY dedup for values larger than the page size limit

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(),

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<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();
 

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<LevelHistogram>,
     definition_level_histogram: Option<LevelHistogram>,
@@ -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::<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 ~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::<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);
+
+        // 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

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<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