This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11003-6038304119027e550189d3fb4de5e98d68431c63 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
commit 81392e3e9925e099b28a0895bf668ef868f3e5b0 Author: Marcelo Tesla <[email protected]> AuthorDate: Wed Oct 28 15:13:21 2026 +0000 fix(parquet): skip empty CDC page flushes in add_data_page (#11003) # Which issue does this PR close? - Closes #10929. # Rationale for this change Content-defined chunking forces a data page break after every chunk except the last. Writing the chunk can already have flushed that page, when the chunk hits `data_page_size_limit` or `data_page_row_count_limit` exactly at the boundary. The forced `add_data_page` then flushes with nothing buffered. For BOOLEAN under `PARQUET_2_0` (RLE), that panics: `RLE value encoder is not initialized`. Other encodings do not panic, but they write a data page with zero values. `should_add_data_page`, `dict_fallback`, and `flush_data_pages` already skip the empty case. `add_data_page` did not. # What changes are included in this PR? `GenericColumnWriter::add_data_page` returns `Ok(())` when `num_buffered_values == 0`. No public API change. # Are these changes tested? - BOOLEAN, 500k rows, `data_page_size_limit(1024)` and CDC 8KiB/16KiB: write succeeds, no empty data pages, roundtrip row count matches - INT32, 500k rows, `data_page_row_count_limit(128)` and the same CDC options: no empty data pages `cargo test -p parquet --lib -- cdc_boolean_small_pages_does_not_panic` and `cdc_int32_row_count_limit_does_not_emit_empty_pages`. `cargo clippy -p parquet --lib -- -D warnings`. # Are there any user-facing changes? BOOLEAN CDC writes that previously panicked now succeed. Empty data pages are no longer emitted for the same forced-break case on other encodings. # AI Disclosure Assisted draft of the empty-buffer guard and regression tests. The skip matches the existing guards on `should_add_data_page`, `dict_fallback`, and `flush_data_pages`. Reviewed and verified with the checks above. --------- Co-authored-by: Marcelo Tesla <[email protected]> --- parquet/src/column/chunker/cdc.rs | 69 ++++++++++++++++++++++++++++++++++++++- parquet/src/column/writer/mod.rs | 7 ++++ 2 files changed, 75 insertions(+), 1 deletion(-) diff --git a/parquet/src/column/chunker/cdc.rs b/parquet/src/column/chunker/cdc.rs index 420bdf4f47..684dfd68a6 100644 --- a/parquet/src/column/chunker/cdc.rs +++ b/parquet/src/column/chunker/cdc.rs @@ -754,7 +754,9 @@ mod arrow_tests { use crate::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; use crate::arrow::arrow_writer::ArrowWriter; use crate::column::writer::LevelDataRef; - use crate::file::properties::{CdcOptions, WriterProperties}; + use crate::file::properties::{ + CdcOptions, WriterProperties, WriterPropertiesBuilder, WriterVersion, + }; use crate::file::reader::{FileReader, SerializedFileReader}; // --- Constants matching C++ TestCDCSingleRowGroup --- @@ -1134,6 +1136,71 @@ mod arrow_tests { buf } + /// 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); + } + #[test] fn cdc_all_null_arrow_column_writes_data_pages() { let array = Arc::new(Int32Array::from(vec![None::<i32>; 4096])) as ArrayRef; diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs index bab0a1b91e..bff91faf28 100644 --- a/parquet/src/column/writer/mod.rs +++ b/parquet/src/column/writer/mod.rs @@ -1488,7 +1488,14 @@ 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. Skip empty pages to preserve + /// strictly increasing first_row_index values in the OffsetIndex. pub(crate) fn add_data_page(&mut self) -> Result<()> { + if self.page_metrics.num_buffered_values == 0 { + return Ok(()); + } + // Extract encoded values let values_data = self.encoder.flush_data_page()?;
