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-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 6b0d5c59 fix(arrow): read ragged CSV format-table rows permissively
(#975)
6b0d5c59 is described below
commit 6b0d5c599e7fa6f99a31f5917630db6e2b9ca2c0
Author: jackylee <[email protected]>
AuthorDate: Sun Oct 4 22:24:31 2026 +0800
fix(arrow): read ragged CSV format-table rows permissively (#975)
---
crates/paimon/src/arrow/format/text.rs | 56 +++++++--
crates/paimon/src/spec/schema.rs | 209 +++++++++++++++++++++++++++++++++
2 files changed, 255 insertions(+), 10 deletions(-)
diff --git a/crates/paimon/src/arrow/format/text.rs
b/crates/paimon/src/arrow/format/text.rs
index d46c5411..61f1ce0f 100644
--- a/crates/paimon/src/arrow/format/text.rs
+++ b/crates/paimon/src/arrow/format/text.rs
@@ -842,20 +842,23 @@ fn decode_text_chunk(
message: format!("Invalid UTF-8 in CSV file: {e}"),
source: Some(Box::new(e)),
})?;
- let row = if line.trim().is_empty() {
+ let mut row = if line.trim().is_empty() {
vec![None; schema.fields().len()]
} else {
parse_csv_line(line, options)?
};
- if row.len() != schema.fields().len() {
- return Err(Error::DataInvalid {
- message: format!(
- "CSV row has {} fields, expected {}",
- row.len(),
- schema.fields().len()
- ),
- source: None,
- });
+ // Read permissively, matching Java `CsvParser` under the
default
+ // `csv.mode=permissive`: a row with fewer fields than the
schema
+ // pads the missing trailing columns with null, and a row
with more
+ // fields drops the extras. Failing the whole scan on a
field-count
+ // mismatch rejected files Java reads fine — e.g. reading
a table
+ // after ADD COLUMN (old files carry one fewer column), or
any
+ // externally produced CSV that omits an optional trailing
column.
+ let field_count = schema.fields().len();
+ if row.len() < field_count {
+ row.resize(field_count, None);
+ } else if row.len() > field_count {
+ row.truncate(field_count);
}
Ok(row)
})
@@ -1483,4 +1486,37 @@ mod tests {
]
);
}
+
+ #[test]
+ fn csv_reads_ragged_rows_permissively() {
+ // A 3-column CSV format-table with a short row (missing the trailing
+ // column, e.g. an old file read after ADD COLUMN) and a long row (an
+ // extra trailing field). Java reads both under the default permissive
+ // mode; the reader must pad/truncate rather than fail the whole scan.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Utf8, true),
+ Field::new("b", DataType::Utf8, true),
+ Field::new("c", DataType::Utf8, true),
+ ]));
+ let options = TextOptions::new(TextKind::Csv,
&std::collections::HashMap::new()).unwrap();
+ let lines: Vec<Vec<u8>> = vec![b"x,y".to_vec(), b"p,q,r".to_vec(),
b"1,2,3,4".to_vec()];
+ let batches = decode_text_chunk(&lines, TextKind::Csv, &options,
&schema).unwrap();
+ let total: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(total, 3);
+ let c = batches[0]
+ .column(2)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .expect("utf8 column");
+ assert!(
+ c.is_null(0),
+ "short row must pad the trailing column with null"
+ );
+ assert_eq!(c.value(1), "r");
+ assert_eq!(
+ c.value(2),
+ "3",
+ "long row must drop the extra trailing field"
+ );
+ }
}
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 095096b2..d3d1913a 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -360,6 +360,34 @@ impl TableSchema {
let field =
DataField::new(id, name.to_string(),
data_type).with_description(comment);
insert_field_with_move(&mut fields, field,
column_move.as_ref(), full_name)?;
+ // A CSV format table reads existing files positionally (no
+ // per-column header mapping), so a new column placed
before an
+ // existing physical column shifts every later one: the
+ // permissive reader would pad/truncate against the wrong
+ // positions and silently mis-assign old rows. Partition
keys
+ // are excluded from the physical CSV layout (as Java's
+ // FormatReadBuilder does), so only non-partition fields
after
+ // the new column count. Appending after the last physical
+ // column is safe. Reject the shifting case, symmetric
with the
+ // DropColumn guard.
+ {
+ let core_options =
CoreOptions::new(&new_schema.options);
+ let pos = field_index(&fields,
name).unwrap_or(fields.len());
+ let shifts_physical = fields
+ .iter()
+ .skip(pos + 1)
+ .any(|f| !new_schema.partition_keys.iter().any(|k|
k == f.name()));
+ if core_options.is_format_table()
+ && core_options.file_format() == "csv"
+ && shifts_physical
+ {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Cannot add non-trailing column '{name}'
to a CSV format table: it shifts the physical column positions of existing
files"
+ ),
+ });
+ }
+ }
}
SchemaChange::RenameColumn {
field_names,
@@ -457,6 +485,32 @@ impl TableSchema {
});
}
}
+ // A CSV format table reads existing files positionally (no
+ // per-column header mapping), so dropping a column before
an
+ // existing physical column shifts every later one: the
+ // permissive reader would truncate the trailing field and
+ // mis-assign the rest, silently returning wrong values.
+ // Partition keys are excluded from the physical CSV
layout (as
+ // Java's FormatReadBuilder does), so only non-partition
fields
+ // after the dropped column count. Dropping the last
physical
+ // column is safe. Reject the shifting case.
+ {
+ let core_options =
CoreOptions::new(&new_schema.options);
+ let shifts_physical = fields
+ .iter()
+ .skip(idx + 1)
+ .any(|f| !new_schema.partition_keys.iter().any(|k|
k == f.name()));
+ if core_options.is_format_table()
+ && core_options.file_format() == "csv"
+ && shifts_physical
+ {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "Cannot drop non-trailing column '{name}'
of a CSV format table: it shifts the physical column positions of existing
files"
+ ),
+ });
+ }
+ }
if fields.len() == 1 {
return Err(crate::Error::Unsupported {
message: "Cannot drop all fields in
table".to_string(),
@@ -5608,6 +5662,161 @@ mod tests {
);
}
+ #[test]
+ fn test_csv_format_table_rejects_dropping_non_trailing_column() {
+ // A CSV format table reads files positionally; dropping a non-trailing
+ // column shifts physical positions and would make the permissive
reader
+ // silently mis-decode old rows. The trailing column is safe to drop.
+ let build = || {
+ TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("label",
DataType::VarChar(VarCharType::string_type()))
+ .column("extra",
DataType::VarChar(VarCharType::string_type()))
+ .option("type", "format-table")
+ .option("file.format", "csv")
+ .build()
+ .unwrap(),
+ )
+ };
+
+ let err = build()
+ .apply_changes(vec![crate::spec::SchemaChange::drop_column(
+ "label".to_string(),
+ )])
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("physical column positions") &&
message.contains("label")),
+ "dropping a non-trailing CSV column should be rejected, got
{err:?}"
+ );
+
+ // Dropping the trailing column keeps positions aligned and is allowed.
+ build()
+ .apply_changes(vec![crate::spec::SchemaChange::drop_column(
+ "extra".to_string(),
+ )])
+ .unwrap();
+ }
+
+ #[test]
+ fn test_csv_format_table_rejects_adding_non_trailing_column() {
+ // Symmetric with the drop guard: a CSV format table reads files
+ // positionally, so a new column that lands anywhere but last shifts
the
+ // physical positions of existing files. The permissive reader would
then
+ // mis-assign old rows (e.g. `(1,"old")` read as `(id, extra, label) =
+ // (1,"old",NULL)`). Appending the column at the end is safe.
+ let build = || {
+ TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("label",
DataType::VarChar(VarCharType::string_type()))
+ .option("type", "format-table")
+ .option("file.format", "csv")
+ .build()
+ .unwrap(),
+ )
+ };
+ let varchar = || DataType::VarChar(VarCharType::string_type());
+
+ // Inserting after a non-last column, or first, shifts existing
positions.
+ for column_move in [
+ crate::spec::ColumnMove::move_after("extra".to_string(),
"id".to_string()),
+ crate::spec::ColumnMove::move_first("extra".to_string()),
+ ] {
+ let err = build()
+ .apply_changes(vec![
+
crate::spec::SchemaChange::add_column_with_description_and_column_move(
+ "extra".to_string(),
+ varchar(),
+ String::new(),
+ column_move,
+ ),
+ ])
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("physical column positions") &&
message.contains("extra")),
+ "adding a non-trailing CSV column should be rejected, got
{err:?}"
+ );
+ }
+
+ // Appending the new column (plain add, or an explicit move to last)
keeps
+ // existing positions, so old files just pad it with null.
+ build()
+ .apply_changes(vec![crate::spec::SchemaChange::add_column(
+ "extra".to_string(),
+ varchar(),
+ )])
+ .unwrap();
+ build()
+ .apply_changes(vec![
+
crate::spec::SchemaChange::add_column_with_description_and_column_move(
+ "extra".to_string(),
+ varchar(),
+ String::new(),
+ crate::spec::ColumnMove::move_last("extra".to_string()),
+ ),
+ ])
+ .unwrap();
+ }
+
+ #[test]
+ fn test_csv_format_table_partition_columns_excluded_from_physical_shift() {
+ // Partition columns are not part of the physical CSV, so the physical
+ // layout of `(id, label, pt)` partitioned by `pt` is `(id, label)`.
+ // Dropping `label` (last physical) and appending `extra` after it are
+ // therefore safe trailing changes and must be accepted.
+ let build = || {
+ TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("label",
DataType::VarChar(VarCharType::string_type()))
+ .column("pt",
DataType::VarChar(VarCharType::string_type()))
+ .partition_keys(vec!["pt".to_string()])
+ .option("type", "format-table")
+ .option("file.format", "csv")
+ .build()
+ .unwrap(),
+ )
+ };
+ let varchar = || DataType::VarChar(VarCharType::string_type());
+
+ // Dropping the last physical column (pt is partition-excluded) is
safe.
+ build()
+ .apply_changes(vec![crate::spec::SchemaChange::drop_column(
+ "label".to_string(),
+ )])
+ .unwrap();
+
+ // Adding after the last physical column (before only `pt`) is safe.
+ build()
+ .apply_changes(vec![
+
crate::spec::SchemaChange::add_column_with_description_and_column_move(
+ "extra".to_string(),
+ varchar(),
+ String::new(),
+ crate::spec::ColumnMove::move_after("extra".to_string(),
"label".to_string()),
+ ),
+ ])
+ .unwrap();
+
+ // A genuine physical shift (dropping `id`, before `label`) is still
rejected.
+ let err = build()
+ .apply_changes(vec![crate::spec::SchemaChange::drop_column(
+ "id".to_string(),
+ )])
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("physical column positions")),
+ "{err:?}"
+ );
+ }
+
#[test]
fn test_bucket_key_history_rejects_destructive_column_changes() {
let table_schema = TableSchema::new(
