You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/07/30 12:29:13 UTC
(fluss) branch main updated: [client][rust] Align fixed-schema scans by column ID (#3771)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 505482e50 [client][rust] Align fixed-schema scans by column ID (#3771)
505482e50 is described below
commit 505482e5081e57b49f282239034da527c83a78d1
Author: slfan1989 <55...@users.noreply.github.com>
AuthorDate: Thu Jul 30 20:29:09 2026 +0800
[client][rust] Align fixed-schema scans by column ID (#3771)
* [client][rust] Align fixed-schema scans by column ID
* [client][rust] Skip fixed target capture for projected scans.
* [client][rust] Move schema alignment validation out of batch decoding.
* [client][rust] Extend fixed-schema test coverage
---
.../fluss/src/client/table/log_fetch_buffer.rs | 174 ++++++++++++++++++++-
.../src/client/table/read_context_resolver.rs | 120 ++++++++++----
.../crates/fluss/src/client/table/scanner.rs | 21 +--
fluss-rust/crates/fluss/src/record/arrow.rs | 118 +++++++++-----
4 files changed, 349 insertions(+), 84 deletions(-)
diff --git a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs b/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs
index 51354f540..b057612d6 100644
--- a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs
+++ b/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,83 @@ 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");
+
+ 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.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`).
diff --git a/fluss-rust/crates/fluss/src/client/table/read_context_resolver.rs b/fluss-rust/crates/fluss/src/client/table/read_context_resolver.rs
index 655891f49..90516a55a 100644
--- a/fluss-rust/crates/fluss/src/client/table/read_context_resolver.rs
+++ b/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,20 @@ 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 {
+ if self.projected_fields.is_some() {
+ return 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 +170,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 +195,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);
@@ -213,3 +226,42 @@ impl ReadContextResolver {
self.projected_fields.as_deref()
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::ReadContextResolver;
+ use crate::error::Result;
+ use crate::metadata::{DataTypes, Schema};
+ use crate::record::{ReadContext, to_arrow_schema};
+ use std::sync::Arc;
+
+ #[test]
+ fn fixed_schema_does_not_capture_target_with_projection() -> Result<()> {
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("name", DataTypes::string())
+ .build()?;
+ let arrow_schema = to_arrow_schema(schema.row_type())?;
+ let projected_fields = vec![0];
+ let projected_row_type = Arc::new(schema.row_type().project(&projected_fields)?);
+ let local_context = Arc::new(ReadContext::with_projection_pushdown(
+ arrow_schema.clone(),
+ Arc::clone(&projected_row_type),
+ projected_fields.clone(),
+ false,
+ )?);
+ let remote_context = Arc::new(ReadContext::with_projection_pushdown(
+ arrow_schema,
+ projected_row_type,
+ projected_fields.clone(),
+ true,
+ )?);
+
+ let resolver =
+ ReadContextResolver::new(1, local_context, remote_context, Some(projected_fields))
+ .with_fixed_schema(&schema);
+
+ assert!(resolver.fixed_target.is_none());
+ Ok(())
+ }
+}
diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs
index 5750a807b..6e4d046e7 100644
--- a/fluss-rust/crates/fluss/src/client/table/scanner.rs
+++ b/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)));
diff --git a/fluss-rust/crates/fluss/src/record/arrow.rs b/fluss-rust/crates/fluss/src/record/arrow.rs
index dbbbcc991..8de9f4b53 100644
--- a/fluss-rust/crates/fluss/src/record/arrow.rs
+++ b/fluss-rust/crates/fluss/src/record/arrow.rs
@@ -20,7 +20,7 @@ use crate::compression::{
ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType,
};
use crate::error::{Error, Result};
-use crate::metadata::{DataField, DataType, RowType};
+use crate::metadata::{DataField, DataType, RowType, UNEXIST_MAPPING};
use crate::record::{ChangeType, ScanRecord};
use crate::row::column_vector::TypedBatch;
use crate::row::column_writer::{ColumnWriter, round_up_to_8};
@@ -1440,7 +1440,7 @@ pub struct ReadContext {
projection: Option<Projection>,
is_from_remote: bool,
fluss_row_type: Option<Arc<RowType>>,
- align_to_target_schema: bool,
+ schema_alignment: Option<Arc<[i32]>>,
}
#[derive(Clone)]
@@ -1466,7 +1466,7 @@ impl ReadContext {
projection: None,
is_from_remote,
fluss_row_type: None,
- align_to_target_schema: false,
+ schema_alignment: None,
}
}
@@ -1487,13 +1487,42 @@ impl ReadContext {
self.row_type.clone()
}
- pub(crate) fn with_target_schema_alignment(mut self, target_schema: SchemaRef) -> ReadContext {
+ pub(crate) fn with_target_schema_alignment(
+ mut self,
+ target_schema: SchemaRef,
+ schema_alignment: Arc<[i32]>,
+ ) -> ReadContext {
debug_assert!(
self.projection.is_none(),
"target schema alignment is not supported with projection"
);
+ debug_assert_eq!(
+ schema_alignment.len(),
+ target_schema.fields().len(),
+ "schema alignment length must match target schema"
+ );
+ debug_assert!(
+ schema_alignment.iter().all(|source_index| {
+ *source_index == UNEXIST_MAPPING
+ || usize::try_from(*source_index)
+ .is_ok_and(|index| index < self.full_schema.fields().len())
+ }),
+ "schema alignment contains an invalid source index"
+ );
+ debug_assert!(
+ target_schema
+ .fields()
+ .iter()
+ .zip(schema_alignment.iter())
+ .all(|(target_field, source_index)| {
+ *source_index == UNEXIST_MAPPING
+ || self.full_schema.field(*source_index as usize).data_type()
+ == target_field.data_type()
+ }),
+ "schema alignment source and target types must match"
+ );
self.target_schema = target_schema;
- self.align_to_target_schema = true;
+ self.schema_alignment = Some(schema_alignment);
self
}
@@ -1568,7 +1597,7 @@ impl ReadContext {
projection: Some(project),
is_from_remote,
fluss_row_type: None,
- align_to_target_schema: false,
+ schema_alignment: None,
})
}
@@ -1611,7 +1640,7 @@ impl ReadContext {
let resolve_schema = {
// if from remote, no projection, need to use full schema
- if self.is_from_remote || self.align_to_target_schema {
+ if self.is_from_remote || self.schema_alignment.is_some() {
self.full_schema.clone()
} else {
// the record batch from server must be ordered by field pos,
@@ -1664,10 +1693,13 @@ impl ReadContext {
}
_ => record_batch,
};
- let record_batch = if self.align_to_target_schema {
- align_record_batch_to_schema(record_batch, self.target_schema.clone())?
- } else {
- record_batch
+ let record_batch = match &self.schema_alignment {
+ Some(schema_alignment) => align_record_batch_to_schema(
+ record_batch,
+ self.target_schema.clone(),
+ schema_alignment,
+ )?,
+ None => record_batch,
};
Ok(record_batch)
}
@@ -1695,10 +1727,13 @@ impl ReadContext {
}
None => record_batch,
};
- let record_batch = if self.align_to_target_schema {
- align_record_batch_to_schema(record_batch, self.target_schema.clone())?
- } else {
- record_batch
+ let record_batch = match &self.schema_alignment {
+ Some(schema_alignment) => align_record_batch_to_schema(
+ record_batch,
+ self.target_schema.clone(),
+ schema_alignment,
+ )?,
+ None => record_batch,
};
Ok(Some(record_batch))
}
@@ -1707,33 +1742,16 @@ impl ReadContext {
fn align_record_batch_to_schema(
record_batch: RecordBatch,
target_schema: SchemaRef,
+ schema_alignment: &[i32],
) -> Result<RecordBatch> {
- if record_batch.schema_ref() == &target_schema {
- return Ok(record_batch);
- }
-
let row_count = record_batch.num_rows();
- let source_schema = record_batch.schema();
let mut columns = Vec::with_capacity(target_schema.fields().len());
- for target_field in target_schema.fields() {
- match source_schema.index_of(target_field.name()) {
- Ok(source_idx) => {
- let column = record_batch.column(source_idx);
- if column.data_type() != target_field.data_type() {
- return Err(Error::UnexpectedError {
- message: format!(
- "Cannot align column '{}' from type {:?} to {:?}",
- target_field.name(),
- column.data_type(),
- target_field.data_type()
- ),
- source: None,
- });
- }
- columns.push(column.clone());
- }
- Err(_) => columns.push(new_null_array(target_field.data_type(), row_count)),
+ for (target_field, source_index) in target_schema.fields().iter().zip(schema_alignment.iter()) {
+ if *source_index == UNEXIST_MAPPING {
+ columns.push(new_null_array(target_field.data_type(), row_count));
+ } else {
+ columns.push(record_batch.column(*source_index as usize).clone());
}
}
@@ -1884,6 +1902,30 @@ mod tests {
use crate::row::{DataGetters, GenericRow};
use crate::test_utils::build_table_info;
+ fn single_int_read_context() -> (ReadContext, SchemaRef) {
+ let row_type = Arc::new(RowType::new(vec![DataField::new(
+ "id",
+ DataTypes::int(),
+ None,
+ )]));
+ let schema = to_arrow_schema(&row_type).expect("arrow schema");
+ (ReadContext::new(schema.clone(), row_type, false), schema)
+ }
+
+ #[test]
+ #[should_panic(expected = "schema alignment length must match target schema")]
+ fn target_schema_alignment_rejects_length_mismatch() {
+ let (read_context, target_schema) = single_int_read_context();
+ read_context.with_target_schema_alignment(target_schema, Arc::from([]));
+ }
+
+ #[test]
+ #[should_panic(expected = "schema alignment contains an invalid source index")]
+ fn target_schema_alignment_rejects_invalid_source_index() {
+ let (read_context, target_schema) = single_int_read_context();
+ read_context.with_target_schema_alignment(target_schema, Arc::from([-2]));
+ }
+
#[test]
fn test_to_array_type() {
assert_eq!(