From 582380bf64bafca431d5f3d52d533314045b16ab Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Sat, 13 Jun 2026 23:34:29 +0800 Subject: [PATCH 1/2] Fix List readers failing on nullable outer rows --- qdp/qdp-core/src/readers/arrow_ipc.rs | 215 ++++++++++++++++++- qdp/qdp-core/src/readers/parquet.rs | 292 ++++++++++++++++++++++++-- 2 files changed, 474 insertions(+), 33 deletions(-) diff --git a/qdp/qdp-core/src/readers/arrow_ipc.rs b/qdp/qdp-core/src/readers/arrow_ipc.rs index 39d9d91480..94ea1faaa4 100644 --- a/qdp/qdp-core/src/readers/arrow_ipc.rs +++ b/qdp/qdp-core/src/readers/arrow_ipc.rs @@ -150,16 +150,12 @@ impl DataReader for ArrowIPCReader { MahoutError::Io("Failed to downcast to ListArray".to_string()) })?; + // Phase 1: find sample_size from non-null rows and validate consistency. for i in 0..list_array.len() { - let value_array = list_array.value(i); - let float_array = value_array - .as_any() - .downcast_ref::() - .ok_or_else(|| { - MahoutError::Io("List values must be Float64".to_string()) - })?; - - let current_size = float_array.len(); + if list_array.is_null(i) { + continue; + } + let current_size = list_array.value_length(i) as usize; if let Some(expected) = sample_size { if current_size != expected { @@ -180,10 +176,53 @@ impl DataReader for ArrowIPCReader { })?; all_data.reserve(new_capacity); } + } + // Phase 2: collect data, handling null outer rows per NullHandling policy. + if list_array.null_count() == 0 { + let values = list_array.values(); + let float_array = values + .as_any() + .downcast_ref::() + .ok_or_else(|| MahoutError::Io("Values must be Float64".to_string()))?; handle_float64_nulls(&mut all_data, float_array, self.null_handling)?; - - num_samples += 1; + num_samples += list_array.len(); + } else { + for i in 0..list_array.len() { + if list_array.is_null(i) { + match self.null_handling { + NullHandling::Reject => { + return Err(MahoutError::InvalidInput( + "Null outer row in List column. Use \ + NullHandling::FillZero to replace with zeros, \ + or clean the data at the source." + .to_string(), + )); + } + NullHandling::FillZero => { + if let Some(ss) = sample_size { + all_data.extend(std::iter::repeat_n(0.0_f64, ss)); + num_samples += 1; + } + // sample_size unknown: skip this null row without counting it. + } + } + } else { + let value_array = list_array.value(i); + let float_array = value_array + .as_any() + .downcast_ref::() + .ok_or_else(|| { + MahoutError::Io("List values must be Float64".to_string()) + })?; + handle_float64_nulls( + &mut all_data, + float_array, + self.null_handling, + )?; + num_samples += 1; + } + } } } @@ -202,3 +241,157 @@ impl DataReader for ArrowIPCReader { Ok((all_data, num_samples, sample_size)) } } + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{ArrayRef, Float64Builder, ListBuilder, RecordBatch}; + use arrow::datatypes::{DataType, Field, Schema}; + use arrow::ipc::writer::FileWriter as ArrowIpcFileWriter; + use std::fs; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + static TEST_FILE_COUNTER: AtomicUsize = AtomicUsize::new(0); + + struct TempTestFile { + path: std::path::PathBuf, + } + + impl TempTestFile { + fn new() -> Self { + let count = TEST_FILE_COUNTER.fetch_add(1, Ordering::SeqCst); + let path = std::env::temp_dir().join(format!( + "mahout_test_arrow_ipc_{}_{}.arrow", + std::process::id(), + count + )); + Self { path } + } + + fn path(&self) -> &std::path::Path { + &self.path + } + } + + impl Drop for TempTestFile { + fn drop(&mut self) { + let _ = fs::remove_file(&self.path); + } + } + + fn write_test_arrow_ipc(schema: Arc, arrays: Vec) -> TempTestFile { + let file = TempTestFile::new(); + let batch = RecordBatch::try_new(schema.clone(), arrays).unwrap(); + let os_file = fs::File::create(file.path()).unwrap(); + let mut writer = ArrowIpcFileWriter::try_new(os_file, &schema).unwrap(); + writer.write(&batch).unwrap(); + writer.finish().unwrap(); + file + } + + fn write_ipc_list_with_null_outer_middle() -> TempTestFile { + // [[1.0, 2.0], null, [3.0, 4.0]] + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.append(false); // null outer row + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + write_test_arrow_ipc(schema, vec![array]) + } + + fn write_ipc_list_with_null_outer_first() -> TempTestFile { + // [null, [1.0, 2.0], [3.0, 4.0]] — null row at position 0 + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.append(false); // null outer row at position 0 + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + write_test_arrow_ipc(schema, vec![array]) + } + + #[test] + fn test_arrow_ipc_reader_null_outer_row_middle_fill_zero() { + // [[1,2], null, [3,4]] with FillZero → [1,2, 0,0, 3,4] + let file = write_ipc_list_with_null_outer_middle(); + let mut reader = ArrowIPCReader::new(file.path(), NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + assert_eq!(data, vec![1.0, 2.0, 0.0, 0.0, 3.0, 4.0]); + assert_eq!(num_samples, 3); + assert_eq!(sample_size, 2); + } + + #[test] + fn test_arrow_ipc_reader_null_outer_row_middle_reject() { + // [[1,2], null, [3,4]] with Reject → error + let file = write_ipc_list_with_null_outer_middle(); + let mut reader = ArrowIPCReader::new(file.path(), NullHandling::Reject).unwrap(); + let result = reader.read_batch(); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("Null outer row")); + } + + #[test] + fn test_arrow_ipc_reader_null_outer_row_first_fill_zero() { + // [null, [1,2], [3,4]] — null at row 0 must not corrupt sample_size + let file = write_ipc_list_with_null_outer_first(); + let mut reader = ArrowIPCReader::new(file.path(), NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + assert_eq!(data, vec![0.0, 0.0, 1.0, 2.0, 3.0, 4.0]); + assert_eq!(num_samples, 3); + assert_eq!(sample_size, 2); + } + + #[test] + fn test_arrow_ipc_reader_cross_batch_all_null_first_fill_zero() { + // Batch 1: [null, null] — sample_size unknown + // Batch 2: [[1,2], [3,4]] — sample_size established here + // All-null leading batch must not corrupt num_samples. + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut b1 = ListBuilder::new(Float64Builder::new()); + b1.append(false); + b1.append(false); + let batch1 = + RecordBatch::try_new(schema.clone(), vec![Arc::new(b1.finish()) as ArrayRef]).unwrap(); + + let mut b2 = ListBuilder::new(Float64Builder::new()); + b2.values().append_slice(&[1.0, 2.0]); + b2.append(true); + b2.values().append_slice(&[3.0, 4.0]); + b2.append(true); + let batch2 = + RecordBatch::try_new(schema.clone(), vec![Arc::new(b2.finish()) as ArrayRef]).unwrap(); + + let file = TempTestFile::new(); + { + let os_file = fs::File::create(file.path()).unwrap(); + let mut writer = ArrowIpcFileWriter::try_new(os_file, &schema).unwrap(); + writer.write(&batch1).unwrap(); + writer.write(&batch2).unwrap(); + writer.finish().unwrap(); + } + + let mut reader = ArrowIPCReader::new(file.path(), NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + assert_eq!(data, vec![1.0, 2.0, 3.0, 4.0]); + assert_eq!(num_samples, 2); + assert_eq!(sample_size, 2); + } +} diff --git a/qdp/qdp-core/src/readers/parquet.rs b/qdp/qdp-core/src/readers/parquet.rs index 30f2deead7..7ead86f1fb 100644 --- a/qdp/qdp-core/src/readers/parquet.rs +++ b/qdp/qdp-core/src/readers/parquet.rs @@ -290,8 +290,12 @@ impl DataReader for ParquetReader { MahoutError::Io("Failed to downcast to ListArray".to_string()) })?; - // Validate all rows have a consistent sample size. + // Validate non-null rows have a consistent sample size. + // Null outer rows return value_length 0, so they must be skipped here. for i in 0..list_array.len() { + if list_array.is_null(i) { + continue; + } let row_len = list_array.value_length(i) as usize; if let Some(expected) = sample_size { if row_len != expected { @@ -306,15 +310,50 @@ impl DataReader for ParquetReader { } } - // Cast the entire flat buffer once (avoids N per-row allocations - // on cross-dtype reads) then extend all_data in one pass. - let flat = list_flat_values(list_array); - extend_floats::<::ArrowType>( - &mut all_data, - &*flat, - self.null_handling, - )?; - num_samples += list_array.len(); + if list_array.null_count() == 0 { + // Fast path: no null outer rows; use flat buffer. + let flat = list_flat_values(list_array); + extend_floats::<::ArrowType>( + &mut all_data, + &*flat, + self.null_handling, + )?; + num_samples += list_array.len(); + } else { + // Null outer rows present; handle per NullHandling policy. + // If sample_size is still unknown (every row in this batch is null), + // FillZero cannot determine how many zeros to write — those null rows + // are skipped and not counted in num_samples. + for i in 0..list_array.len() { + if list_array.is_null(i) { + match self.null_handling { + NullHandling::Reject => { + return Err(MahoutError::InvalidInput( + "Null outer row in List column. Use \ + NullHandling::FillZero to replace with zeros, \ + or clean the data at the source." + .to_string(), + )); + } + NullHandling::FillZero => { + if let Some(ss) = sample_size { + all_data.extend(std::iter::repeat_n(T::default(), ss)); + num_samples += 1; + } + // sample_size unknown: skip this null row. + } + } + } else { + let row = list_array.value(i); + extend_floats::<::ArrowType>( + &mut all_data, + &*row, + self.null_handling, + )?; + num_samples += 1; + } + } + } } DataType::FixedSizeList(_, size) => { let list_array = column @@ -487,7 +526,7 @@ impl DataReader for ParquetStreamingReader { break; } all_data.extend_from_slice(&buffer[..written]); - num_samples += written / self.sample_size.unwrap_or(1); + num_samples += written / self.sample_size.unwrap_or(1).max(1); } let sample_size = self @@ -561,10 +600,25 @@ impl StreamingDataReader for ParquetStreamingReader { continue; } - let current_sample_size = list_array.value_length(0) as usize; - - // Validate all rows in this batch have a consistent sample size. - for i in 1..list_array.len() { + // Find sample_size from the first non-null row. + // Null outer rows return value_length 0 and must be skipped. + let first_non_null = + (0..list_array.len()).find(|&i| !list_array.is_null(i)); + let current_sample_size = match first_non_null { + Some(i) => list_array.value_length(i) as usize, + None => match self.sample_size { + // All rows null but sample_size known from an earlier batch. + Some(ss) => ss, + // All rows null and sample_size unknown: skip batch. + None => continue, + }, + }; + + // Validate all non-null rows in this batch. + for i in 0..list_array.len() { + if list_array.is_null(i) { + continue; + } let row_len = list_array.value_length(i) as usize; if row_len != current_sample_size { return Err(MahoutError::InvalidInput(format!( @@ -574,13 +628,45 @@ impl StreamingDataReader for ParquetStreamingReader { } } - // Cast the entire flat buffer once (avoids N per-row allocations - // on cross-dtype reads). - let flat = list_flat_values(list_array); - let batch_values = collect_floats::<::ArrowType>( - &*flat, - self.null_handling, - )?; + let batch_values = if list_array.null_count() == 0 { + // Fast path: no null outer rows; use flat buffer. + let flat = list_flat_values(list_array); + collect_floats::<::ArrowType>( + &*flat, + self.null_handling, + )? + } else { + // Null outer rows present; handle per NullHandling policy. + let mut vals = Vec::new(); + for i in 0..list_array.len() { + if list_array.is_null(i) { + match self.null_handling { + NullHandling::Reject => { + return Err(MahoutError::InvalidInput( + "Null outer row in List column. Use \ + NullHandling::FillZero to replace with \ + zeros, or clean the data at the source." + .to_string(), + )); + } + NullHandling::FillZero => { + vals.extend(std::iter::repeat_n( + T::default(), + current_sample_size, + )); + } + } + } else { + let row = list_array.value(i); + extend_floats::<::ArrowType>( + &mut vals, + &*row, + self.null_handling, + )?; + } + } + vals + }; (current_sample_size, batch_values) } @@ -969,6 +1055,40 @@ mod tests { // --- NullHandling tests --- + fn write_list_parquet_with_null_outer_middle() -> TempTestFile { + // [[1.0, 2.0], null, [3.0, 4.0]] + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.append(false); // null outer row + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + write_test_parquet(schema, vec![array]) + } + + fn write_list_parquet_with_null_outer_first() -> TempTestFile { + // [null, [1.0, 2.0], [3.0, 4.0]] — null row at position 0 seeds sample_size + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.append(false); // null outer row at position 0 + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + write_test_parquet(schema, vec![array]) + } + fn write_list_parquet_with_nulls() -> TempTestFile { let item_field = Arc::new(Field::new("item", DataType::Float64, true)); let list_field = Field::new("data", DataType::List(item_field.clone()), true); @@ -1119,4 +1239,132 @@ mod tests { }; assert!(err_msg.contains("Expected List or List")); } + + // --- Null outer row tests --- + + #[test] + fn test_parquet_reader_null_outer_row_middle_fill_zero() { + // [[1,2], null, [3,4]] with FillZero → [1,2, 0,0, 3,4] + let file = write_list_parquet_with_null_outer_middle(); + let mut reader = + ParquetReader::::new(file.path(), None, NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + assert_eq!(data, vec![1.0, 2.0, 0.0, 0.0, 3.0, 4.0]); + assert_eq!(num_samples, 3); + assert_eq!(sample_size, 2); + } + + #[test] + fn test_parquet_reader_null_outer_row_middle_reject() { + // [[1,2], null, [3,4]] with Reject → error + let file = write_list_parquet_with_null_outer_middle(); + let mut reader = + ParquetReader::::new(file.path(), None, NullHandling::Reject).unwrap(); + let result = reader.read_batch(); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("Null outer row")); + } + + #[test] + fn test_parquet_reader_null_outer_row_first_fill_zero() { + // [null, [1,2], [3,4]] — null at row 0 must not corrupt sample_size + let file = write_list_parquet_with_null_outer_first(); + let mut reader = + ParquetReader::::new(file.path(), None, NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + assert_eq!(data, vec![0.0, 0.0, 1.0, 2.0, 3.0, 4.0]); + assert_eq!(num_samples, 3); + assert_eq!(sample_size, 2); + } + + #[test] + fn test_parquet_streaming_reader_null_outer_row_middle_fill_zero() { + // [[1,2], null, [3,4]] with FillZero → [1,2, 0,0, 3,4] + let file = write_list_parquet_with_null_outer_middle(); + let mut reader = + ParquetStreamingReader::::new(file.path(), None, NullHandling::FillZero).unwrap(); + let mut buffer = vec![0.0_f64; 16]; + let written = reader.read_chunk(&mut buffer).unwrap(); + assert_eq!(&buffer[..written], &[1.0, 2.0, 0.0, 0.0, 3.0, 4.0]); + } + + #[test] + fn test_parquet_streaming_reader_null_outer_row_middle_reject() { + // [[1,2], null, [3,4]] with Reject → error + let file = write_list_parquet_with_null_outer_middle(); + let mut reader = + ParquetStreamingReader::::new(file.path(), None, NullHandling::Reject).unwrap(); + let mut buffer = vec![0.0_f64; 16]; + let result = reader.read_chunk(&mut buffer); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("Null outer row")); + } + + #[test] + fn test_parquet_streaming_reader_null_outer_row_first_fill_zero() { + // [null, [1,2], [3,4]] — null at row 0 must not seed sample_size to 0 + let file = write_list_parquet_with_null_outer_first(); + let mut reader = + ParquetStreamingReader::::new(file.path(), None, NullHandling::FillZero).unwrap(); + let mut buffer = vec![0.0_f64; 16]; + let written = reader.read_chunk(&mut buffer).unwrap(); + assert_eq!(&buffer[..written], &[0.0, 0.0, 1.0, 2.0, 3.0, 4.0]); + } + + #[test] + fn test_parquet_reader_cross_batch_all_null_first_fill_zero() { + // [null, null, [1,2], [3,4]] read with batch_size=2: + // batch 1 = [null, null] — sample_size unknown; must be skipped, not counted + // batch 2 = [[1,2], [3,4]] — sample_size established here + // Verifies that num_samples is not corrupted by the all-null leading batch. + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.append(false); // null row 0 + builder.append(false); // null row 1 + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + let file = write_test_parquet(schema, vec![array]); + // batch_size=2 splits into two batches: [null,null] then [[1,2],[3,4]]. + let mut reader = + ParquetReader::::new(file.path(), Some(2), NullHandling::FillZero).unwrap(); + let (data, num_samples, sample_size) = reader.read_batch().unwrap(); + // All-null first batch is skipped (sample_size unknown → no zeros, no count). + assert_eq!(data, vec![1.0, 2.0, 3.0, 4.0]); + assert_eq!(num_samples, 2); + assert_eq!(sample_size, 2); + } + + #[test] + fn test_parquet_streaming_reader_cross_batch_all_null_first_fill_zero() { + // [null, null, [1,2], [3,4]] with batch_size=2, FillZero: + // batch 1 = [null, null] — sample_size unknown; skipped + // batch 2 = [[1,2], [3,4]] — sample_size established; data written + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.append(false); + builder.append(false); + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + let file = write_test_parquet(schema, vec![array]); + let mut reader = + ParquetStreamingReader::::new(file.path(), Some(2), NullHandling::FillZero) + .unwrap(); + let mut buffer = vec![0.0_f64; 16]; + let written = reader.read_chunk(&mut buffer).unwrap(); + assert_eq!(&buffer[..written], &[1.0, 2.0, 3.0, 4.0]); + } } From 71335c15d0d11e4a63d31a889feda7510527054f Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Mon, 15 Jun 2026 22:09:09 +0800 Subject: [PATCH 2/2] Fix ParquetStreamingReader rejecting all-null leading batches --- qdp/qdp-core/src/readers/parquet.rs | 40 +++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/qdp/qdp-core/src/readers/parquet.rs b/qdp/qdp-core/src/readers/parquet.rs index 7ead86f1fb..ac78722138 100644 --- a/qdp/qdp-core/src/readers/parquet.rs +++ b/qdp/qdp-core/src/readers/parquet.rs @@ -609,8 +609,18 @@ impl StreamingDataReader for ParquetStreamingReader { None => match self.sample_size { // All rows null but sample_size known from an earlier batch. Some(ss) => ss, - // All rows null and sample_size unknown: skip batch. - None => continue, + // All rows null and sample_size unknown. + None => { + if self.null_handling == NullHandling::Reject { + return Err(MahoutError::InvalidInput( + "Null outer row in List column. Use \ + NullHandling::FillZero to replace with \ + zeros, or clean the data at the source." + .to_string(), + )); + } + continue; + } }, }; @@ -1341,6 +1351,32 @@ mod tests { assert_eq!(sample_size, 2); } + #[test] + fn test_parquet_streaming_reader_cross_batch_all_null_first_reject() { + // [null, null, [1,2], [3,4]] with batch_size=2, Reject: + // batch 1 = [null, null] — sample_size unknown; must error, not silently skip + let item_field = Arc::new(Field::new("item", DataType::Float64, true)); + let list_field = Field::new("data", DataType::List(item_field.clone()), true); + let schema = Arc::new(Schema::new(vec![list_field])); + + let mut builder = ListBuilder::new(Float64Builder::new()); + builder.append(false); + builder.append(false); + builder.values().append_slice(&[1.0, 2.0]); + builder.append(true); + builder.values().append_slice(&[3.0, 4.0]); + builder.append(true); + let array = Arc::new(builder.finish()) as ArrayRef; + + let file = write_test_parquet(schema, vec![array]); + let mut reader = + ParquetStreamingReader::::new(file.path(), Some(2), NullHandling::Reject).unwrap(); + let mut buffer = vec![0.0_f64; 16]; + let result = reader.read_chunk(&mut buffer); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("Null outer row")); + } + #[test] fn test_parquet_streaming_reader_cross_batch_all_null_first_fill_zero() { // [null, null, [1,2], [3,4]] with batch_size=2, FillZero: