fix
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 @@ "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 @@ 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 @@ 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 @@ &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 @@ 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 @@ } } - 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 @@ /// 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 @@ 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 @@ 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 @@ } } - // 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 @@ .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 @@ 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 @@ } } - 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 @@ /// 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 @@ Self { file_io, table_path, - cache: HashMap::new(), + cache: Arc::new(Mutex::new(HashMap::new())), } } @@ -63,19 +67,37 @@ /// 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()); + } } - Ok(self.cache.get(&schema_id).unwrap()) + + // 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(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 @@ ) 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()