M-Tesla commented on code in PR #11003:
URL: https://github.com/apache/arrow-rs/pull/11003#discussion_r4395959082
##########
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:
Sorry I left this. Both tests now go through write_cdc_batch.
##########
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:
Sorry for the wait. The comment now keeps the empty pages out so
first_row_index stays strictly increasing in the OffsetIndex.
##########
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
Review Comment:
Sorry I missed this. The two tests finish in 0.02s without miri.
##########
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)]
Review Comment:
Sorry, the first note was thin. Both ignores now have a same-line reason:
the row count is large enough to hit CDC chunk boundaries.
##########
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:
Sorry I skipped this. The no-op line is in, next to the OffsetIndex note.
--
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]
