samvallad33 opened a new pull request, #10912:
URL: https://github.com/apache/arrow-rs/pull/10912
Draft, part of #7973. Single byte array columns read correctly now. Multi
column needs a structural decision that I want from you before I build it, and
separately, the regression tests already on main do not test anything.
## What overflows
`OffsetBuffer::try_push` fails at
`parquet/src/arrow/buffer/offset_buffer.rs:77-78` once the values written for
one batch pass `i32::MAX`. At the default batch size of 8192 that caps the
average value at 256KB. The file in the issue holds 5MB rows, so about 409 of
them fit.
`DELTA_LENGTH_BYTE_ARRAY` does not even get an error there. It panics, at
`parquet/src/arrow/array_reader/byte_array.rs:514`, because that path builds
offsets with `expect` instead of `try_push`.
## Why the three earlier attempts could not have worked
#9362, #9369 and #9504 all stop the byte array decoder early and return a
short count. That cannot work as a local change, for four separate reasons, all
on main today:
1. `parquet/src/column/reader.rs:292-298` turns a short values read into a
hard error, `insufficient values read from column`.
2. The definition levels for those values were already consumed at
`parquet/src/column/reader.rs:276`, before the values decoder runs at line 292.
By the time the decoder discovers the next value does not fit, the level stream
has moved past it and there is nothing to unwind it with.
3. `parquet/src/arrow/array_reader/mod.rs:195-227` treats any short read
from the record reader as an exhausted column chunk and calls `pages.next()`. A
short read that is not end of chunk therefore drops the rest of the chunk
silently.
4. `parquet/src/arrow/record_reader/mod.rs:147-162` loops while
`has_next()`. A decoder that returns 0 while the page still holds data spins
forever.
So the stop has to happen before the levels are read, not inside the values
decoder.
## What this PR does
Cap the batch before any levels are consumed.
- New `ColumnValueDecoder::values_capacity`, default `None`. It returns an
upper bound on the values that can still be decoded into the output buffer from
the current page, or `None` when the whole remainder is guaranteed to fit.
- `GenericColumnReader::read_records_with_reservation` calls it once per
data page, before decoding levels, and lowers the record budget. For a non
repeated column a record contributes at most one value, so capping records by a
value budget is always safe.
- A `stopped_for_capacity` flag distinguishes a capacity stop from an
exhausted chunk. `GenericRecordReader::read_records` stops instead of spinning,
and the free `read_records` in `array_reader/mod.rs` stops instead of advancing
the page iterator.
- `ByteArrayColumnValueDecoder` implements `values_capacity` for all four
byte array encodings.
Hot path cost. `values_capacity` is a defaulted method reached through a
generic type parameter, not a trait object, so for every decoder that does not
override it the call and the branch fold away at monomorphisation. For the byte
array decoder the common case is one comparison against a bound the decoder
already holds:
- PLAIN, bytes left in the page. Those include the 4 byte length prefixes,
so it is a strict over estimate.
- DELTA_LENGTH_BYTE_ARRAY, value bytes left in the page.
- DELTA_BYTE_ARRAY, values left times the longest value in the page. A delta
byte array value can be longer than the bytes it occupies because of the shared
prefix, so the page length is not a bound. The longest value is computed once
from the length arrays the decoder already materialises.
- RLE_DICTIONARY, keys left times the longest dictionary value, computed
once per dictionary page.
Only when that comparison says the remainder might not fit does anything
scan, and by then we are already in the multi gigabyte regime.
`LargeUtf8` and `LargeBinary` are untouched. `OffsetSizeTrait::IS_LARGE` is
a constant, so `values_capacity` reduces to `None` when monomorphised for `i64`.
## The regression tests on main do not reach the reader
`parquet/tests/arrow_reader/large_string_overflow.rs` from #9361 builds its
input with `BinaryBuilder`, 1024 values of 3MB each. That is 3GB, so
`append_value` panics on row 683 inside the test helper, at
`arrow-array/src/builder/generic_bytes_builder.rs:87`. The panic message is the
exact string the tests assert on, so all four pass, and they pass without a
parquet file ever being written.
```
$ RUST_BACKTRACE=1 cargo test -p parquet --features
arrow,async,test_common,experimental \
--test arrow_reader large_binary_plain_encoding_overflow -- --nocapture
--test-threads=1
running 1 test
test large_string_overflow::large_binary_plain_encoding_overflow - should
panic ...
thread '...' panicked at
arrow-array/src/builder/generic_bytes_builder.rs:87:57:
byte array offset overflow
stack backtrace:
5: <...GenericByteBuilder<GenericBinaryType<i32>>>::next_offset
at arrow-array/src/builder/generic_bytes_builder.rs:87:57
6:
<...GenericByteBuilder<GenericBinaryType<i32>>>::append_value::<&Vec<u8>>
at arrow-array/src/builder/generic_bytes_builder.rs:110:40
7: arrow_reader::large_string_overflow::make_large_binary_array
at ./tests/arrow_reader/large_string_overflow.rs:43:17
8:
arrow_reader::large_string_overflow::large_binary_plain_encoding_overflow
at ./tests/arrow_reader/large_string_overflow.rs:80:17
ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 124 filtered
out; finished in 0.33s
```
0.33s, and a stack that never reaches parquet. They would keep passing after
any fix, which is why they cannot serve as acceptance criteria.
This PR replaces them. The input is written in 100 row chunks so no single
input array is oversized, the row group is 700 values of 3MB for 2.1GB, and the
test reads the file back and checks every row survives in order across however
many batches the reader chooses. All four encodings run in one test function on
purpose: each holds a little over 2GB while decoding, and four separate tests
let the harness run them in parallel, which needs four times the memory.
Against main, with only the test file changed:
```
running 4 tests
test large_string_overflow::large_binary_delta_byte_array_encoding_overflow
... FAILED
test large_string_overflow::large_binary_delta_length_encoding_overflow ...
FAILED
test large_string_overflow::large_binary_plain_encoding_overflow ... FAILED
test large_string_overflow::large_binary_rle_dictionary_encoding_overflow
... FAILED
---- large_binary_plain_encoding_overflow stdout ----
called `Result::unwrap()` on an `Err` value:
ParquetError("Parquet error: index overflow decoding byte array")
---- large_binary_delta_length_encoding_overflow stdout ----
thread '...' panicked at parquet/src/arrow/array_reader/byte_array.rs:514:36:
index overflow decoding byte array
test result: FAILED. 0 passed; 4 failed; 0 ignored; 0 measured; 121 filtered
out; finished in 130.53s
```
With the source changes:
```
running 4 tests
test large_string_overflow::large_binary_delta_byte_array_encoding_overflow
... ok
test large_string_overflow::large_binary_delta_length_encoding_overflow ...
ok
test large_string_overflow::large_binary_plain_encoding_overflow ... ok
test large_string_overflow::large_binary_rle_dictionary_encoding_overflow
... ok
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 121 filtered
out; finished in 132.38s
```
That was with the four tests still separate. They are one function now,
which is what is in the branch.
## What works and what does not
Works:
- A byte array column read on its own, which is the `select length(html)
from evil.parquet` case in the issue. The reader emits shorter batches and
every row comes back in order.
- All four byte array encodings.
Not covered yet, and I made each of these fail with a message that names the
limit rather than let it return wrong data:
- More than one column in the projection. `StructArrayReader::read_records`
at `parquet/src/arrow/array_reader/struct_array.rs:72-90` requires every child
to return the same count. The stop point is only discovered part way through
the batch, after the other columns have already decoded past it, and a leaf
cannot be rewound. So `read_row_group` on the 8 column HuggingFace file still
fails.
- The predicate cache. `CachedArrayReader` maps row offsets to batches at
fixed multiples of `batch_size`,
`parquet/src/arrow/array_reader/cached_array_reader.rs:141`, so a short read
that is not end of column would make every later batch id point at the wrong
rows.
- Row filters that use the mask cursor, for the same reason: the mask is
built for the chunk before the read.
- Repeated columns, a list or map of strings. A record can hold many values
there, so a value budget does not convert into a record budget. The cap is
disabled when a repetition level decoder is present.
- `byte_array_dictionary.rs`, which produces a `DictionaryArray`, has its
own value decoder and is unchanged.
## The design question
Multi column is the part I want your call on, because it needs one of two
structural changes and I do not want to guess wrong a fourth time:
- Option A, the record reader learns to carry over.
`GenericRecordReader::consume_record_data` currently hands out the whole
buffer, `parquet/src/arrow/record_reader/mod.rs:205-209`. It would gain a split
so a leaf can emit N records and keep the rest, which means `ValuesBuffer`
gains a split too, and the definition and repetition level buffers with it.
- Option B, the struct reader carries over at the array level. Children read
as they do now, `StructArrayReader` emits the minimum length across children
and holds the tails as sliced `ArrayRef`s. A child that is holding a tail does
not read again until that tail drains, which keeps it to slicing only, never
`concat`, so a carried tail plus a fresh batch cannot overflow on its own.
B is the smaller change and touches no hot loop. A is more invasive but
keeps the accounting in one place and would also cover the predicate cache.
Tell me which you want and I will finish it here.
## About the benchmark result on #9369
The bot run on #9369 reported its largest regressions on
`FIXED_LEN_BYTE_ARRAY/Float16Array`, 1.57x and 1.53x on the plain encoded
cases. That number cannot be work. #9369 touched `byte_array.rs`,
`arrow_reader/mod.rs`, `offset_buffer.rs` and one test. The Float16 benchmark
builds its reader with `make_fixed_len_byte_array_reader` at
`parquet/benches/arrow_reader.rs:1037`, and `fixed_len_byte_array.rs` contains
no reference to any of that:
```
$ grep -n "offset_buffer\|OffsetBuffer\|byte_array::"
parquet/src/arrow/array_reader/fixed_len_byte_array.rs
NONE
```
A change that cannot execute on a code path cannot slow that path by 57
percent, so what the bot measured there was either run to run noise or code
layout and inlining drift.
## A noise floor, which nobody has published for this harness
I ran `--bench arrow_reader` filtered to `Float16Array|BinaryArray/plain`,
six times, serially, on an idle M1 Max. `main2` and `main3` are two runs of the
identical binary from `cbbb56b`, back to back, no rebuild in between. `branch2`
and `branch3` are two runs of this branch. `main1` and `branch1` were each the
first run after their build and I show `main1` separately below because of what
it does. Criterion defaults, times in microseconds.
```
benchmark main2
main3 branch2 branch3 main/main branch/main
BinaryArray/plain encoded, mandatory, no NULLs 862.1
881.1 874.6 880.3 2.2% 1.007x
BinaryArray/plain encoded, optional, half NULLs 748.4
769.2 809.5 786.5 2.8% 1.052x
BinaryArray/plain encoded, optional, no NULLs 903.4
852.8 885.2 884.3 5.9% 1.008x
Float16Array/byte_stream_split, mandatory, no NULLs 119.8
119.8 120.0 120.2 0.0% 1.002x
Float16Array/byte_stream_split, optional, half NULLs 214.0
203.3 203.9 203.5 5.2% 0.976x
Float16Array/byte_stream_split, optional, no NULLs 134.9
122.5 122.4 122.4 10.1% 0.951x
Float16Array/plain, mandatory, no NULLs 59.5
59.4 59.7 59.7 0.1% 1.004x
Float16Array/plain, optional, half NULLs 172.3
172.6 172.3 176.4 0.2% 1.011x
Float16Array/plain, optional, no NULLs 62.3
62.0 62.3 62.8 0.6% 1.006x
```
Two things fall out.
The same binary against itself reaches 10.1 percent apart on
`Float16Array/byte_stream_split encoded, optional, no NULLs`, and `main1` is
worse still: 212.3 microseconds on `Float16Array/byte_stream_split encoded,
mandatory, no NULLs` against 119.8 for `main2` and 119.8 for `main3`. That is
1.77x between two runs of bytes that are identical, on the same machine,
minutes apart. A single paired run cannot tell that apart from a real
regression.
But it does not let #9369 off, and I want to be straight about that. The two
benchmarks it was charged 1.57x and 1.53x for are `Float16Array/plain encoded,
mandatory, no NULLs` and the optional no NULLs variant, and on this hardware
those two are the steadiest in the whole set, 0.1 percent and 0.6 percent main
against main. So what the bot saw there was probably not sampling noise either.
Given that `fixed_len_byte_array.rs` cannot reach a single line #9369 touched,
the remaining explanation is code layout and inlining drift, and that is worth
knowing before a fourth PR gets closed over it.
For this branch on those same two benchmarks: 1.004x and 1.006x. On the
three `BinaryArray/plain` benchmarks, which are the path this branch actually
adds code to, 1.007x, 1.052x and 1.008x against a main to main spread of 2.2,
2.8 and 5.9 percent on the same three.
This is my hardware, one filtered subset, criterion defaults. It is not your
CI box and I am not claiming it replaces a run there. Say the word and I will
widen the filter or the repetitions. What I would suggest either way is that
the bot get a mode that runs a commit against itself and posts the spread,
because right now every performance objection in this area is being made
against an unmeasured baseline.
## Gates
On aarch64-apple-darwin, rust 1.97.1, with `parquet-testing` and `testing`
checked out.
```
$ cargo fmt --all -- --check # clean
$ cargo clippy -p parquet --features arrow,async,test_common,experimental \
--all-targets -- -D warnings # clean
$ cargo test -p parquet --features arrow,async,test_common,experimental
test result: ok. 1337 passed; 0 failed; 0 ignored; 0 measured; 0 filtered
out; finished in 12.74s
test result: ok. 121 passed; 0 failed; 1 ignored; 0 measured; 0 filtered
out; finished in 142.97s
test result: ok. 15 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out;
finished in 4.04s
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out;
finished in 0.01s
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out;
finished in 0.00s
test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out;
finished in 0.01s
test result: ok. 83 passed; 0 failed; 7 ignored; 0 measured; 0 filtered out;
finished in 23.73s
```
The 142.97s target is `arrow_reader`, which includes the 2.1GB overflow test.
--
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]
