alamb commented on code in PR #10505:
URL: https://github.com/apache/arrow-rs/pull/10505#discussion_r3816919041
##########
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
Review Comment:
I found the mention of DELTA_BYTE_ARRAY confusing here as it appears to
come out of the blue -- maybe it can be listed at the bottom of the comment
with PLAIN and DELTA_LENGTH_BYTE_ARRAY
##########
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
Review Comment:
again here it seems like the mention of PLAIN and DELTA_BYTE_ARRAY is out of
the blue (as in why not also include something about DICT encoding too...
Maybe we could just simplify this comment to to say
```rust
/// Returns true if the encoder compresses each value against the value
/// before it *within the current page*.
///
/// Used for encodings such as `DELTA_BYTE_ARRAY` where
/// a page boundary is not free: flushing discards the
/// previous value, so the first value of the next page is stored in
full.
/// The column writer
/// uses this to exempt a page's mandatory first value from the data page
/// byte limit; see [`should_add_data_page`].
```
##########
parquet/src/column/writer/mod.rs:
##########
@@ -248,6 +248,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,
Review Comment:
Does this mean that for large initial values, the page size will be
"max_page_size + size_of(first value in page)" ?
If so, it might make sense to call this `page_size_exemption` or something
like that -- it isn't really bounding the minimum page size (what I think if
when I read floor) but rather is being subtracted before the limit is applied.
##########
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
Review Comment:
the mention of large values / shared prefixes seems like a detail of
DELTA_BYTE_ARRAY -- the first sentence of this comment is probably enough
##########
parquet/src/column/writer/mod.rs:
##########
@@ -1039,6 +1051,42 @@ 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.
Review Comment:
would be nice to make this a real doc link to make sure it doesn't get stale
##########
parquet/src/column/writer/mod.rs:
##########
@@ -2960,6 +3011,159 @@ 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.
+ // `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.
Review Comment:
each value differs from the previous value? Not sure what byte zero means
##########
parquet/src/column/writer/mod.rs:
##########
@@ -2960,6 +3011,159 @@ 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.
+ // `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
Review Comment:
ah! here it even already uses the term `exempted` !
##########
parquet/src/column/writer/mod.rs:
##########
@@ -2960,6 +3011,159 @@ 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.
+ // `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_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
Review Comment:
this is a lot of detail of the internal working of the encoder -- maybe we
can just trim this down to explain what the actual problem is (page size is too
large) and the conditions needed to trigger it (a single null mini batch?)
##########
parquet/src/column/writer/mod.rs:
##########
@@ -1039,6 +1051,42 @@ 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 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`.
Review Comment:
please make a doc comment so it stays linked and not stale
##########
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`.
Review Comment:
If we are going to link to something can it please be a full on Rustdoc link
(e.g. <pre>[`foo`]</pre>) so the doc CI check can make sure it doesn't go stale?
##########
parquet/src/column/writer/mod.rs:
##########
@@ -2960,6 +3011,159 @@ 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.
+ // `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
Review Comment:
I think the rest of this paragraph are getting into a bunch of
implementation details
Simply stating that `A value larger than `data_page_size_limit` blows the
limit on its own resulting in single row pages` is the key idea
##########
parquet/tests/arrow_writer/layout.rs:
##########
@@ -748,6 +748,49 @@ 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 identical values. Cutting one
+ // page per value would reset the encoder's shared-prefix state every
Review Comment:
again woudl recommend removing the implementation details and focus on what
data is expected here -- namely that it should be possible to get all 32 values
on the page, even though it is over the page budget
##########
parquet/src/column/writer/mod.rs:
##########
@@ -2960,6 +3011,159 @@ 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.
+ // `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, \
Review Comment:
What does it mean "one value's worth of bytes" -- the check is that the
encoding reduces the size to be 2x the size of a single value
According to claude The ideal output for 16 identical values is one value
stored in full plus 15 zero-length suffixes -- perhaps that would be clearer to
say explictly / in comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
