[client][rust] Align fixed-schema scans by column ID

fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs

@@ -984,13 +984,17 @@ mod tests {
         ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType,
         DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
     };
-    use crate::metadata::{DataField, DataTypes, PhysicalTablePath, RowType, TablePath};
+    use crate::metadata::{
+        Column, DataField, DataTypes, PhysicalTablePath, RowType, Schema, TableDescriptor,
+        TableInfo, TablePath,
+    };
     use crate::record::{
         APPEND_ONLY_FLAG_MASK, ATTRIBUTES_OFFSET, LENGTH_LENGTH, LENGTH_OFFSET, LOG_OVERHEAD,
         MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, ReadContext, to_arrow_schema,
     };
     use crate::row::GenericRow;
     use crate::test_utils::{build_table_info, build_table_info_with_columns};
+    use arrow::array::{Array, StringArray};
     use std::sync::Arc;
 
     fn expect_data<T>(result: FetchResult<T>) -> T {
@@ -1017,6 +1021,94 @@ mod tests {
         )))
     }
 
+    fn table_info_with_column_ids(
+        table_path: TablePath,
+        schema_id: i32,
+        columns: Vec<Column>,
+    ) -> Result<TableInfo> {
+        let schema = Schema::builder().with_columns(columns).build()?;
+        let descriptor = TableDescriptor::builder()
+            .schema(schema)
+            .distributed_by(Some(1), vec![])
+            .build()?;
+        Ok(TableInfo::of(table_path, 1, schema_id, descriptor, 0, 0))
+    }
+
+    fn fetch_fixed_schema_batch(
+        source_table_info: Arc<TableInfo>,
+        target_table_info: &TableInfo,
+        row: &GenericRow,
+        is_remote: bool,
+    ) -> Result<RecordBatch> {
+        let source_schema_id = source_table_info.get_schema_id();
+        let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(
+            source_table_info.get_table_path().clone(),
+        )));
+        let mut builder = MemoryLogRecordsArrowBuilder::new(
+            source_schema_id,
+            source_table_info.get_row_type(),
+            false,
+            ArrowCompressionInfo {
+                compression_type: ArrowCompressionType::None,
+                compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+            },
+            usize::MAX,
+            Arc::new(ArrowCompressionRatioEstimator::default()),
+        )?;
+        let record = WriteRecord::for_append(
+            Arc::clone(&source_table_info),
+            physical_table_path,
+            source_schema_id,
+            row,
+        );
+        builder.append(&record)?;
+        let data = builder.build()?;
+
+        let target_arrow_schema = to_arrow_schema(target_table_info.get_row_type())?;
+        let target_row_type = Arc::new(target_table_info.get_row_type().clone());
+        let local_ctx = Arc::new(
+            ReadContext::new(
+                target_arrow_schema.clone(),
+                Arc::clone(&target_row_type),
+                false,
+            )
+            .with_fluss_row_type(Arc::clone(&target_row_type)),
+        );
+        let remote_ctx = Arc::new(
+            ReadContext::new(target_arrow_schema, Arc::clone(&target_row_type), true)
+                .with_fluss_row_type(target_row_type),
+        );
+        let resolver = Arc::new(
+            ReadContextResolver::new(
+                target_table_info.get_schema_id() as i16,
+                local_ctx,
+                remote_ctx,
+                None,
+            )
+            .with_fixed_schema(target_table_info.get_schema()),
+        );
+        let mut fetch = DefaultCompletedFetch::new(
+            TableBucket::new(1, 0),
+            LogRecordsBatches::new(data.clone()),
+            data.len(),
+            Arc::clone(&resolver),
+            is_remote,
+            0,
+            0,
+        );
+
+        assert!(matches!(
+            fetch.fetch_batches(1)?,
+            FetchResult::SchemaRequired(schema_id) if schema_id == source_schema_id as i16
+        ));
+        resolver.register_schema(source_schema_id as i16, source_table_info.get_schema())?;
+        Ok(expect_data(fetch.fetch_batches(1)?)
+            .into_iter()
+            .next()
+            .expect("one fixed-schema batch")
+            .0)
+    }
+
     struct ErrorPendingFetch {
         table_bucket: TableBucket,
     }
@@ -1193,7 +1285,8 @@ mod tests {
                 .with_fluss_row_type(new_row_type_arc),
         );
         let resolver = Arc::new(
-            ReadContextResolver::new(1, local_ctx, remote_ctx, None).with_fixed_schema(true),
+            ReadContextResolver::new(1, local_ctx, remote_ctx, None)
+                .with_fixed_schema(new_table_info.get_schema()),
         );
 
         let mut fetch = DefaultCompletedFetch::new(
@@ -1222,6 +1315,76 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn fixed_schema_fetch_batches_preserves_renamed_column_by_id() -> Result<()> {
+        let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+        let source_table_info = Arc::new(table_info_with_column_ids(
+            table_path.clone(),
+            0,
+            vec![
+                Column::new("id", DataTypes::int()).with_id(0),
+                Column::new("old_name", DataTypes::string()).with_id(1),
+            ],
+        )?);
+        let target_table_info = table_info_with_column_ids(
+            table_path,
+            1,
+            vec![
+                Column::new("id", DataTypes::int()).with_id(0),
+                Column::new("new_name", DataTypes::string()).with_id(1),
+            ],
+        )?;
+        let mut row = GenericRow::new(2);
+        row.set_field(0, 1_i32);
+        row.set_field(1, "alice");
+
+        for is_remote in [false, true] {
+            let batch = fetch_fixed_schema_batch(
+                Arc::clone(&source_table_info),
+                &target_table_info,
+                &row,
+                is_remote,
+            )?;
+            assert_eq!(batch.schema().field(1).name(), "new_name");
+            let names = batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .expect("renamed string column");
+            assert_eq!(names.null_count(), 0);
+            assert_eq!(names.value(0), "alice");
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn fixed_schema_fetch_batches_does_not_alias_same_name_with_different_id() -> Result<()> {
+        let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+        let source_table_info = Arc::new(table_info_with_column_ids(
+            table_path.clone(),
+            0,
+            vec![
+                Column::new("id", DataTypes::int()).with_id(0),
+                Column::new("name", DataTypes::string()).with_id(1),
+            ],
+        )?);
+        let target_table_info = table_info_with_column_ids(
+            table_path,
+            1,
+            vec![
+                Column::new("id", DataTypes::int()).with_id(0),
+                Column::new("name", DataTypes::string()).with_id(2),
+            ],
+        )?;
+        let mut row = GenericRow::new(2);
+        row.set_field(0, 1_i32);
+        row.set_field(1, "alice");
+
+        let batch = fetch_fixed_schema_batch(source_table_info, &target_table_info, &row, false)?;
+        assert_eq!(batch.column(1).null_count(), 1);
+        Ok(())
+    }
+
     /// A `-U`/`+U` pair must not be split across polls: even when `max_records`
     /// falls between them, `fetch_records` pulls the matching `+U` so the batch
     /// ends on a complete pair (mirrors Java `CompletedFetch.fetchRecords`).

fluss-rust/crates/fluss/src/client/table/read_context_resolver.rs

@@ -29,7 +29,7 @@
 
 use crate::client::ClientSchemaGetter;
 use crate::error::{Error, Result};
-use crate::metadata::{RowType, Schema};
+use crate::metadata::{RowType, Schema, index_mapping};
 use crate::record::{ReadContext, to_arrow_schema};
 use arrow_schema::SchemaRef;
 use parking_lot::RwLock;
@@ -47,11 +47,10 @@ pub(crate) struct ReadContextResolver {
     /// Used to lazily fetch schema versions when decoding a batch whose schema
     /// was not prewarmed by the fetch response path.
     schema_getter: Option<Arc<ClientSchemaGetter>>,
-    /// When true, contexts for older schemas still decode with their write-time
-    /// schema, then align the output to the scanner creation schema.
-    fixed_schema: bool,
-    fixed_target_schema: Option<SchemaRef>,
-    fixed_target_row_type: Option<Arc<RowType>>,
+    /// Fixed-schema target captured when the scanner is created. When set,
+    /// older batches decode with their write-time schema and then align to
+    /// this target by stable Fluss column ID.
+    fixed_target: Option<FixedSchemaTarget>,
 }
 
 /// A pair of ReadContexts for local and remote reads.
@@ -60,6 +59,12 @@ struct ResolvedContexts {
     remote: Arc<ReadContext>,
 }
 
+struct FixedSchemaTarget {
+    fluss_schema: Schema,
+    arrow_schema: SchemaRef,
+    row_type: Arc<RowType>,
+}
+
 impl ReadContextResolver {
     /// Create a new resolver with the initial schema's ReadContexts.
     pub fn new(
@@ -81,9 +86,7 @@ impl ReadContextResolver {
             contexts: RwLock::new(map),
             projected_fields,
             schema_getter: None,
-            fixed_schema: false,
-            fixed_target_schema: None,
-            fixed_target_row_type: None,
+            fixed_target: None,
         }
     }
 
@@ -92,23 +95,17 @@ impl ReadContextResolver {
         self
     }
 
-    pub fn with_fixed_schema(mut self, fixed_schema: bool) -> Self {
-        self.fixed_schema = fixed_schema;
-        if fixed_schema {
-            let fixed_target = {
-                let guard = self.contexts.read();
-                guard
-                    .get(&self.initial_schema_id)
-                    .map(|ctx| (ctx.local.target_schema(), ctx.local.row_type_arc()))
-            };
-            if let Some((target_schema, target_row_type)) = fixed_target {
-                self.fixed_target_schema = Some(target_schema);
-                self.fixed_target_row_type = Some(target_row_type);
-            }
-        } else {
-            self.fixed_target_schema = None;
-            self.fixed_target_row_type = None;
-        }
+    pub fn with_fixed_schema(mut self, target_fluss_schema: &Schema) -> Self {
+        self.fixed_target = {
+            let guard = self.contexts.read();
+            guard
+                .get(&self.initial_schema_id)
+                .map(|ctx| FixedSchemaTarget {
+                    fluss_schema: target_fluss_schema.clone(),
+                    arrow_schema: ctx.local.target_schema(),
+                    row_type: ctx.local.row_type_arc(),
+                })
+        };
         self
     }
 
@@ -170,10 +167,23 @@ impl ReadContextResolver {
         let source_row_type = schema.row_type();
         let source_arrow_schema = to_arrow_schema(source_row_type)?;
         let source_row_type_arc = Arc::new(source_row_type.clone());
-        let output_row_type = self
-            .fixed_target_row_type
-            .clone()
-            .unwrap_or_else(|| source_row_type_arc.clone());
+        let fixed_target = self
+            .fixed_target
+            .as_ref()
+            .map(|target| {
+                index_mapping(schema, &target.fluss_schema).map(|mapping| {
+                    (
+                        target.arrow_schema.clone(),
+                        target.row_type.clone(),
+                        Arc::<[i32]>::from(mapping.into_boxed_slice()),
+                    )
+                })
+            })
+            .transpose()?;
+        let output_row_type = fixed_target
+            .as_ref()
+            .map(|(_, row_type, _)| row_type.clone())
+            .unwrap_or(source_row_type_arc);
 
         let mut local_context =
             ReadContext::new(source_arrow_schema.clone(), output_row_type.clone(), false)
@@ -182,11 +192,11 @@ impl ReadContextResolver {
             ReadContext::new(source_arrow_schema, output_row_type.clone(), true)
                 .with_fluss_row_type(output_row_type);
 
-        if self.fixed_schema {
-            if let Some(target_schema) = &self.fixed_target_schema {
-                local_context = local_context.with_target_schema_alignment(target_schema.clone());
-                remote_context = remote_context.with_target_schema_alignment(target_schema.clone());
-            }
+        if let Some((target_schema, _, schema_alignment)) = fixed_target {
+            local_context = local_context
+                .with_target_schema_alignment(target_schema.clone(), Arc::clone(&schema_alignment));
+            remote_context =
+                remote_context.with_target_schema_alignment(target_schema, schema_alignment);
         }
 
         let local_context = Arc::new(local_context);

fluss-rust/crates/fluss/src/client/table/scanner.rs

@@ -1234,16 +1234,17 @@ impl LogFetcher {
         );
 
         let initial_schema_id = table_info.get_schema_id() as i16;
-        let resolver = Arc::new(
-            ReadContextResolver::new(
-                initial_schema_id,
-                read_context,
-                remote_read_context,
-                projected_fields,
-            )
-            .with_schema_getter(Arc::clone(&schema_getter))
-            .with_fixed_schema(fixed_schema),
-        );
+        let mut resolver = ReadContextResolver::new(
+            initial_schema_id,
+            read_context,
+            remote_read_context,
+            projected_fields,
+        )
+        .with_schema_getter(Arc::clone(&schema_getter));
+        if fixed_schema {
+            resolver = resolver.with_fixed_schema(table_info.get_schema());
+        }
+        let resolver = Arc::new(resolver);
 
         let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
         let log_fetch_buffer = Arc::new(LogFetchBuffer::new(Arc::clone(&resolver)));