[client][rust] Align fixed-schema scans by column ID - #3771
Conversation
|
While reviewing the Rust Arrow fixed-schema read path, I noticed that historical data is currently aligned by field name rather than by stable column ID. Under typical schema-evolution semantics, this could prevent old data from being matched correctly after a column rename, so I initially considered it a bug. However, after local testing and further investigation, I confirmed that Fluss currently supports only a limited set of schema-evolution operations, such as adding nullable columns. Although the client exposes a Nevertheless, the Arrow fixed-schema path is inconsistent with the KV |
fresh-borzoni
left a comment
fresh-borzoni
left a comment
There was a problem hiding this comment.
@slfan1989 Thank you, good catch, left a couple of comments
Since this is about making the arrow and KV read paths agree on how columns are identified - there's a third read path that doesn't handle schema_id at all: batch_scanner.rs:192 decodes every log batch with the current table schema, while decode_kv_batch right below builds a decoder per schema id. Doesn't a limit scan over pre-ADD-COLUMN batches fail there?
Can you check if it's a bug, and if it's what it looks like - file an issue/PR?
| projected_fields, | ||
| ) | ||
| .with_schema_getter(Arc::clone(&schema_getter)); | ||
| if fixed_schema { |
There was a problem hiding this comment.
with_fixed_schema runs even when projected_fields is Some, so fixed_target pairs a full-schema mapping with the projected arrow schema and row type from the initial context. Dead today since register_schema returns early under projection, but the two halves disagree and the only thing that would catch it is a per-batch Schema alignment has N indexes for K target fields.
Mb skip the capture when projection is active? WDYT
| ) -> Result<RecordBatch> { | ||
| if record_batch.schema_ref() == &target_schema { | ||
| return Ok(record_batch); | ||
| if schema_alignment.len() != target_schema.fields().len() { |
There was a problem hiding this comment.
nit: This length check (and the usize::try_from + bounds check below) validates a per-context invariant on every batch.
Should we assert it once in with_target_schema_alignment, next to the existing debug_assert!(self.projection.is_none()), and leave the it clean here?
| 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)?; |
There was a problem hiding this comment.
nit: ..._preserves_renamed_column_by_id loops [false, true], this one is local-only. Same loop here?
2 participants
Purpose
Linked issue: close #3770
The Rust fixed-schema log scanner previously aligned historical Arrow batches with the scanner schema by field name.
This behavior was inconsistent with the KV
FixedSchemaDecoder, which uses stable Fluss column IDs. Name-based alignment may lose values when a column keeps the same ID but changes its name, or incorrectly reuse values whendifferent column IDs have the same name.
This PR aligns fixed-schema log reads by column ID, consistent with
FixedSchemaDecoder.Brief change log
index_mappingutility, using stable column IDs.ReadContextfor both local and remote log reads.Tests
Added Rust unit tests.
API and Format
No public API or storage format changes.
Documentation
No documentation changes are required.