This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-mosaic.git
The following commit(s) were added to refs/heads/main by this push:
new cd041fd fix(core): validate writer batch schema (#70)
cd041fd is described below
commit cd041fd8548c4996895f36d98ef716a65050f4a8
Author: QuakeWang <[email protected]>
AuthorDate: Wed Sep 30 07:15:56 2026 +0800
fix(core): validate writer batch schema (#70)
---
core/src/writer.rs | 102 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 101 insertions(+), 1 deletion(-)
diff --git a/core/src/writer.rs b/core/src/writer.rs
index 6de041e..a977049 100644
--- a/core/src/writer.rs
+++ b/core/src/writer.rs
@@ -334,6 +334,7 @@ impl<S: OutputFile> MosaicWriter<S> {
),
));
}
+ self.validate_batch_schema(batch)?;
for (i, col) in self.schema.columns.iter().enumerate() {
let array = batch.column(self.batch_col_map[i]);
@@ -391,6 +392,25 @@ impl<S: OutputFile> MosaicWriter<S> {
Ok(())
}
+ fn validate_batch_schema(&self, batch: &RecordBatch) -> io::Result<()> {
+ let batch_schema = batch.schema();
+ for (i, col) in self.schema.columns.iter().enumerate() {
+ let batch_field = batch_schema.field(self.batch_col_map[i]);
+ if batch_field.name() != &col.name {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!(
+ "field name mismatch at column {}: schema has '{}' but
batch has '{}'",
+ self.batch_col_map[i],
+ col.name,
+ batch_field.name()
+ ),
+ ));
+ }
+ }
+ Ok(())
+ }
+
fn flush_row_group(&mut self) -> io::Result<()> {
if self.current_row_group_rows == 0 {
return Ok(());
@@ -1040,7 +1060,7 @@ mod tests {
let batch2 = RecordBatch::try_new(
Arc::new(Schema::new(vec![
- Field::new("id", DataType::Int32, false),
+ Field::new("id", DataType::Int32, true),
Field::new("name", DataType::Utf8, true),
])),
vec![
@@ -1051,6 +1071,86 @@ mod tests {
.unwrap();
let result = writer.write_batch(&batch2);
assert!(result.is_ok());
+
+ let batch3 = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(StringArray::from(vec![None, Some("world")])),
+ ],
+ )
+ .unwrap();
+ let result = writer.write_batch(&batch3);
+ assert!(result.is_ok());
+ }
+
+ #[test]
+ fn test_write_batch_rejects_reordered_schema() {
+ let arrow_schema = Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, false),
+ ]);
+ let out = MemOutputFile::new();
+ let mut writer = MosaicWriter::new(out, &arrow_schema,
WriterOptions::default()).unwrap();
+
+ let batch = RecordBatch::try_new(
+ Arc::new(arrow_schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int32Array::from(vec![10, 20])),
+ ],
+ )
+ .unwrap();
+ writer.write_batch(&batch).unwrap();
+
+ let reordered_schema = Schema::new(vec![
+ Field::new("b", DataType::Int32, false),
+ Field::new("a", DataType::Int32, false),
+ ]);
+ let reordered_batch = RecordBatch::try_new(
+ Arc::new(reordered_schema),
+ vec![
+ Arc::new(Int32Array::from(vec![30, 40])),
+ Arc::new(Int32Array::from(vec![3, 4])),
+ ],
+ )
+ .unwrap();
+
+ let err = writer.write_batch(&reordered_batch).unwrap_err();
+ assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
+ assert!(err.to_string().contains("field name mismatch"));
+ }
+
+ #[test]
+ fn test_write_batch_name_mismatch_reports_mapped_batch_index() {
+ let arrow_schema = Schema::new(vec![
+ Field::new("z", DataType::Int32, false),
+ Field::new("a", DataType::Int32, false),
+ ]);
+ let out = MemOutputFile::new();
+ let mut writer = MosaicWriter::new(out, &arrow_schema,
WriterOptions::default()).unwrap();
+
+ let mismatched_batch = RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("z", DataType::Int32, false),
+ Field::new("x", DataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int32Array::from(vec![3, 4])),
+ ],
+ )
+ .unwrap();
+
+ let err = writer.write_batch(&mismatched_batch).unwrap_err();
+ assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
+ assert_eq!(
+ err.to_string(),
+ "field name mismatch at column 1: schema has 'a' but batch has 'x'"
+ );
}
#[test]
