JingsongLi commented on code in PR #681: URL: https://github.com/apache/paimon-rust/pull/681#discussion_r3818245609
########## crates/paimon/src/arrow/schema_evolution.rs: ########## @@ -19,12 +19,373 @@ //! //! Reference: [org.apache.paimon.schema.SchemaEvolutionUtil](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java) -use crate::spec::DataField; +use crate::arrow::paimon_type_to_arrow; +use crate::spec::{DataField, DataType}; +use arrow_array::builder::{BinaryBuilder, StringBuilder}; +use arrow_array::types::{ + ArrowPrimitiveType, Date32Type, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, + Int8Type, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType, +}; +use arrow_array::{Array, ArrayRef, BinaryArray, PrimitiveArray, StringArray}; +use arrow_cast::cast; use std::collections::HashMap; +use std::sync::Arc; /// Sentinel value indicating a field does not exist in the data schema. pub const NULL_FIELD_INDEX: i32 = -1; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SchemaEvolutionCast { + Identity, + NumericPrimitive, + Decimal, + DateToTimestamp, + TimestampToDate, + CharacterString, + BinaryString, +} + +/// Return whether a schema-evolution cast has a real executor. +/// +/// This is deliberately narrower than SQL explicit casting. Schema admission +/// uses this function in addition to the logical Paimon cast rules, so every +/// accepted type change can be executed while reading old files. +pub(crate) fn schema_evolution_cast_implemented(source: &DataType, target: &DataType) -> bool { + resolve_schema_evolution_cast(source, target).is_some() +} + +fn resolve_schema_evolution_cast( + source: &DataType, + target: &DataType, +) -> Option<SchemaEvolutionCast> { + if same_type_ignoring_nullability(source, target) { + return is_top_level_scalar(source).then_some(SchemaEvolutionCast::Identity); + } + + match (source, target) { + (DataType::Char(_) | DataType::VarChar(_), DataType::Char(_) | DataType::VarChar(_)) => { + Some(SchemaEvolutionCast::CharacterString) + } + ( + DataType::Binary(_) | DataType::VarBinary(_), + DataType::Binary(_) | DataType::VarBinary(_), + ) => Some(SchemaEvolutionCast::BinaryString), + (source, target) if is_numeric_primitive(source) && is_numeric_primitive(target) => { + Some(SchemaEvolutionCast::NumericPrimitive) + } + (source, DataType::Decimal(_)) + if is_integer_numeric(source) || matches!(source, DataType::Decimal(_)) => + { + Some(SchemaEvolutionCast::Decimal) + } + (DataType::Date(_), DataType::Timestamp(timestamp)) if timestamp.precision() <= 3 => { + Some(SchemaEvolutionCast::DateToTimestamp) + } + (DataType::Timestamp(_), DataType::Date(_)) => Some(SchemaEvolutionCast::TimestampToDate), + _ => None, + } +} + +fn is_top_level_scalar(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::Boolean(_) + | DataType::TinyInt(_) + | DataType::SmallInt(_) + | DataType::Int(_) + | DataType::BigInt(_) + | DataType::Decimal(_) + | DataType::Double(_) + | DataType::Float(_) + | DataType::Binary(_) + | DataType::VarBinary(_) + | DataType::Char(_) + | DataType::VarChar(_) + | DataType::Date(_) + | DataType::LocalZonedTimestamp(_) + | DataType::Time(_) + | DataType::Timestamp(_) + ) +} + +fn is_integer_numeric(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) | DataType::BigInt(_) + ) +} + +fn is_numeric_primitive(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::TinyInt(_) + | DataType::SmallInt(_) + | DataType::Int(_) + | DataType::BigInt(_) + | DataType::Double(_) + | DataType::Float(_) + ) +} + +pub(crate) fn same_type_ignoring_nullability(source: &DataType, target: &DataType) -> bool { + match ( + source.copy_with_nullable(true), + target.copy_with_nullable(true), + ) { + (Ok(source), Ok(target)) => source == target, + _ => false, + } +} + +/// Cast one physical Arrow column according to Paimon schema-evolution +/// semantics. Arrow is used only after the Paimon source/target pair has been +/// resolved to a supported executor. +pub(crate) fn cast_array_for_schema_evolution( + array: &ArrayRef, + source: &DataType, + target: &DataType, +) -> crate::Result<ArrayRef> { + // Unchanged complex fields can still appear in an old file selected by a + // different table schema version. They need no ALTER TYPE executor, but + // nested Arrow fields may still carry file metadata which must be removed + // before constructing a batch with the current logical schema. + if same_type_ignoring_nullability(source, target) { + let target_arrow_type = paimon_type_to_arrow(target)?; + if array.data_type() == &target_arrow_type { + return Ok(array.clone()); + } + return cast(array.as_ref(), &target_arrow_type).map_err(|error| { + crate::Error::UnexpectedError { + message: format!( + "Failed schema evolution Arrow normalization from {source:?} to {target:?}: {error}" + ), + source: Some(Box::new(error)), + } + }); + } + let executor = + resolve_schema_evolution_cast(source, target).ok_or_else(|| crate::Error::Unsupported { + message: format!( + "Schema evolution cast from {source:?} to {target:?} is not implemented" + ), + })?; + + match executor { + SchemaEvolutionCast::Identity => Ok(array.clone()), + SchemaEvolutionCast::NumericPrimitive => cast_numeric_primitive(array, source, target), + SchemaEvolutionCast::Decimal => { + let target_arrow_type = paimon_type_to_arrow(target)?; + cast(array.as_ref(), &target_arrow_type).map_err(|error| { + crate::Error::UnexpectedError { + message: format!( + "Failed schema evolution cast from {source:?} to {target:?}: {error}" + ), + source: Some(Box::new(error)), + } + }) + } + SchemaEvolutionCast::DateToTimestamp => cast_date_to_timestamp(array, target), + SchemaEvolutionCast::TimestampToDate => cast_timestamp_to_date(array, source), + SchemaEvolutionCast::CharacterString => cast_character_string(array, target), + SchemaEvolutionCast::BinaryString => cast_binary_string(array, target), + } +} + +fn downcast_primitive<'a, T: ArrowPrimitiveType>( + array: &'a ArrayRef, + semantic_type: &str, +) -> crate::Result<&'a PrimitiveArray<T>> { + array + .as_any() + .downcast_ref::<PrimitiveArray<T>>() + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "Expected {semantic_type} array with physical type {:?}, found {:?}", + T::DATA_TYPE, + array.data_type() + ), + source: None, + }) +} + +macro_rules! cast_numeric_primitive_array { + ($array:expr, $source:ty, $target:expr) => {{ + let values = downcast_primitive::<$source>($array, "numeric")?; + let result: ArrayRef = match $target { + DataType::TinyInt(_) => { + Arc::new(values.unary::<_, Int8Type>(|value| value as i32 as i8)) + } + DataType::SmallInt(_) => { + Arc::new(values.unary::<_, Int16Type>(|value| value as i32 as i16)) + } + DataType::Int(_) => Arc::new(values.unary::<_, Int32Type>(|value| value as i32)), + DataType::BigInt(_) => Arc::new(values.unary::<_, Int64Type>(|value| value as i64)), + DataType::Float(_) => Arc::new(values.unary::<_, Float32Type>(|value| value as f32)), + DataType::Double(_) => Arc::new(values.unary::<_, Float64Type>(|value| value as f64)), + _ => unreachable!("numeric primitive executor requires a primitive numeric target"), + }; + Ok(result) + }}; +} + +fn cast_numeric_primitive( + array: &ArrayRef, + source: &DataType, + target: &DataType, +) -> crate::Result<ArrayRef> { + // Java Number converts floating-point values to byte/short through int, + // combining saturating float-to-int conversion with wrapping integer + // narrowing. The macro uses that two-stage conversion for those targets. + match source { + DataType::TinyInt(_) => cast_numeric_primitive_array!(array, Int8Type, target), + DataType::SmallInt(_) => cast_numeric_primitive_array!(array, Int16Type, target), + DataType::Int(_) => cast_numeric_primitive_array!(array, Int32Type, target), + DataType::BigInt(_) => cast_numeric_primitive_array!(array, Int64Type, target), + DataType::Float(_) => cast_numeric_primitive_array!(array, Float32Type, target), + DataType::Double(_) => cast_numeric_primitive_array!(array, Float64Type, target), + _ => unreachable!("numeric primitive executor requires a primitive numeric source"), + } +} + +fn cast_date_to_timestamp(array: &ArrayRef, target: &DataType) -> crate::Result<ArrayRef> { + let values = downcast_primitive::<Date32Type>(array, "DATE")?; + let DataType::Timestamp(timestamp) = target else { + unreachable!("date executor requires a TIMESTAMP target") + }; + let result: ArrayRef = match timestamp.precision() { + 0..=3 => Arc::new( + values.unary::<_, TimestampMillisecondType>(|value| i64::from(value) * 86_400_000), + ), + _ => unreachable!("DATE to TIMESTAMP precision above 3 is not admitted"), + }; + Ok(result) +} + +fn cast_timestamp_to_date(array: &ArrayRef, source: &DataType) -> crate::Result<ArrayRef> { + let DataType::Timestamp(timestamp) = source else { + unreachable!("timestamp executor requires a TIMESTAMP source") + }; + let result: ArrayRef = match timestamp.precision() { + 0..=3 => Arc::new( + downcast_primitive::<TimestampMillisecondType>(array, "TIMESTAMP")? + .unary::<_, Date32Type>(|value| (value / 86_400_000) as i32), Review Comment: [P1] Use floor division for pre-epoch timestamps\n\nRust integer division truncates toward zero, but Paimon Java uses . Consequently becomes epoch day instead of , and becomes instead of , silently shifting pre-1970 non-midnight values and residual-filter results by one day. Please use in the millisecond, microsecond, and nanosecond branches and change the new test expectation to ; the current test passes only because it codifies the wrong truncation. ########## crates/paimon/src/catalog/rest/rest_catalog.rs: ########## @@ -278,6 +279,25 @@ impl Catalog for RESTCatalog { changes: Vec<SchemaChange>, ignore_if_not_exists: bool, ) -> Result<()> { + if changes + .iter() + .any(|change| matches!(change, SchemaChange::UpdateColumnType { .. })) + { + let table = match self.get_table(identifier).await { + Ok(table) => table, + Err(Error::TableNotExist { .. }) if ignore_if_not_exists => return Ok(()), + Err(error) => return Err(error), + }; + let new_schema = apply_schema_changes(table.schema(), &changes, identifier)?; + validate_type_evolution_precommit( + table.file_io(), + table.location(), + table.schema(), + &new_schema, + ) + .await?; Review Comment: [P1] Fence this preflight with the schema version\n\nThe validation is based on a prior , while carries only and no expected schema ID. For example, this client can validate , a concurrent Java client can commit and write a Decimal file, and then this stale POST is applied by the server as (supported by Java). The final history now requires , which this PR's Rust executor does not implement, so Rust fails to read the intermediate file despite the preflight succeeding. The preflight must run atomically on the server, or the request must assert the schema ID and retry validation on conflict. ########## crates/paimon/src/table/data_file_reader.rs: ########## @@ -397,42 +420,45 @@ impl DataFileReader { None } else { let data_field = &data_fields.as_ref().unwrap()[data_idx as usize]; - batch_schema - .index_of(data_field.name()) - .ok() - .map(|col_idx| batch.column(col_idx)) + match batch_schema.index_of(data_field.name()) { + Ok(col_idx) => Some(( + batch.column(col_idx), + decoded_data_type(data_field, &format_read_fields)?, + )), + Err(_) => None, + } } } else if let Some(ref df) = data_fields { - batch_schema - .index_of(df[i].name()) - .ok() - .map(|col_idx| batch.column(col_idx)) + let data_field = &df[i]; + match batch_schema.index_of(data_field.name()) { + Ok(col_idx) => Some(( + batch.column(col_idx), + decoded_data_type(data_field, &format_read_fields)?, + )), + Err(_) => None, + } } else { batch_schema .index_of(target_field.name()) .ok() - .map(|col_idx| batch.column(col_idx)) + .map(|col_idx| (batch.column(col_idx), read_type[i].data_type())) }; match source_col { - Some(col) => { - if col.data_type() == target_field.data_type() { - columns.push(col.clone()); - } else { - let casted = cast(col, target_field.data_type()).map_err(|e| { - Error::UnexpectedError { - message: format!( - "Failed to cast column '{}' from {:?} to {:?}: {e}", - target_field.name(), - col.data_type(), - target_field.data_type() - ), - source: Some(Box::new(e)), - } - })?; - columns.push(casted); - } - } + Some((col, source_type)) => columns.push( + cast_array_for_schema_evolution( Review Comment: [P1] Keep in-flight readers able to read post-ALTER files\n\nThis replacement of the generic Arrow cast assumes the table schema is always newer than the file schema, but long-lived providers keep a fixed while their scans plan the latest snapshot. A provider registered under can therefore see a new file after another client performs ; here the source is Decimal and the target is the provider's Int schema, and the new one-way executor rejects even though the previous Arrow cast handled it. This breaks running readers/jobs across ALTER. Please either implement the reverse executors required by old-schema readers or fence/refresh providers so a fixed schema cannot scan newer file schemas. ########## crates/paimon/src/catalog/filesystem.rs: ########## @@ -458,29 +459,13 @@ impl Catalog for FileSystemCatalog { full_name: identifier.full_name(), })?; - let new_schema = current - .apply_changes(changes) - .map_err(|e| fill_table_name(e, identifier))?; + let new_schema = apply_schema_changes(¤t, &changes, identifier)?; + validate_type_evolution_precommit(&self.file_io, &table_path, ¤t, &new_schema) Review Comment: [P1] Include the schema version in the index-build commit guard\n\nThis preflight and schema save are not coordinated with global-index commits, and an ALTER does not advance the snapshot ID. An index build planned at snapshot S/schema N can therefore commit after this saves schema N+1 because still sees S; the commit stamps the index with N and also creates a new snapshot whose is the stale N. Fresh Rust drops the index, time travel/branch creation observes the wrong schema, and an old Java reader has no build-schema filter and may use the stale index. Please atomically guard index commits on both snapshot and schema ID (or serialize them with ALTER), and stamp snapshots from the latest schema as Java does. ########## crates/paimon/src/spec/index_manifest.rs: ########## @@ -71,7 +71,8 @@ pub const INDEX_MANIFEST_ENTRY_SCHEMA: &str = r#"{ {"name": "_INDEX_FIELD_ID", "type": "int"}, {"name": "_EXTRA_FIELD_IDS", "type": ["null", {"type": "array", "items": "int"}], "default": null}, {"name": "_INDEX_META", "type": ["null", "bytes"], "default": null}, - {"name": "_SOURCE_META", "type": ["null", "bytes"], "default": null} + {"name": "_SOURCE_META", "type": ["null", "bytes"], "default": null}, + {"name": "_BUILD_SCHEMA_ID", "type": ["null", "long"], "default": null} Review Comment: [P2] Do not let old Java rewrites erase this marker\n\nCurrent Java has six fields, and reads all active entries and serializes a new combined manifest. An old Java writer therefore drops this seventh field on any index/DV manifest rewrite. Back on new Rust, a valid index built after type evolution is treated as legacy/incompatible and filtered out, while final overlap validation still retains the physical entry and blocks rebuilding it until an explicit drop. A trailing top-level field fixes the old-Rust decode issue but is still lost by an old serializer. Please gate emission on a Java version that preserves the marker, or encode the provenance in an opaque payload old writers round-trip unchanged. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
