etseidl commented on PR #11997: URL: https://github.com/apache/arrow-rs/pull/11997#issuecomment-6640481401
Thanks @adriangb. Looked good to me 😅, but my agent found a nit. ## Review: changes requested Reviewed head `cf99e7ad168dd9c97f1aa2848ddea799c04498f5` against base `b231cf15efa5dbbbe4ae8503849aa6ab3cfc4e0c`. ### P1 - Lookahead can disable page splitting for `ListView` strings **Location:** [`byte_budget_chunker.rs:239-244`](https://github.com/apache/arrow-rs/blob/cf99e7ad168dd9c97f1aa2848ddea799c04498f5/parquet/src/column/writer/byte_budget_chunker.rs#L239-L244) The larger gather window can contain reordered or repeated child indices from valid `ListView` arrays. However, the Arrow offset-array encoder's [fast estimate](https://github.com/apache/arrow-rs/blob/cf99e7ad168dd9c97f1aa2848ddea799c04498f5/parquet/src/arrow/arrow_writer/byte_array.rs#L1017-L1045) assumes increasing indices and bounds payload using only the first and last offsets. For a window such as `[0, 1, ..., 27, 0, ..., 0]`, that estimate ignores the large values in the middle and reports that everything fits. The new early return then skips necessary sub-batching. **Confirmed with a production `ArrowWriter` reproducer:** - Nullable `ListView<Utf8>`, PLAIN encoding, dictionary disabled. - First 1,024 rows: 28 singleton lists referencing children `0..27`, then 996 null lists. - Next 996 rows: singleton lists referencing child `0`. - Child `0` is one byte; the other 27 are 256 KiB. - Page budget: 16 KiB; write-batch size: 1,024. | Result | Base | PR head | |---|---:|---:| | Data pages | 8 | 2 | | Maximum page-body size | 1,048,604 bytes | **7,078,017 bytes** | Both files round-trip, but the PR packs all 27 large strings into one page. These are separate singleton records, not one indivisible oversized record. Scaling those 27 strings to 76 MiB each would produce 2,151,677,952 payload bytes, exceeding the format's signed 32-bit page-size limit. That large-allocation case was not executed. **Why this belongs to this PR:** the faulty span assumption predates the change, but the current chunk's indices are increasing and correctly sized on the base. Looking into later chunks introduces the backwards jump and creates this regression: - Base sizes `[0, 1, ..., 27]` and returns `2`, including the first overflowing value. - PR sizes `[0, 1, ..., 27, 0, ..., 0]` and incorrectly returns `1,024`. - The PR therefore skips splitting the current chunk, violating the newly documented prefix contract. **Suggested correction:** use the span shortcut only with a proven strictly increasing index sequence; fall back to cumulative sizing for reordered or repeated gathers. Add an Arrow-encoder regression with an increasing current-chunk prefix followed by reordered lookahead. The new tests use the generic slice encoder with identity indices and miss this case. <details> <summary>Reproducer</summary> Save this as `parquet/examples/listview_lookahead_repro.rs` and run it on each revision: ```bash # On base b231cf15efa5dbbbe4ae8503849aa6ab3cfc4e0c: cargo run -p parquet --no-default-features --features arrow \ --example listview_lookahead_repro -- base /tmp/listview-base.parquet # On PR head cf99e7ad168dd9c97f1aa2848ddea799c04498f5: cargo run -p parquet --no-default-features --features arrow \ --example listview_lookahead_repro -- head /tmp/listview-head.parquet ``` The same source was tested through external Cargo projects using path dependencies on the two checkouts. ```rust use std::{fs::File, sync::Arc}; use arrow_array::{Array, ListArray, ListViewArray, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; use parquet::{ arrow::{ArrowWriter, arrow_reader::ParquetRecordBatchReaderBuilder}, basic::{Compression, Encoding}, file::{ properties::WriterProperties, reader::{FileReader, SerializedFileReader}, }, }; fn main() -> Result<(), Box<dyn std::error::Error>> { let args: Vec<String> = std::env::args().collect(); let label = args.get(1).expect("revision label"); let output = args.get(2).expect("output parquet path"); const LARGE_BYTES: usize = 256 * 1024; const PAGE_LIMIT: usize = 16 * 1024; const BATCH_SIZE: usize = 1024; const ROWS: usize = 2020; // Child 0 is small; children 1..27 are large. let mut strings = vec!["s".to_owned()]; for idx in 1..28 { strings.push( char::from(b'A' + (idx - 1) as u8) .to_string() .repeat(LARGE_BYTES), ); } let children = Arc::new(StringArray::from(strings)); let child_field = Arc::new(Field::new_list_field(DataType::Utf8, false)); let schema = Arc::new(Schema::new(vec![Field::new( "lists", DataType::ListView(child_field.clone()), true, )])); // First chunk: children 0..27, then 996 null lists. // Later rows: 996 singleton lists referencing child 0. let offsets: Vec<i32> = (0..ROWS) .map(|row| if row < 28 { row as i32 } else { 0 }) .collect(); let validity: Vec<bool> = (0..ROWS) .map(|row| row < 28 || row >= BATCH_SIZE) .collect(); let sizes: Vec<i32> = validity.iter().map(|valid| i32::from(*valid)).collect(); let lists = ListViewArray::new( child_field, offsets.into(), sizes.into(), children.clone(), Some(validity.into()), ); assert_eq!(lists.null_count(), 996); let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(lists)])?; let properties = WriterProperties::builder() .set_dictionary_enabled(false) .set_encoding(Encoding::PLAIN) .set_compression(Compression::UNCOMPRESSED) .set_data_page_size_limit(PAGE_LIMIT) .set_write_batch_size(BATCH_SIZE) .build(); let mut writer = ArrowWriter::try_new(File::create(output)?, schema, Some(properties))?; writer.write(&batch)?; writer.close()?; println!( "revision={label} rows={ROWS} large_values=27 \ large_value_bytes={LARGE_BYTES} page_limit={PAGE_LIMIT} \ write_batch_size={BATCH_SIZE}" ); let file_reader = SerializedFileReader::new(File::open(output)?)?; assert_eq!(file_reader.num_row_groups(), 1); let row_group = file_reader.get_row_group(0)?; let mut pages = row_group.get_column_page_reader(0)?; let mut page_count = 0; let mut max_page_bytes = 0; while let Some(page) = pages.get_next_page()? { assert!(page.is_data_page(), "dictionary unexpectedly enabled"); page_count += 1; max_page_bytes = max_page_bytes.max(page.buffer().len()); println!( "page={page_count} level_values={} body_bytes={}", page.num_values(), page.buffer().len(), ); } println!( "summary data_pages={page_count} max_page_body_bytes={max_page_bytes}" ); // Check every row's validity, singleton-list length, and string contents. let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(output)?)? .with_batch_size(BATCH_SIZE) .build()?; let mut row = 0; for decoded in reader { let decoded = decoded?; let array = decoded.column(0); for local in 0..decoded.num_rows() { let expected_valid = row < 28 || row >= BATCH_SIZE; assert_eq!(!array.is_null(local), expected_valid, "row {row} validity"); if expected_valid { let child = if let Some(lists) = array.as_any().downcast_ref::<ListArray>() { lists.value(local) } else if let Some(lists) = array.as_any().downcast_ref::<ListViewArray>() { lists.value(local) } else { panic!("unexpected decoded type {:?}", array.data_type()); }; assert_eq!(child.len(), 1, "row {row} list length"); let string = child.as_any().downcast_ref::<StringArray>().unwrap(); let expected_idx = if row < 28 { row } else { 0 }; assert_eq!( string.value(0), children.value(expected_idx), "row {row} child value", ); } row += 1; } } assert_eq!(row, ROWS); println!("roundtrip=OK checked_rows={row}"); Ok(()) } ``` Observed page-body sizes: ```text Base: page=1 level_values=4 body_bytes=786461 page=2 level_values=4 body_bytes=1048604 page=3 level_values=4 body_bytes=1048604 page=4 level_values=4 body_bytes=1048604 page=5 level_values=4 body_bytes=1048604 page=6 level_values=4 body_bytes=1048604 page=7 level_values=4 body_bytes=1048604 page=8 level_values=1992 body_bytes=4997 summary data_pages=8 max_page_body_bytes=1048604 roundtrip=OK checked_rows=2020 PR head: page=1 level_values=1024 body_bytes=7078017 page=2 level_values=996 body_bytes=4994 summary data_pages=2 max_page_body_bytes=7078017 roundtrip=OK checked_rows=2020 ``` </details> -- 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]
