This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch feat/schema-evolution-read
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/feat/schema-evolution-read by
this push:
new d9c4793 fix
d9c4793 is described below
commit d9c4793f9d5b79edc3891a5e7d4430fe5a85fbdc
Author: JingsongLi <[email protected]>
AuthorDate: Fri Apr 3 12:34:45 2026 +0800
fix
---
crates/integration_tests/tests/read_tables.rs | 42 +++++++
crates/paimon/src/arrow/mod.rs | 2 +-
crates/paimon/src/arrow/reader.rs | 161 +++++++++++++++-----------
crates/paimon/src/table/schema_manager.rs | 52 ++++++---
dev/spark/provision.py | 28 +++++
5 files changed, 203 insertions(+), 82 deletions(-)
diff --git a/crates/integration_tests/tests/read_tables.rs
b/crates/integration_tests/tests/read_tables.rs
index 8a4ad8f..aa4442c 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -1124,3 +1124,45 @@ async fn test_read_data_evolution_type_promotion() {
"Data evolution + type promotion: INT should be cast to BIGINT, MERGE
INTO updates value"
);
}
+
+/// Test reading a table after ALTER TABLE DROP COLUMN.
+/// Old data files have the dropped column; reader should ignore it.
+#[tokio::test]
+async fn test_read_schema_evolution_drop_column() {
+ let (_, batches) =
scan_and_read_with_fs_catalog("schema_evolution_drop_column", None).await;
+
+ // Verify the dropped column 'score' is not present in the output.
+ for batch in &batches {
+ assert!(
+ batch.column_by_name("score").is_none(),
+ "Dropped column 'score' should not appear in output"
+ );
+ }
+
+ let mut rows: Vec<(i32, String)> = Vec::new();
+ for batch in &batches {
+ let id = batch
+ .column_by_name("id")
+ .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+ .expect("id");
+ let name = batch
+ .column_by_name("name")
+ .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+ .expect("name");
+ for i in 0..batch.num_rows() {
+ rows.push((id.value(i), name.value(i).to_string()));
+ }
+ }
+ rows.sort_by_key(|(id, _)| *id);
+
+ assert_eq!(
+ rows,
+ vec![
+ (1, "alice".into()),
+ (2, "bob".into()),
+ (3, "carol".into()),
+ (4, "dave".into()),
+ ],
+ "Old rows should be readable after DROP COLUMN, with only remaining
columns"
+ );
+}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index a46c779..bd68653 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -16,7 +16,7 @@
// under the License.
mod reader;
-pub mod schema_evolution;
+pub(crate) mod schema_evolution;
pub use crate::arrow::reader::ArrowReaderBuilder;
diff --git a/crates/paimon/src/arrow/reader.rs
b/crates/paimon/src/arrow/reader.rs
index 561f969..5387093 100644
--- a/crates/paimon/src/arrow/reader.rs
+++ b/crates/paimon/src/arrow/reader.rs
@@ -101,7 +101,7 @@ impl ArrowReader {
let batch_size = self.batch_size;
let splits: Vec<DataSplit> = data_splits.to_vec();
let read_type = self.read_type;
- let mut schema_manager = self.schema_manager;
+ let schema_manager = self.schema_manager;
let table_schema_id = self.table_schema_id;
Ok(try_stream! {
for split in splits {
@@ -171,13 +171,8 @@ impl ArrowReader {
let batch_size = self.batch_size;
let splits: Vec<DataSplit> = data_splits.to_vec();
let read_type = self.read_type;
- let table_field_names: Vec<String> =
- table_fields.iter().map(|f| f.name().to_string()).collect();
- let projected_column_names: Vec<String> = read_type
- .iter()
- .map(|field| field.name().to_string())
- .collect();
- let mut schema_manager = self.schema_manager;
+ let table_fields: Vec<DataField> = table_fields.to_vec();
+ let schema_manager = self.schema_manager;
let table_schema_id = self.table_schema_id;
Ok(try_stream! {
@@ -205,8 +200,7 @@ impl ArrowReader {
&file_io,
&split,
&read_type,
- &projected_column_names,
- &table_field_names,
+ &table_fields,
schema_manager.clone(),
table_schema_id,
batch_size,
@@ -265,9 +259,11 @@ fn read_single_file_stream(
match mapping {
Some(ref idx_map) => {
// Only read data fields that are referenced by the index
mapping.
+ // Dedup by data field index to avoid duplicate parquet column
projections.
+ let mut seen = std::collections::HashSet::new();
let fields_to_read: Vec<DataField> = idx_map
.iter()
- .filter(|&&idx| idx != NULL_FIELD_INDEX)
+ .filter(|&&idx| idx != NULL_FIELD_INDEX &&
seen.insert(idx))
.map(|&idx| df[idx as usize].clone())
.collect();
(fields_to_read, Some(idx_map.clone()))
@@ -391,7 +387,16 @@ fn read_single_file_stream(
}
}
- let result = RecordBatch::try_new(target_schema.clone(),
columns).map_err(|e| {
+ let result = if columns.is_empty() {
+ RecordBatch::try_new_with_options(
+ target_schema.clone(),
+ columns,
+
&arrow_array::RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ } else {
+ RecordBatch::try_new(target_schema.clone(), columns)
+ }
+ .map_err(|e| {
Error::UnexpectedError {
message: format!("Failed to build schema-evolved
RecordBatch: {e}"),
source: Some(Box::new(e)),
@@ -405,6 +410,9 @@ fn read_single_file_stream(
/// Merge multiple files column-wise for data evolution, streaming with
bounded memory.
///
+/// Uses field IDs (not column names) to resolve which file provides which
column,
+/// ensuring correctness across schema evolution (column rename, add, drop).
+///
/// Opens all file readers simultaneously and maintains a cursor (current
batch + offset)
/// per file. Each poll slices up to `batch_size` rows from each file's
current batch,
/// assembles columns from the winning files, and yields the merged batch.
When a file's
@@ -413,9 +421,8 @@ fn merge_files_by_columns(
file_io: &FileIO,
split: &DataSplit,
read_type: &[DataField],
- projected_column_names: &[String],
- table_field_names: &[String],
- mut schema_manager: SchemaManager,
+ table_fields: &[DataField],
+ schema_manager: SchemaManager,
table_schema_id: i64,
batch_size: Option<usize>,
) -> crate::Result<ArrowRecordBatchStream> {
@@ -429,28 +436,54 @@ fn merge_files_by_columns(
let split = split.clone();
let data_files: Vec<DataFileMeta> = data_files.to_vec();
let read_type = read_type.to_vec();
- let projected_column_names = projected_column_names.to_vec();
- let table_field_names = table_field_names.to_vec();
+ let table_fields = table_fields.to_vec();
let output_batch_size = batch_size.unwrap_or(1024);
+ let target_schema = build_target_arrow_schema(&read_type)?;
Ok(try_stream! {
- // Determine which columns each file provides and resolve conflicts by
max_sequence_number.
- // column_name -> (file_index, max_sequence_number)
- let mut column_source: HashMap<String, (usize, i64)> = HashMap::new();
+ // Pre-load schemas and collect field IDs + data_fields per file.
+ // file_idx -> (field_ids, Option<Vec<DataField>>)
+ let mut file_info: HashMap<usize, (Vec<i32>, Option<Vec<DataField>>)>
= HashMap::new();
for (file_idx, file_meta) in data_files.iter().enumerate() {
- let file_columns: Vec<String> = if let Some(ref wc) =
file_meta.write_cols {
- wc.clone()
- } else if file_meta.schema_id != table_schema_id {
+ let (field_ids, data_fields) = if file_meta.schema_id !=
table_schema_id {
let file_schema =
schema_manager.schema(file_meta.schema_id).await?;
- file_schema.fields().iter().map(|f|
f.name().to_string()).collect()
+ let file_fields = file_schema.fields();
+
+ let ids: Vec<i32> = if let Some(ref wc) = file_meta.write_cols
{
+ // write_cols names are from the file's schema at write
time.
+ wc.iter()
+ .filter_map(|name| file_fields.iter().find(|f|
f.name() == name).map(|f| f.id()))
+ .collect()
+ } else {
+ file_fields.iter().map(|f| f.id()).collect()
+ };
+
+ (ids, Some(file_fields.to_vec()))
} else {
- table_field_names.clone()
+ let ids: Vec<i32> = if let Some(ref wc) = file_meta.write_cols
{
+ // write_cols names are from the current table schema.
+ wc.iter()
+ .filter_map(|name| table_fields.iter().find(|f|
f.name() == name).map(|f| f.id()))
+ .collect()
+ } else {
+ table_fields.iter().map(|f| f.id()).collect()
+ };
+
+ (ids, None)
};
- for col in &file_columns {
- let entry = column_source
- .entry(col.clone())
+ file_info.insert(file_idx, (field_ids, data_fields));
+ }
+
+ // Determine which file provides each field ID, resolving conflicts by
max_sequence_number.
+ // field_id -> (file_index, max_sequence_number)
+ let mut field_id_source: HashMap<i32, (usize, i64)> = HashMap::new();
+ for (file_idx, file_meta) in data_files.iter().enumerate() {
+ let (ref field_ids, _) = file_info[&file_idx];
+ for &fid in field_ids {
+ let entry = field_id_source
+ .entry(fid)
.or_insert((file_idx, i64::MIN));
if file_meta.max_sequence_number > entry.1 {
*entry = (file_idx, file_meta.max_sequence_number);
@@ -458,24 +491,24 @@ fn merge_files_by_columns(
}
}
- // For each file, determine which projected columns to read from it.
- // file_index -> Vec<column_name>
+ // For each projected field, determine which file provides it (by
field ID).
+ // file_index -> Vec<column_name> (target column names)
let mut file_read_columns: HashMap<usize, Vec<String>> =
HashMap::new();
- for col_name in &projected_column_names {
- if let Some(&(file_idx, _)) = column_source.get(col_name) {
+ for field in &read_type {
+ if let Some(&(file_idx, _)) = field_id_source.get(&field.id()) {
file_read_columns
.entry(file_idx)
.or_default()
- .push(col_name.clone());
+ .push(field.name().to_string());
}
}
- // For each projected column, record (file_index, column_name) for
assembly.
- let column_plan: Vec<(Option<usize>, String)> = projected_column_names
+ // For each projected field, record (file_index, target_column_name)
for assembly.
+ let column_plan: Vec<(Option<usize>, String)> = read_type
.iter()
- .map(|col_name| {
- let file_idx = column_source.get(col_name).map(|&(idx, _)|
idx);
- (file_idx, col_name.clone())
+ .map(|field| {
+ let file_idx = field_id_source.get(&field.id()).map(|&(idx,
_)| idx);
+ (file_idx, field.name().to_string())
})
.collect();
@@ -492,20 +525,14 @@ fn merge_files_by_columns(
.filter_map(|col_name| read_type.iter().find(|f| f.name() ==
col_name).cloned())
.collect();
- let file_meta = &data_files[file_idx];
- let data_fields: Option<Vec<DataField>> = if file_meta.schema_id
!= table_schema_id {
- let data_schema =
schema_manager.schema(file_meta.schema_id).await?;
- Some(data_schema.fields().to_vec())
- } else {
- None
- };
+ let (_, ref data_fields) = file_info[&file_idx];
let stream = read_single_file_stream(
file_io.clone(),
split.clone(),
data_files[file_idx].clone(),
file_read_type,
- data_fields,
+ data_fields.clone(),
batch_size,
None,
)?;
@@ -559,20 +586,25 @@ fn merge_files_by_columns(
let rows_to_emit = remaining.min(output_batch_size);
// Slice each file's current batch and assemble columns.
+ // Use the target schema so that missing columns are null-filled.
let mut columns: Vec<Arc<dyn arrow_array::Array>> =
Vec::with_capacity(column_plan.len());
- let mut schema_fields: Vec<ArrowField> =
Vec::with_capacity(column_plan.len());
-
- for (file_idx_opt, col_name) in &column_plan {
- if let Some(file_idx) = file_idx_opt {
- if let Some((batch, offset)) = file_cursors.get(file_idx) {
- if let Ok(col_idx) = batch.schema().index_of(col_name)
{
- let col = batch.column(col_idx).slice(*offset,
rows_to_emit);
- columns.push(col);
-
schema_fields.push(batch.schema().field(col_idx).clone());
- }
- }
- }
+
+ for (i, (file_idx_opt, col_name)) in
column_plan.iter().enumerate() {
+ let target_field = &target_schema.fields()[i];
+ let col = file_idx_opt
+ .and_then(|file_idx| file_cursors.get(&file_idx))
+ .and_then(|(batch, offset)| {
+ batch
+ .schema()
+ .index_of(col_name)
+ .ok()
+ .map(|col_idx|
batch.column(col_idx).slice(*offset, rows_to_emit))
+ });
+
+ columns.push(col.unwrap_or_else(|| {
+ arrow_array::new_null_array(target_field.data_type(),
rows_to_emit)
+ }));
}
// Advance all cursors.
@@ -582,14 +614,11 @@ fn merge_files_by_columns(
}
}
- if !columns.is_empty() {
- let schema = Arc::new(ArrowSchema::new(schema_fields));
- let merged = RecordBatch::try_new(schema, columns).map_err(|e|
Error::UnexpectedError {
- message: format!("Failed to build merged RecordBatch:
{e}"),
- source: Some(Box::new(e)),
- })?;
- yield merged;
- }
+ let merged = RecordBatch::try_new(target_schema.clone(),
columns).map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build merged RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ yield merged;
}
}
.boxed())
diff --git a/crates/paimon/src/table/schema_manager.rs
b/crates/paimon/src/table/schema_manager.rs
index 7b937e0..057dc3f 100644
--- a/crates/paimon/src/table/schema_manager.rs
+++ b/crates/paimon/src/table/schema_manager.rs
@@ -22,6 +22,7 @@
use crate::io::FileIO;
use crate::spec::TableSchema;
use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
const SCHEMA_DIR: &str = "schema";
const SCHEMA_PREFIX: &str = "schema-";
@@ -33,13 +34,16 @@ const SCHEMA_PREFIX: &str = "schema-";
/// is written with an incremented ID. Data files record which schema they
were written with
/// via `DataFileMeta.schema_id`.
///
+/// The schema cache is shared across clones via `Arc`, so multiple readers
+/// (e.g. parallel split streams) benefit from a single cache.
+///
/// Reference:
[org.apache.paimon.schema.SchemaManager](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java)
#[derive(Debug, Clone)]
pub struct SchemaManager {
file_io: FileIO,
table_path: String,
- /// Cache of loaded schemas by ID.
- cache: HashMap<i64, TableSchema>,
+ /// Shared cache of loaded schemas by ID.
+ cache: Arc<Mutex<HashMap<i64, Arc<TableSchema>>>>,
}
impl SchemaManager {
@@ -47,7 +51,7 @@ impl SchemaManager {
Self {
file_io,
table_path,
- cache: HashMap::new(),
+ cache: Arc::new(Mutex::new(HashMap::new())),
}
}
@@ -63,19 +67,37 @@ impl SchemaManager {
/// Load a schema by ID. Returns cached version if available.
///
+ /// The cache is shared across all clones of this `SchemaManager`, so
loading
+ /// a schema in one stream makes it available to all other streams reading
+ /// from the same table.
+ ///
/// Reference:
[SchemaManager.schema(long)](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java)
- pub async fn schema(&mut self, schema_id: i64) ->
crate::Result<&TableSchema> {
- if !self.cache.contains_key(&schema_id) {
- let path = self.schema_path(schema_id);
- let input = self.file_io.new_input(&path)?;
- let bytes = input.read().await?;
- let schema: TableSchema =
- serde_json::from_slice(&bytes).map_err(|e|
crate::Error::DataInvalid {
- message: format!("Failed to parse schema file: {path}"),
- source: Some(Box::new(e)),
- })?;
- self.cache.insert(schema_id, schema);
+ pub async fn schema(&self, schema_id: i64) ->
crate::Result<Arc<TableSchema>> {
+ // Fast path: check cache under a short lock.
+ {
+ let cache = self.cache.lock().unwrap();
+ if let Some(schema) = cache.get(&schema_id) {
+ return Ok(schema.clone());
+ }
+ }
+
+ // Cache miss — load from file (no lock held during I/O).
+ let path = self.schema_path(schema_id);
+ let input = self.file_io.new_input(&path)?;
+ let bytes = input.read().await?;
+ let schema: TableSchema =
+ serde_json::from_slice(&bytes).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("Failed to parse schema file: {path}"),
+ source: Some(Box::new(e)),
+ })?;
+ let schema = Arc::new(schema);
+
+ // Insert into shared cache (short lock).
+ {
+ let mut cache = self.cache.lock().unwrap();
+ cache.entry(schema_id).or_insert_with(|| schema.clone());
}
- Ok(self.cache.get(&schema_id).unwrap())
+
+ Ok(schema)
}
}
diff --git a/dev/spark/provision.py b/dev/spark/provision.py
index 0277255..dc24199 100644
--- a/dev/spark/provision.py
+++ b/dev/spark/provision.py
@@ -401,6 +401,34 @@ def main():
)
spark.sql("DROP TABLE data_evolution_type_promotion_updates")
+ # ===== Schema Evolution: Drop Column =====
+ # Old files have (id, name, score); after ALTER TABLE DROP COLUMN, table
has (id, name).
+ # Reader should ignore the dropped column when reading old files.
+ spark.sql(
+ """
+ CREATE TABLE IF NOT EXISTS schema_evolution_drop_column (
+ id INT,
+ name STRING,
+ score INT
+ ) USING paimon
+ """
+ )
+ spark.sql(
+ """
+ INSERT INTO schema_evolution_drop_column VALUES
+ (1, 'alice', 100),
+ (2, 'bob', 200)
+ """
+ )
+ spark.sql("ALTER TABLE schema_evolution_drop_column DROP COLUMN score")
+ spark.sql(
+ """
+ INSERT INTO schema_evolution_drop_column VALUES
+ (3, 'carol'),
+ (4, 'dave')
+ """
+ )
+
if __name__ == "__main__":
main()
