Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
ef40c46
test(parquet): add page-size regression tests for large values
adriangb May 25, 2026
17f7cc8
fix(parquet): bound data page byte size for large variable-width values
adriangb May 25, 2026
0ee44d9
fix(parquet): bound page size for arrow byte-array column writes
adriangb May 25, 2026
c2a7fba
fix(parquet): bound dictionary page size during dictionary encoding
adriangb May 25, 2026
e05ffca
fix(parquet): include boundary value in mini-batch byte-budget count
adriangb May 28, 2026
4769588
test(parquet): update arrow_writer_layout dict-spill expectations
adriangb May 28, 2026
c018f5f
test(parquet): add arrow_writer_layout coverage for dictionary + Fixe…
alamb Jun 3, 2026
8574187
test(parquet): add page-size regression tests for large values
adriangb May 25, 2026
e20609a
fix(parquet): bound data page byte size for large variable-width values
adriangb May 25, 2026
7298c48
fix(parquet): bound page size for arrow byte-array column writes
adriangb May 25, 2026
ef7800b
fix(parquet): bound dictionary page size during dictionary encoding
adriangb May 25, 2026
b3446a5
fix(parquet): include boundary value in mini-batch byte-budget count
adriangb May 28, 2026
bcdb878
test(parquet): update arrow_writer_layout dict-spill expectations
adriangb May 28, 2026
7d305a6
Merge pull request #16 from alamb/alamb/parquet-page-size-layout-tests
adriangb Jun 3, 2026
3a7161f
fix(parquet): clean up byte-array page-sizing dead code + lint
adriangb Jun 3, 2026
67587ad
test(parquet): trim verbose comments in arrow_writer_layout tests
alamb Jun 3, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 196 additions & 1 deletion parquet/src/arrow/arrow_writer/byte_array.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,12 @@ use crate::geospatial::statistics::GeospatialStatistics;
use crate::schema::types::ColumnDescPtr;
use crate::util::bit_util::num_required_bits;
use crate::util::interner::{Interner, Storage};
use arrow_array::types::ByteArrayType;
use arrow_array::{
Array, ArrayAccessor, BinaryArray, BinaryViewArray, DictionaryArray, FixedSizeBinaryArray,
LargeBinaryArray, LargeStringArray, StringArray, StringViewArray,
GenericByteArray, LargeBinaryArray, LargeStringArray, StringArray, StringViewArray,
};
use arrow_buffer::{ArrowNativeType, Buffer};
use arrow_schema::DataType;

macro_rules! downcast_dict_impl {
Expand Down Expand Up @@ -481,6 +483,88 @@ impl ColumnValueEncoder for ByteArrayEncoder {
Ok(())
}

fn count_values_within_byte_budget_gather(
values: &Self::Values,
indices: &[usize],
byte_budget: usize,
) -> Option<usize> {
// `ByteArrayEncoder` only ever writes via `write_gather`, so this
// is the relevant method.
//
// Two-stage walk for the simple offset-buffer byte array types:
// 1. If indices are contiguous, compute the total payload in
// O(1) via a single subtraction on the offsets buffer.
// When the total fits the budget — the overwhelmingly
// common "small values" case — return immediately.
// 2. Otherwise, walk per-value byte sizes from the offsets
// buffer (still cheap, no slice/UTF-8 construction) and
// exit at the first value that pushes the cumulative sum
// past the budget. This bounds skewed distributions: an
// outlier value is caught wherever it lands in the chunk.
let count = match values.data_type() {
DataType::Utf8 => count_within_budget_offsets(
values.as_any().downcast_ref::<StringArray>().unwrap(),
indices,
byte_budget,
),
DataType::LargeUtf8 => count_within_budget_offsets(
values.as_any().downcast_ref::<LargeStringArray>().unwrap(),
indices,
byte_budget,
),
DataType::Binary => count_within_budget_offsets(
values.as_any().downcast_ref::<BinaryArray>().unwrap(),
indices,
byte_budget,
),
DataType::LargeBinary => count_within_budget_offsets(
values.as_any().downcast_ref::<LargeBinaryArray>().unwrap(),
indices,
byte_budget,
),
// View arrays carry each value's length in the low 32 bits of
// its u128 view word, so lengths are scannable without touching
// any data buffer — and the common small-value case skips even
// that scan via an O(1) conservative bound.
DataType::Utf8View => {
let array = values.as_any().downcast_ref::<StringViewArray>().unwrap();
count_within_budget_views(
array.views(),
indices,
byte_budget,
max_view_value_len(array.data_buffers()),
)
}
DataType::BinaryView => {
let array = values.as_any().downcast_ref::<BinaryViewArray>().unwrap();
count_within_budget_views(
array.views(),
indices,
byte_budget,
max_view_value_len(array.data_buffers()),
)
}
// For arrow Dictionary input, treat every chunk as fitting
// and stay on the batched path. The arrow array being
// Dictionary-encoded in the first place implies its values
// are small enough that dedup is worthwhile, which is the
// opposite of the "5 MiB blob per row" case this fix
// targets. Doing a per-value walk through dict keys (each
// value lookup is keys[i] → values[key] → slice) on every
// chunk costs ~+30-80% vs `main` after writer-dict spill,
// and there is essentially nothing to bound.
DataType::Dictionary(_, _) => indices.len(),
// Every byte-array type `ByteArrayEncoder` is constructed for
// has an explicit arm above. A `Dictionary(value = FixedSizeBinary)`
// column hits the `Dictionary(_, _)` arm (its `values.data_type()`
// is `Dictionary`), and a bare `FixedSizeBinary` column is routed
// to the generic column writer, never this encoder — so no other
// type can reach here.
data_type => unreachable!("ByteArrayEncoder cannot be constructed for {data_type:?}"),
};
Some(count)
}

fn num_values(&self) -> usize {
match &self.dict_encoder {
Some(encoder) => encoder.indices.len(),
Expand Down Expand Up @@ -593,6 +677,117 @@ where
}
}

/// Upper bound on any single value's byte length in a view array.
///
/// An inline view stores at most 12 bytes; an
/// out-of-line view's data is a contiguous slice of exactly one data
/// buffer, so it cannot be longer than the largest data buffer. This is a
/// loose bound (a value is usually far smaller than a whole buffer) but it
/// is O(number of buffers) and always sound.
fn max_view_value_len(buffers: &[Buffer]) -> usize {
/// Bytes that fit inline in a u128 view word (the rest is len + prefix).
const MAX_INLINE_VIEW_LEN: usize = 12;
buffers
.iter()
.map(|b| b.len())
.max()
.unwrap_or(0)
.max(MAX_INLINE_VIEW_LEN)
}

/// Two-stage budget count for view arrays (`Utf8View`, `BinaryView`).
///
/// 1. View arrays have no prefix-sum offsets buffer, so the exact O(1)
/// span subtraction `count_within_budget_offsets` uses is unavailable.
/// But a *conservative* O(1) bound is: every value is at most
/// `max_value_len` bytes, so the whole chunk fits the budget when
/// `n * (max_value_len + 4) <= byte_budget`. This skips the per-value
/// walk for the common small-value case — what view arrays are built
/// for, and exactly the case where there is nothing to bound.
/// 2. Otherwise scan per-value lengths from the low 32 bits of each u128
/// view word (no data-buffer dereference) and stop at the first value
/// that pushes the cumulative sum past the budget.
fn count_within_budget_views(
views: &[u128],
indices: &[usize],
byte_budget: usize,
max_value_len: usize,
) -> usize {
// Stage 1: O(1) conservative upper bound.
let per_value = max_value_len + std::mem::size_of::<u32>();
if indices.len().saturating_mul(per_value) <= byte_budget {
return indices.len();
}
// Stage 2: exact per-value scan.
let mut cum: usize = 0;
for (i, idx) in indices.iter().enumerate() {
let len = (views[*idx] as u32) as usize;
cum = cum.saturating_add(len + std::mem::size_of::<u32>());
if cum > byte_budget {
return i + 1;
}
}
indices.len()
}

/// Two-stage fast path for `GenericByteArray<O>`
/// (Utf8/LargeUtf8/Binary/LargeBinary).
///
/// `indices` are assumed sorted ascending — they always are here, since
/// they come from `non_null_indices`, which is built in array order.
///
/// 1. The span `offsets[last+1] - offsets[first]` is an O(1) upper
/// bound on the chunk's payload: it covers every array position in
/// `[first, last]`, a superset of `indices`. For a non-null chunk
/// `indices` *is* that whole range; for a chunk drawn from a
/// nullable column the skipped positions are nulls, whose offset
/// delta is zero, so the span still equals the exact payload.
/// Either way, if the upper bound fits the budget every value
/// fits — return `indices.len()` with no per-value work. This
/// covers the overwhelmingly common "small values" case for both
/// non-null *and* nullable columns.
/// 2. Otherwise the chunk is genuinely near the budget: walk per-index
/// lengths from the offsets buffer directly (no slice/UTF-8
/// construction) and stop at the first value that pushes the
/// cumulative sum past the budget.
fn count_within_budget_offsets<T: ByteArrayType>(
values: &GenericByteArray<T>,
indices: &[usize],
byte_budget: usize,
) -> usize {
if indices.is_empty() {
return 0;
}
let n = indices.len();
let first = indices[0];
let last = indices[n - 1];
let offsets = values.value_offsets();
let prefix_overhead = std::mem::size_of::<u32>();

// Stage 1: O(1) span upper bound. Skips Stage 2 in the common case —
// including nullable columns, whose `indices` are sparse. The earlier
// `last - first + 1 == n` contiguity gate forced every nullable chunk
// onto the O(n) Stage 2 walk even though the span check is valid for
// any sorted index set.
if last >= first {
let payload = (offsets[last + 1] - offsets[first]).as_usize();
if payload + n * prefix_overhead <= byte_budget {
return n;
}
}

// Stage 2: scan per-index lengths from the offsets buffer.
let mut cum: usize = 0;
for (i, idx) in indices.iter().enumerate() {
let len = (offsets[idx + 1] - offsets[*idx]).as_usize() + prefix_overhead;
cum = cum.saturating_add(len);
if cum > byte_budget {
return i + 1;
}
}
n
}

/// Computes the min and max for the provided array and indices
///
/// This is a free function so it can be used with `downcast_op!`
Expand Down
Loading
Loading