fresh-borzoni commented on code in PR #654: URL: https://github.com/apache/fluss-rust/pull/654#discussion_r3521521592
########## crates/fluss/src/client/table/read_context_resolver.rs: ########## @@ -0,0 +1,186 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Per-schema `ReadContext` cache for schema evolution support. +//! +//! In DYNAMIC mode (no projection), records are returned with their write-time +//! schema: old-schema batches return fewer columns, new-schema batches return +//! more columns. +//! +//! When projection is active, the schema is pinned at scanner creation time +//! and all batches use the initial ReadContext regardless of schema_id. + +use crate::error::Result; +use crate::metadata::Schema; +use crate::record::{ReadContext, to_arrow_schema}; +use parking_lot::RwLock; +use std::collections::HashMap; +use std::sync::Arc; + +/// Resolves `ReadContext` per schema version to support schema evolution. +pub(crate) struct ReadContextResolver { + /// Schema ID at scanner creation time. + initial_schema_id: i16, + /// ReadContexts keyed by schema_id. Contains both local and remote contexts. + contexts: RwLock<HashMap<i16, ResolvedContexts>>, + /// When Some, projection is active and schema is pinned to the initial one. + projected_fields: Option<Vec<usize>>, +} + +/// A pair of ReadContexts for local and remote reads. +struct ResolvedContexts { + local: ReadContext, + remote: ReadContext, +} + +impl ReadContextResolver { + /// Create a new resolver with the initial schema's ReadContexts. + pub fn new( + initial_schema_id: i16, + local_context: ReadContext, + remote_context: ReadContext, + projected_fields: Option<Vec<usize>>, + ) -> Self { + let mut map = HashMap::new(); + map.insert( + initial_schema_id, + ResolvedContexts { + local: local_context, + remote: remote_context, + }, + ); + Self { + initial_schema_id, + contexts: RwLock::new(map), + projected_fields, + } + } + + /// Resolve the ReadContext for the given schema_id. + /// Returns the initial context if projection is active (schema pinned). + /// Returns None if the schema_id is not yet cached. + pub fn resolve(&self, schema_id: i16, is_remote: bool) -> Option<ReadContext> { + // If projection is active, always return the initial context + let effective_id = if self.projected_fields.is_some() { + self.initial_schema_id + } else { + schema_id + }; + + let guard = self.contexts.read(); + guard.get(&effective_id).map(|ctx| { + if is_remote { + ctx.remote.clone() Review Comment: this returns an owned ReadContext clone per batch, where it used to be a borrow. Fine in the no-projection case, but with projection it deep-copies the Vecs on every batch. Could we store Arc<ReadContext> and return an Arc clone? Then it's a single refcount bump and the projection Vecs aren't copied this way ########## crates/fluss/src/client/table/log_fetch_buffer.rs: ########## @@ -518,7 +525,8 @@ impl DefaultCompletedFetch { }; let log_batch = log_batch_result?; - let mut record_batch = log_batch.record_batch(&self.read_context)?; + let read_context = self.resolve_context_for_batch(&log_batch)?; Review Comment: After a schema change, batches here have different column counts, but the reader advertises one fixed schema. Consumers like DataFusion/pyarrow crash on that. Should we pad each batch up to the advertised schema with null columns for the missing ones. Also probably needs a test, the IT only covers the row scanner ########## crates/fluss/src/client/table/scanner.rs: ########## @@ -1581,7 +1607,7 @@ impl LogFetcher { Self::pending_remote_fetches( remote_log_downloader.clone(), log_fetch_buffer.clone(), - remote_read_context.clone(), + Arc::clone(&resolver), Review Comment: Schema prewarm only runs on the local path, not here. So reading tiered data written under an older schema fails with No ReadContext found for schema_id N. The local branch below fetches missing schemas before decoding and the remote path needs the same. WDYT? -- 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]
