haohuaijin commented on code in PR #11003:
URL: https://github.com/apache/arrow-rs/pull/11003#discussion_r4280323354
##########
parquet/src/column/chunker/cdc.rs:
##########
@@ -1134,6 +1134,89 @@ mod arrow_tests {
buf
}
+ /// Chunk sizes that still force an empty CDC page flush.
+ fn empty_page_repro_cdc_options() -> CdcOptions {
+ CdcOptions {
+ min_chunk_size: 8 * 1024,
+ max_chunk_size: 16 * 1024,
+ norm_level: 0,
+ }
+ }
+
+ /// Write a single-column batch with CDC plus the page-size / row-count
+ /// limits that make a forced chunk boundary flush an already-empty page.
+ fn write_boolean_cdc_with_small_pages(num_rows: usize) -> Vec<u8> {
+ let values: Vec<bool> = (0..num_rows).map(|i| i % 7 == 0).collect();
+ let col = Arc::new(BooleanArray::from(values)) as ArrayRef;
+ let batch = RecordBatch::try_from_iter([("flag", col)]).unwrap();
+
+ let props = WriterProperties::builder()
+ .set_writer_version(WriterVersion::PARQUET_2_0)
+ .set_data_page_size_limit(1024)
+ .set_content_defined_chunking(Some(empty_page_repro_cdc_options()))
+ .build();
+
+ let mut buf = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut buf, batch.schema(),
Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ buf
+ }
+
+ fn count_data_pages(data: &[u8]) -> (usize, usize) {
+ let info = get_column_info(data, 0);
+ let total = info.iter().map(|c| c.page_lengths.len()).sum();
+ let empty = info
+ .iter()
+ .flat_map(|c| c.page_lengths.iter())
+ .filter(|&&n| n == 0)
+ .count();
+ (total, empty)
+ }
+
+ #[test]
+ #[cfg_attr(miri, ignore)] // large enough to hit CDC chunk boundaries
+ fn cdc_boolean_small_pages_does_not_panic() {
+ let data = write_boolean_cdc_with_small_pages(32_768);
+ let (total, empty) = count_data_pages(&data);
+ assert!(
+ total > 1,
+ "CDC should produce multiple data pages, got {total}"
+ );
+ assert_eq!(empty, 0, "forced CDC page breaks must not emit empty
pages");
+
+ let readback = read_batches(&data);
+ let rows: usize = readback.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(rows, 32_768);
+ }
+
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn cdc_int32_row_count_limit_does_not_emit_empty_pages() {
+ let values: Vec<i32> = (0..8_192).map(|i| i % 97).collect();
+ let col = Arc::new(Int32Array::from(values)) as ArrayRef;
+ let batch = RecordBatch::try_from_iter([("a", col)]).unwrap();
+
+ let props = WriterProperties::builder()
+ .set_writer_version(WriterVersion::PARQUET_2_0)
+ .set_dictionary_enabled(false)
+ .set_data_page_row_count_limit(128)
+ .set_content_defined_chunking(Some(empty_page_repro_cdc_options()))
+ .build();
+
+ let mut buf = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut buf, batch.schema(),
Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+
+ let (total, empty) = count_data_pages(&buf);
+ assert!(
+ total > 1,
+ "CDC should produce multiple data pages, got {total}"
+ );
+ assert_eq!(empty, 0, "forced CDC page breaks must not emit empty
pages");
+ }
+
Review Comment:
```suggestion
/// Write a batch with common CDC settings and test-specific writer
properties.
fn write_cdc_batch(batch: &RecordBatch, builder: WriterPropertiesBuilder) ->
Vec<u8> {
let props = builder
.set_writer_version(WriterVersion::PARQUET_2_0)
.set_content_defined_chunking(Some(CdcOptions {
min_chunk_size: 8 * 1024,
max_chunk_size: 16 * 1024,
norm_level: 0,
}))
.build();
let mut buf = Vec::new();
let mut writer = ArrowWriter::try_new(&mut buf, batch.schema(),
Some(props)).unwrap();
writer.write(batch).unwrap();
writer.close().unwrap();
buf
}
fn assert_multiple_nonempty_data_pages(data: &[u8]) {
let info = get_column_info(data, 0);
let total: usize = info.iter().map(|c| c.page_lengths.len()).sum();
let empty = info
.iter()
.flat_map(|c| c.page_lengths.iter())
.filter(|&&n| n == 0)
.count();
assert!(
total > 1,
"CDC should produce multiple data pages, got {total}"
);
assert_eq!(empty, 0, "forced CDC page breaks must not emit empty pages");
}
#[test]
#[cfg_attr(miri, ignore)] // large enough to hit CDC chunk boundaries
fn cdc_boolean_small_pages_does_not_panic() {
let num_rows = 32_768;
let values: Vec<bool> = (0..num_rows).map(|i| i % 7 == 0).collect();
let col = Arc::new(BooleanArray::from(values)) as ArrayRef;
let batch = RecordBatch::try_from_iter([("flag", col)]).unwrap();
let props = WriterProperties::builder().set_data_page_size_limit(1024);
let data = write_cdc_batch(&batch, props);
assert_multiple_nonempty_data_pages(&data);
let readback = read_batches(&data);
let rows: usize = readback.iter().map(|b| b.num_rows()).sum();
assert_eq!(rows, num_rows);
}
#[test]
#[cfg_attr(miri, ignore)] // large enough to hit CDC chunk boundaries
fn cdc_int32_row_count_limit_does_not_emit_empty_pages() {
let values: Vec<i32> = (0..8_192).map(|i| i % 97).collect();
let col = Arc::new(Int32Array::from(values)) as ArrayRef;
let batch = RecordBatch::try_from_iter([("a", col)]).unwrap();
let props = WriterProperties::builder()
.set_dictionary_enabled(false)
.set_data_page_row_count_limit(128);
let data = write_cdc_batch(&batch, props);
assert_multiple_nonempty_data_pages(&data);
}
```
small suggestion
##########
parquet/src/column/writer/mod.rs:
##########
@@ -1466,7 +1466,17 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
/// Adds data page.
/// Data page is either buffered in case of dictionary encoding or written
directly.
+ ///
+ /// A call with nothing buffered is a no-op. Content-defined chunking
forces a page
+ /// break at chunk boundaries, and writing the chunk can already have
flushed the
+ /// page (the chunk hits `data_page_size_limit` or
`data_page_row_count_limit`).
+ /// Flushing an empty BOOLEAN RLE encoder panics; other encodings would
emit a
+ /// zero-value data page.
Review Comment:
```suggestion
///
/// Skip empty pages to preserve strictly increasing first_row_index
values
/// in the OffsetIndex.
```
--
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]
