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

[slfan1989]

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 when
different column IDs have the same name.

This PR aligns fixed-schema log reads by column ID, consistent with FixedSchemaDecoder.

Brief change log

• Capture the fixed target Fluss schema when creating a fixed-schema log scanner.
• Build source-to-target mappings with the existing index_mapping utility, using stable column IDs.
• Pass the mapping to Arrow ReadContext for both local and remote log reads.
• Preserve values when source and target columns have the same ID.
• Fill target columns with NULL when their IDs do not exist in the source schema.
• Prevent columns with the same name but different IDs from being incorrectly matched.
• Add regression tests for local and remote fixed-schema decoding.

Tests

Added Rust unit tests.

API and Format

No public API or storage format changes.

Documentation

No documentation changes are required.

[slfan1989]

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 RENAME COLUMN interface, the corresponding functionality has not yet been implemented on the server. Therefore, this issue cannot currently be reproduced through the complete client-server workflow.

Nevertheless, the Arrow fixed-schema path is inconsistent with the KV FixedSchemaDecoder in how fields are aligned. To provide consistent semantics across both read paths and establish a reliable foundation for supporting additional schema-evolution operations in the future, I submit a PR that changes Arrow fixed-schema alignment from field names to stable column IDs.

[fresh-borzoni]

@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?

[fresh-borzoni]

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

[slfan1989]

Thanks for pointing this out. Since projection pins the resolver to the initial context, the fixed target is not used in this case. I updated with_fixed_schema to skip fixed_target capture when projection is active and added a regression test to cover this behavior.

[fresh-borzoni]

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?

[slfan1989]

Thanks for the suggestion. I moved the alignment length, source-index bounds, and type invariant checks into with_target_schema_alignment. The per-batch alignment path now only applies the prevalidated mapping, while RecordBatch::try_new remains the final schema validation. I also added tests for invalid alignment lengths and source indexes.

[fresh-borzoni]

nit: ..._preserves_renamed_column_by_id loops [false, true], this one is local-only. Same loop here?

[slfan1989]

Updated this test to loop over [false, true], so it now covers both local and remote read paths, consistent with ...preserves_renamed_column_by_id.

[slfan1989]

@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?

@fresh-borzoni Thanks for pointing this out. I confirmed that this is indeed a bug. I reproduced it with an Arrow log batch written before an ADD COLUMN.

The limit-scan log path decodes every batch using the current table schema, which fails when reading the older batch because the newly added field is absent.

The log batch already carries its schema_id, so this path should resolve the source schema per batch and align it with the current target schema, consistent with the regular log-fetch and KV paths.

I’ll track this in a separate issue(#3808) and follow-up PR, and link them here once created.

[fresh-borzoni]

@slfan1989 Thank you, LGTM 👍

[slfan1989]

@slfan1989 Thank you, LGTM 👍

@fresh-borzoni Thanks for the review and the helpful comments! I really appreciate it.