diff --git a/fluss-rust/bindings/python/fluss/__init__.pyi b/fluss-rust/bindings/python/fluss/__init__.pyi index 3f7e94b482..9e7b8e75fb 100644 --- a/fluss-rust/bindings/python/fluss/__init__.pyi +++ b/fluss-rust/bindings/python/fluss/__init__.pyi @@ -503,6 +503,13 @@ class TableScan: Self for method chaining. """ ... + def with_fixed_schema(self, fixed_schema: bool) -> "TableScan": + """Control whether schema-evolved batches are aligned to the scanner schema. + + When enabled, older-schema batches are padded/projected to the scanner + creation schema. When disabled, batches keep their write-time schema. + """ + ... def limit(self, n: int) -> "TableScan": """Set a positive row limit for the scan. diff --git a/fluss-rust/bindings/python/src/table.rs b/fluss-rust/bindings/python/src/table.rs index 521a9b3a71..ca8c0556e1 100644 --- a/fluss-rust/bindings/python/src/table.rs +++ b/fluss-rust/bindings/python/src/table.rs @@ -429,6 +429,7 @@ pub struct TableScan { metadata: Arc, table_info: fcore::metadata::TableInfo, projection: Option, + fixed_schema: bool, limit: Option, } @@ -464,6 +465,19 @@ impl TableScan { slf } + /// Control whether schema-evolved log batches are aligned to the scanner schema. + /// + /// When enabled, batches written with older schemas are padded/projected to + /// the schema captured when the scanner is created. When disabled, batches + /// keep their write-time schema. + pub fn with_fixed_schema( + mut slf: PyRefMut<'_, Self>, + fixed_schema: bool, + ) -> PyRefMut<'_, Self> { + slf.fixed_schema = fixed_schema; + slf + } + /// Set a positive row limit, enabling `create_bucket_batch_scanner()`. /// /// Args: @@ -566,12 +580,14 @@ impl TableScan { let metadata = self.metadata.clone(); let table_info = self.table_info.clone(); let projection = self.projection.clone(); + let fixed_schema = self.fixed_schema; future_into_py(py, async move { let fluss_table = fcore::client::FlussTable::new(&conn, metadata, table_info.clone()); let projection_indices = resolve_projection_indices(&projection, &table_info)?; - let table_scan = apply_projection(fluss_table.new_scan(), projection)?; + let table_scan = apply_projection(fluss_table.new_scan(), projection)? + .with_fixed_schema(fixed_schema); let admin = conn .get_admin() @@ -699,6 +715,7 @@ impl FlussTable { metadata: self.metadata.clone(), table_info: self.table_info.clone(), projection: None, + fixed_schema: false, limit: None, } } 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 dee68c9557..794fe9df92 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 @@ -18,6 +18,7 @@ use arrow::array::RecordBatch; use parking_lot::Mutex; +use crate::client::table::read_context_resolver::ReadContextResolver; use crate::client::table::remote_log::{ PrefetchPermit, RemoteLogDownloadFuture, RemoteLogFile, RemoteLogSegment, }; @@ -64,8 +65,9 @@ pub trait CompletedFetch: Send + Sync { fn api_error(&self) -> Option<&ApiError>; fn fetch_error_context(&self) -> Option<&FetchErrorContext>; fn take_error(&mut self) -> Option; - fn fetch_records(&mut self, max_records: usize) -> Result>; - fn fetch_batches(&mut self, max_batches: usize) -> Result>; + fn fetch_records(&mut self, max_records: usize) -> Result>>; + fn fetch_batches(&mut self, max_batches: usize) + -> Result>>; fn is_consumed(&self) -> bool; fn records_read(&self) -> usize; fn drain(&mut self); @@ -76,6 +78,28 @@ pub trait CompletedFetch: Send + Sync { fn next_fetch_offset(&self) -> i64; } +/// Result of synchronously decoding data already available in a completed +/// fetch. Missing schemas are surfaced to the async scanner layer instead of +/// blocking the current Tokio worker from inside this synchronous trait. +pub(crate) enum FetchResult { + /// Decoded data which is ready to return. The value may be empty when the + /// completed fetch has reached its end. + Data(T), + /// Decoding is paused on the current raw batch. The async scanner must + /// fetch/register this schema and then retry the same completed fetch. + SchemaRequired(i16), +} +enum FetchStep { + /// The current raw batch cannot be decoded until this schema ID is loaded. + SchemaRequired(i16), + + /// One record or Arrow batch was produced, and this completed fetch may + /// still contain more data to read. + InProgress(T), + /// The completed fetch has no more raw batches or records. + End, +} + /// Represents a pending fetch that is waiting to be completed pub trait PendingFetch: Send + Sync { fn table_bucket(&self) -> &TableBucket; @@ -85,7 +109,7 @@ pub trait PendingFetch: Send + Sync { /// Thread-safe buffer for completed fetches pub struct LogFetchBuffer { - read_context: ReadContext, + resolver: Arc, completed_fetches: Mutex>>, pending_fetches: Mutex>>>, next_in_line_fetch: Mutex>>, @@ -94,9 +118,9 @@ pub struct LogFetchBuffer { } impl LogFetchBuffer { - pub fn new(read_context: ReadContext) -> Self { + pub fn new(resolver: Arc) -> Self { Self { - read_context, + resolver, completed_fetches: Mutex::new(VecDeque::new()), pending_fetches: Mutex::new(HashMap::new()), next_in_line_fetch: Mutex::new(None), @@ -168,7 +192,7 @@ impl LogFetchBuffer { api_error, fetch_error_context, fetch_offset, - self.read_context.clone(), + Arc::clone(&self.resolver), ); self.completed_fetches .lock() @@ -226,7 +250,7 @@ impl LogFetchBuffer { table_bucket.clone(), error, -1, - self.read_context.clone(), + Arc::clone(&self.resolver), ); completed_to_push.push(Box::new(error_fetch)); } @@ -339,7 +363,8 @@ pub struct DefaultCompletedFetch { fetch_error_context: Option, error: Option, log_record_batch: LogRecordsBatches, - read_context: ReadContext, + resolver: Arc, + is_remote: bool, next_fetch_offset: i64, high_watermark: i64, size_in_bytes: usize, @@ -348,6 +373,15 @@ pub struct DefaultCompletedFetch { records_read: usize, current_record_iterator: Option, current_record_batch: Option, + /// A raw batch which has been read from memory/the remote file but cannot + /// yet be decoded because its schema is not cached. Keeping it here makes + /// schema fetching cancellation-safe and avoids advancing through the + /// rest of a remote segment. + pending_record_batch: Option, + /// Records decoded before discovering that the matching `+U` needs a + /// schema which is not cached yet. They remain uncommitted until the pair + /// can be returned together. + pending_records: Vec, last_record: Option, cached_record_error: Option, corrupt_last_record: bool, @@ -358,7 +392,8 @@ impl DefaultCompletedFetch { table_bucket: TableBucket, log_record_batch: LogRecordsBatches, size_in_bytes: usize, - read_context: ReadContext, + resolver: Arc, + is_remote: bool, fetch_offset: i64, high_watermark: i64, ) -> Self { @@ -368,7 +403,8 @@ impl DefaultCompletedFetch { fetch_error_context: None, error: None, log_record_batch, - read_context, + resolver, + is_remote, next_fetch_offset: fetch_offset, high_watermark, size_in_bytes, @@ -377,6 +413,8 @@ impl DefaultCompletedFetch { records_read: 0, current_record_iterator: None, current_record_batch: None, + pending_record_batch: None, + pending_records: Vec::new(), last_record: None, cached_record_error: None, corrupt_last_record: false, @@ -387,7 +425,7 @@ impl DefaultCompletedFetch { table_bucket: TableBucket, error: Error, fetch_offset: i64, - read_context: ReadContext, + resolver: Arc, ) -> Self { Self { table_bucket, @@ -395,7 +433,8 @@ impl DefaultCompletedFetch { fetch_error_context: None, error: Some(error), log_record_batch: LogRecordsBatches::new(Vec::new()), - read_context, + resolver, + is_remote: false, next_fetch_offset: fetch_offset, high_watermark: -1, size_in_bytes: 0, @@ -404,6 +443,8 @@ impl DefaultCompletedFetch { records_read: 0, current_record_iterator: None, current_record_batch: None, + pending_record_batch: None, + pending_records: Vec::new(), last_record: None, cached_record_error: None, corrupt_last_record: false, @@ -415,7 +456,7 @@ impl DefaultCompletedFetch { api_error: ApiError, fetch_error_context: FetchErrorContext, fetch_offset: i64, - read_context: ReadContext, + resolver: Arc, ) -> Self { Self { table_bucket, @@ -423,7 +464,8 @@ impl DefaultCompletedFetch { fetch_error_context: Some(fetch_error_context), error: None, log_record_batch: LogRecordsBatches::new(Vec::new()), - read_context, + resolver, + is_remote: false, next_fetch_offset: fetch_offset, high_watermark: -1, size_in_bytes: 0, @@ -432,6 +474,8 @@ impl DefaultCompletedFetch { records_read: 0, current_record_iterator: None, current_record_batch: None, + pending_record_batch: None, + pending_records: Vec::new(), last_record: None, cached_record_error: None, corrupt_last_record: false, @@ -439,7 +483,7 @@ impl DefaultCompletedFetch { } /// Get the next fetched record, handling batch iteration and record skipping - fn next_fetched_record(&mut self) -> Result> { + fn next_fetched_record(&mut self) -> Result> { loop { if let Some(record) = self .current_record_iterator @@ -447,18 +491,34 @@ impl DefaultCompletedFetch { .and_then(Iterator::next) { if record.offset() >= self.next_fetch_offset { - return Ok(Some(record)); + return Ok(FetchStep::InProgress(record)); } - } else if let Some(batch_result) = self.log_record_batch.next() { - let batch = batch_result?; - self.current_record_iterator = Some(batch.records(&self.read_context)?); - self.current_record_batch = Some(batch); } else { - if let Some(batch) = self.current_record_batch.take() { - self.next_fetch_offset = batch.next_log_offset(); + if self.pending_record_batch.is_none() { + let Some(batch_result) = self.log_record_batch.next() else { + if let Some(batch) = self.current_record_batch.take() { + self.next_fetch_offset = batch.next_log_offset(); + } + self.drain(); + return Ok(FetchStep::End); + }; + self.pending_record_batch = Some(batch_result?); } - self.drain(); - return Ok(None); + + let batch = self + .pending_record_batch + .as_ref() + .expect("pending batch must be set before schema resolution"); + let Some(read_context) = self.resolve_context_for_batch(batch) else { + return Ok(FetchStep::SchemaRequired(batch.schema_id())); + }; + + let batch = self + .pending_record_batch + .take() + .expect("pending batch must still be present after schema resolution"); + self.current_record_iterator = Some(batch.records(&read_context)?); + self.current_record_batch = Some(batch); } } } @@ -467,18 +527,22 @@ impl DefaultCompletedFetch { /// pushed. A fetch error is cached and surfaced on a later call rather than /// propagated mid-batch (mirrors one iteration of Java /// `CompletedFetch.fetchRecord`). - fn fetch_one_record(&mut self, out: &mut Vec) -> bool { + fn fetch_one_record(&mut self, out: &mut Vec) -> FetchStep<()> { if self.cached_record_error.is_none() { self.corrupt_last_record = true; match self.next_fetched_record() { - Ok(Some(record)) => { + Ok(FetchStep::InProgress(record)) => { self.corrupt_last_record = false; self.last_record = Some(record); } - Ok(None) => { + Ok(FetchStep::End) => { self.corrupt_last_record = false; self.last_record = None; } + Ok(FetchStep::SchemaRequired(schema_id)) => { + self.corrupt_last_record = false; + return FetchStep::SchemaRequired(schema_id); + } Err(e) => { self.cached_record_error = Some(e.to_string()); } @@ -486,13 +550,24 @@ impl DefaultCompletedFetch { } let Some(record) = self.last_record.take() else { - return false; + return FetchStep::End; }; - self.next_fetch_offset = record.offset() + 1; - self.records_read += 1; out.push(record); - true + FetchStep::InProgress(()) + } + + fn finish_records(&mut self, records: Vec) -> FetchResult> { + if let Some(last) = records.last() { + // Reaching the end of the underlying fetch may advance past the + // last visible record (for example over an empty/control tail). + // Preserve that terminal offset instead of moving it backwards. + if !self.consumed { + self.next_fetch_offset = last.offset() + 1; + } + self.records_read += records.len(); + } + FetchResult::Data(records) } fn fetch_error(&self) -> Error { @@ -510,15 +585,29 @@ impl DefaultCompletedFetch { } /// Get the next batch with its base offset. /// Returns (RecordBatch, base_offset) where base_offset is the offset of the first record. - fn next_fetched_batch(&mut self) -> Result> { + fn next_fetched_batch(&mut self) -> Result> { loop { - let Some(log_batch_result) = self.log_record_batch.next() else { - self.drain(); - return Ok(None); + if self.pending_record_batch.is_none() { + let Some(log_batch_result) = self.log_record_batch.next() else { + self.drain(); + return Ok(FetchStep::End); + }; + self.pending_record_batch = Some(log_batch_result?); + } + + let log_batch = self + .pending_record_batch + .as_ref() + .expect("pending batch must be set before schema resolution"); + let Some(read_context) = self.resolve_context_for_batch(log_batch) else { + return Ok(FetchStep::SchemaRequired(log_batch.schema_id())); }; - let log_batch = log_batch_result?; - let mut record_batch = log_batch.record_batch(&self.read_context)?; + let log_batch = self + .pending_record_batch + .take() + .expect("pending batch must still be present after schema resolution"); + let mut record_batch = log_batch.record_batch(&read_context)?; // Skip empty batches if record_batch.num_rows() == 0 { @@ -541,9 +630,15 @@ impl DefaultCompletedFetch { self.next_fetch_offset = log_batch.next_log_offset(); self.records_read += record_batch.num_rows(); - return Ok(Some((record_batch, effective_base_offset))); + return Ok(FetchStep::InProgress((record_batch, effective_base_offset))); } } + + /// Resolve the ReadContext for a given batch based on its schema_id. + fn resolve_context_for_batch(&self, batch: &LogRecordBatch) -> Option> { + let schema_id = batch.schema_id(); + self.resolver.resolve(schema_id, self.is_remote) + } } impl CompletedFetch for DefaultCompletedFetch { @@ -563,7 +658,7 @@ impl CompletedFetch for DefaultCompletedFetch { self.error.take() } - fn fetch_records(&mut self, max_records: usize) -> Result> { + fn fetch_records(&mut self, max_records: usize) -> Result>> { if let Some(error) = self.error.take() { return Err(error); } @@ -582,14 +677,26 @@ impl CompletedFetch for DefaultCompletedFetch { } if self.consumed { - return Ok(Vec::new()); + return Ok(FetchResult::Data(Vec::new())); } - let mut scan_records = Vec::new(); - - for _ in 0..max_records { - if !self.fetch_one_record(&mut scan_records) { - break; + let mut scan_records = std::mem::take(&mut self.pending_records); + + while scan_records.len() < max_records { + match self.fetch_one_record(&mut scan_records) { + FetchStep::InProgress(()) => {} + FetchStep::End => break, + FetchStep::SchemaRequired(schema_id) => { + if scan_records.is_empty() + || scan_records + .last() + .is_some_and(|record| *record.change_type() == ChangeType::UpdateBefore) + { + self.pending_records = scan_records; + return Ok(FetchResult::SchemaRequired(schema_id)); + } + return Ok(self.finish_records(scan_records)); + } } } @@ -601,17 +708,26 @@ impl CompletedFetch for DefaultCompletedFetch { .last() .is_some_and(|record| *record.change_type() == ChangeType::UpdateBefore) { - self.fetch_one_record(&mut scan_records); + match self.fetch_one_record(&mut scan_records) { + FetchStep::InProgress(()) | FetchStep::End => {} + FetchStep::SchemaRequired(schema_id) => { + self.pending_records = scan_records; + return Ok(FetchResult::SchemaRequired(schema_id)); + } + } } if self.cached_record_error.is_some() && scan_records.is_empty() { return Err(self.fetch_error()); } - Ok(scan_records) + Ok(self.finish_records(scan_records)) } - fn fetch_batches(&mut self, max_batches: usize) -> Result> { + fn fetch_batches( + &mut self, + max_batches: usize, + ) -> Result>> { if let Some(error) = self.error.take() { return Err(error); } @@ -626,19 +742,25 @@ impl CompletedFetch for DefaultCompletedFetch { } if self.consumed { - return Ok(Vec::new()); + return Ok(FetchResult::Data(Vec::new())); } let mut batches = Vec::with_capacity(max_batches.min(16)); for _ in 0..max_batches { match self.next_fetched_batch()? { - Some(batch_with_offset) => batches.push(batch_with_offset), - None => break, + FetchStep::InProgress(batch_with_offset) => batches.push(batch_with_offset), + FetchStep::End => break, + FetchStep::SchemaRequired(schema_id) => { + if batches.is_empty() { + return Ok(FetchResult::SchemaRequired(schema_id)); + } + return Ok(FetchResult::Data(batches)); + } } } - Ok(batches) + Ok(FetchResult::Data(batches)) } fn is_consumed(&self) -> bool { @@ -657,6 +779,8 @@ impl CompletedFetch for DefaultCompletedFetch { self.cached_record_error = None; self.corrupt_last_record = false; self.last_record = None; + self.pending_record_batch = None; + self.pending_records.clear(); } fn size_in_bytes(&self) -> usize { @@ -714,11 +838,14 @@ impl CompletedFetch for RemoteCompletedFetch { self.inner.take_error() } - fn fetch_records(&mut self, max_records: usize) -> Result> { + fn fetch_records(&mut self, max_records: usize) -> Result>> { self.inner.fetch_records(max_records) } - fn fetch_batches(&mut self, max_batches: usize) -> Result> { + fn fetch_batches( + &mut self, + max_batches: usize, + ) -> Result>> { self.inner.fetch_batches(max_batches) } @@ -766,7 +893,7 @@ pub struct RemotePendingFetch { pos_in_log_segment: i32, fetch_offset: i64, high_watermark: i64, - read_context: ReadContext, + resolver: Arc, } impl RemotePendingFetch { @@ -776,7 +903,7 @@ impl RemotePendingFetch { pos_in_log_segment: i32, fetch_offset: i64, high_watermark: i64, - read_context: ReadContext, + resolver: Arc, ) -> Self { Self { segment, @@ -784,7 +911,7 @@ impl RemotePendingFetch { pos_in_log_segment, fetch_offset, high_watermark, - read_context, + resolver, } } } @@ -836,7 +963,8 @@ impl PendingFetch for RemotePendingFetch { self.segment.table_bucket.clone(), log_record_batch, size_in_bytes, - self.read_context, + self.resolver, + true, // is_remote self.fetch_offset, self.high_watermark, ); @@ -852,6 +980,7 @@ impl PendingFetch for RemotePendingFetch { mod tests { use super::*; use crate::client::WriteRecord; + use crate::client::table::read_context_resolver::ReadContextResolver; use crate::compression::{ ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, @@ -862,16 +991,31 @@ mod tests { MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, ReadContext, to_arrow_schema, }; use crate::row::GenericRow; - use crate::test_utils::build_table_info; + use crate::test_utils::{build_table_info, build_table_info_with_columns}; use std::sync::Arc; - fn test_read_context() -> Result { + fn expect_data(result: FetchResult) -> T { + match result { + FetchResult::Data(data) => data, + FetchResult::SchemaRequired(schema_id) => { + panic!("unexpected missing schema {schema_id}") + } + } + } + + fn test_resolver() -> Result> { let row_type = RowType::new(vec![DataField::new("id", DataTypes::int(), None)]); - Ok(ReadContext::new( - to_arrow_schema(&row_type)?, - Arc::new(row_type), + let arrow_schema = to_arrow_schema(&row_type)?; + let row_type_arc = Arc::new(row_type); + let local_ctx = Arc::new(ReadContext::new( + arrow_schema.clone(), + row_type_arc.clone(), false, - )) + )); + let remote_ctx = Arc::new(ReadContext::new(arrow_schema, row_type_arc, true)); + Ok(Arc::new(ReadContextResolver::new( + 1, local_ctx, remote_ctx, None, + ))) } struct ErrorPendingFetch { @@ -897,7 +1041,7 @@ mod tests { #[tokio::test] async fn await_not_empty_returns_wakeup_error() { - let buffer = LogFetchBuffer::new(test_read_context().unwrap()); + let buffer = LogFetchBuffer::new(test_resolver().unwrap()); buffer.wakeup(); let result = buffer.await_not_empty(Duration::from_millis(10)).await; @@ -906,7 +1050,7 @@ mod tests { #[tokio::test] async fn await_not_empty_returns_pending_error() { - let buffer = LogFetchBuffer::new(test_read_context().unwrap()); + let buffer = LogFetchBuffer::new(test_resolver().unwrap()); let table_bucket = TableBucket::new(1, 0); buffer.pend(Box::new(ErrorPendingFetch { table_bucket: table_bucket.clone(), @@ -950,26 +1094,112 @@ mod tests { let data = builder.build()?; let log_records = LogRecordsBatches::new(data.clone()); - let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + let arrow_schema = to_arrow_schema(&row_type)?; + let row_type_arc = Arc::new(row_type); + let local_ctx = Arc::new(ReadContext::new( + arrow_schema.clone(), + row_type_arc.clone(), + false, + )); + let remote_ctx = Arc::new(ReadContext::new(arrow_schema, row_type_arc, true)); + let resolver = Arc::new(ReadContextResolver::new(1, local_ctx, remote_ctx, None)); let mut fetch = DefaultCompletedFetch::new( TableBucket::new(1, 0), log_records, data.len(), - read_context, + resolver, + false, 0, 0, ); - let records = fetch.fetch_records(10)?; + let records = expect_data(fetch.fetch_records(10)?); assert_eq!(records.len(), 1); assert_eq!(records[0].offset(), 0); - let empty = fetch.fetch_records(10)?; + let empty = expect_data(fetch.fetch_records(10)?); assert!(empty.is_empty()); Ok(()) } + #[test] + fn fixed_schema_fetch_batches_pads_missing_columns() -> Result<()> { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + let old_fields = vec![DataField::new("id", DataTypes::int(), None)]; + let new_fields = vec![ + DataField::new("id", DataTypes::int(), None), + DataField::new("name", DataTypes::string(), None), + ]; + let old_table_info = Arc::new(build_table_info_with_columns( + table_path.clone(), + 1, + 1, + old_fields, + )); + let old_row_type = old_table_info.get_row_type().clone(); + let new_table_info = build_table_info_with_columns(table_path.clone(), 1, 1, new_fields); + let new_row_type = new_table_info.get_row_type().clone(); + let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); + + let mut builder = MemoryLogRecordsArrowBuilder::new( + 0, + &old_row_type, + false, + ArrowCompressionInfo { + compression_type: ArrowCompressionType::None, + compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, + }, + usize::MAX, + Arc::new(ArrowCompressionRatioEstimator::default()), + )?; + + let mut row = GenericRow::new(1); + row.set_field(0, 1_i32); + let record = WriteRecord::for_append(old_table_info.clone(), physical_table_path, 1, &row); + builder.append(&record)?; + + let data = builder.build()?; + let new_arrow_schema = to_arrow_schema(&new_row_type)?; + let new_row_type_arc = Arc::new(new_row_type); + let local_ctx = Arc::new( + ReadContext::new(new_arrow_schema.clone(), new_row_type_arc.clone(), false) + .with_fluss_row_type(new_row_type_arc.clone()), + ); + let remote_ctx = Arc::new( + ReadContext::new(new_arrow_schema.clone(), new_row_type_arc.clone(), true) + .with_fluss_row_type(new_row_type_arc), + ); + let resolver = Arc::new( + ReadContextResolver::new(1, local_ctx, remote_ctx, None).with_fixed_schema(true), + ); + + let mut fetch = DefaultCompletedFetch::new( + TableBucket::new(1, 0), + LogRecordsBatches::new(data.clone()), + data.len(), + Arc::clone(&resolver), + false, + 0, + 0, + ); + + assert!(matches!( + fetch.fetch_batches(10)?, + FetchResult::SchemaRequired(0) + )); + resolver.register_schema(0, old_table_info.get_schema())?; + + let batches = expect_data(fetch.fetch_batches(10)?); + assert_eq!(batches.len(), 1); + let batch = &batches[0].0; + assert_eq!(batch.schema(), new_arrow_schema); + assert_eq!(batch.num_columns(), 2); + 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`). @@ -1018,25 +1248,36 @@ mod tests { let new_len = ((data.len() - LOG_OVERHEAD) as i32).to_le_bytes(); data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH].copy_from_slice(&new_len); - let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + let read_context = Arc::new(ReadContext::new( + to_arrow_schema(&row_type)?, + Arc::new(row_type.clone()), + false, + )); + let remote_ctx = Arc::new(ReadContext::new( + to_arrow_schema(&row_type)?, + Arc::new(row_type), + true, + )); + let resolver = Arc::new(ReadContextResolver::new(1, read_context, remote_ctx, None)); let mut fetch = DefaultCompletedFetch::new( TableBucket::new(1, 0), LogRecordsBatches::new(data.clone()), data.len(), - read_context, + resolver, + false, 0, 0, ); // max_records=2 cuts between -U (offset 1) and +U (offset 2); the pairing // guard pulls the +U so the batch ends on a complete pair. - let records = fetch.fetch_records(2)?; + let records = expect_data(fetch.fetch_records(2)?); assert_eq!(records.len(), 3, "pair guard should append the matching +U"); assert_eq!(*records[0].change_type(), ChangeType::Insert); assert_eq!(*records[1].change_type(), ChangeType::UpdateBefore); assert_eq!(*records[2].change_type(), ChangeType::UpdateAfter); - let rest = fetch.fetch_records(10)?; + let rest = expect_data(fetch.fetch_records(10)?); assert!(rest.is_empty(), "all records consumed"); Ok(()) diff --git a/fluss-rust/crates/fluss/src/client/table/mod.rs b/fluss-rust/crates/fluss/src/client/table/mod.rs index 657a44bfe8..41cbadab0b 100644 --- a/fluss-rust/crates/fluss/src/client/table/mod.rs +++ b/fluss-rust/crates/fluss/src/client/table/mod.rs @@ -30,6 +30,7 @@ mod lookup; mod log_fetch_buffer; mod partition_getter; +mod read_context_resolver; mod reader; mod remote_log; mod scanner; 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 new file mode 100644 index 0000000000..655891f49c --- /dev/null +++ b/fluss-rust/crates/fluss/src/client/table/read_context_resolver.rs @@ -0,0 +1,215 @@ +// 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. +//! +//! When fixed-schema mode is active, batches are still decoded with their +//! write-time schema and then aligned to the scanner creation schema. + +use crate::client::ClientSchemaGetter; +use crate::error::{Error, Result}; +use crate::metadata::{RowType, Schema}; +use crate::record::{ReadContext, to_arrow_schema}; +use arrow_schema::SchemaRef; +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>, + /// When Some, projection is active and schema is pinned to the initial one. + projected_fields: Option>, + /// Used to lazily fetch schema versions when decoding a batch whose schema + /// was not prewarmed by the fetch response path. + schema_getter: Option>, + /// 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, + fixed_target_row_type: Option>, +} + +/// A pair of ReadContexts for local and remote reads. +struct ResolvedContexts { + local: Arc, + remote: Arc, +} + +impl ReadContextResolver { + /// Create a new resolver with the initial schema's ReadContexts. + pub fn new( + initial_schema_id: i16, + local_context: Arc, + remote_context: Arc, + projected_fields: Option>, + ) -> 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, + schema_getter: None, + fixed_schema: false, + fixed_target_schema: None, + fixed_target_row_type: None, + } + } + + pub fn with_schema_getter(mut self, schema_getter: Arc) -> Self { + self.schema_getter = Some(schema_getter); + 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; + } + self + } + + /// 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> { + // 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 { + Arc::clone(&ctx.remote) + } else { + Arc::clone(&ctx.local) + } + }) + } + + /// Fetch and register one schema version without blocking a Tokio worker. + /// + /// Callers should invoke this only after encountering a concrete batch + /// whose schema is absent. This keeps remote log files streaming: no scan + /// of the remaining segment is needed to discover all schema IDs up front. + pub async fn fetch_and_register(&self, schema_id: i16) -> Result<()> { + if self.projected_fields.is_some() || self.contexts.read().contains_key(&schema_id) { + return Ok(()); + } + + let schema_getter = self + .schema_getter + .as_ref() + .ok_or_else(|| Error::UnexpectedError { + message: format!("No schema getter configured for schema_id {schema_id}"), + source: None, + })? + .clone(); + let schema = schema_getter.get_schema(schema_id as i32).await?; + self.register_schema(schema_id, &schema)?; + Ok(()) + } + + /// Register a new schema by its ID. Builds ReadContexts from the Schema. + /// No-op if already cached or if projection is active. + pub fn register_schema(&self, schema_id: i16, schema: &Schema) -> Result<()> { + if self.projected_fields.is_some() { + // Projection pins the schema, no need to register new ones + return Ok(()); + } + if self.contexts.read().contains_key(&schema_id) { + return Ok(()); + } + + 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 mut local_context = + ReadContext::new(source_arrow_schema.clone(), output_row_type.clone(), false) + .with_fluss_row_type(output_row_type.clone()); + let mut remote_context = + 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()); + } + } + + let local_context = Arc::new(local_context); + let remote_context = Arc::new(remote_context); + + self.contexts.write().insert( + schema_id, + ResolvedContexts { + local: local_context, + remote: remote_context, + }, + ); + Ok(()) + } + + /// Returns the initial schema ID. + pub fn initial_schema_id(&self) -> i16 { + self.initial_schema_id + } + + /// Returns the projection fields if active, used for fetch request pushdown. + #[allow(dead_code)] + pub fn projected_fields(&self) -> Option<&[usize]> { + self.projected_fields.as_deref() + } +} diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs index 287fcd06e3..aadd1cb869 100644 --- a/fluss-rust/crates/fluss/src/client/table/scanner.rs +++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs @@ -22,8 +22,9 @@ use crate::client::metadata::Metadata; use crate::client::table::batch_scanner::LimitBatchScanner; use crate::client::table::log_fetch_buffer::{ CompletedFetch, DefaultCompletedFetch, FetchErrorAction, FetchErrorContext, FetchErrorLogLevel, - LogFetchBuffer, RemotePendingFetch, + FetchResult, LogFetchBuffer, RemotePendingFetch, }; +use crate::client::table::read_context_resolver::ReadContextResolver; use crate::client::table::remote_log::{RemoteLogDownloader, RemoteLogFetchInfo}; use crate::config::Config; use crate::error::Error::UnsupportedOperation; @@ -64,6 +65,8 @@ pub struct TableScan<'a> { metadata: Arc, /// Column indices to project. None means all columns, Some(vec) means only the specified columns (non-empty). projected_fields: Option>, + /// Whether to align evolved schemas back to the scanner creation schema. + fixed_schema: bool, /// Optional row limit. When set, callers may construct a [`BatchScanner`] for a one-shot bounded scan. limit: Option, } @@ -75,10 +78,25 @@ impl<'a> TableScan<'a> { table_info, metadata, projected_fields: None, + fixed_schema: true, limit: None, } } + /// Controls how log scanners handle schema evolution. + /// + /// When enabled, batches written with older schemas are decoded with their + /// write-time schema and then aligned to the schema captured when the + /// scanner is created; missing columns are returned as nulls. This is the + /// default for both row and batch log scanners, ensuring that a scan exposes + /// one stable schema. When explicitly disabled, records and batches keep + /// their write-time schema and may have different column counts across + /// schema changes. + pub fn with_fixed_schema(mut self, fixed_schema: bool) -> Self { + self.fixed_schema = fixed_schema; + self + } + /// Sets a row limit for the scan, enabling [`Self::create_bucket_batch_scanner`]. /// /// The limit must be positive. A limit is incompatible with the log @@ -326,12 +344,15 @@ impl<'a> TableScan<'a> { pub fn create_log_scanner(self) -> Result { self.reject_limit("LogScanner")?; validate_scan_support_inner(&self.table_info.table_path, &self.table_info, true)?; + let admin = self.conn.get_admin()?; let inner = LogScannerInner::new( &self.table_info, self.metadata.clone(), self.conn.get_connections(), self.conn.config(), self.projected_fields, + self.fixed_schema, + admin, )?; Ok(LogScanner { inner: Arc::new(inner), @@ -347,12 +368,15 @@ impl<'a> TableScan<'a> { pub fn create_record_batch_log_scanner(self) -> Result { self.reject_limit("RecordBatchLogScanner")?; validate_scan_support(&self.table_info.table_path, &self.table_info)?; + let admin = self.conn.get_admin()?; let inner = LogScannerInner::new( &self.table_info, self.metadata.clone(), self.conn.get_connections(), self.conn.config(), self.projected_fields, + self.fixed_schema, + admin, )?; Ok(RecordBatchLogScanner { inner: Arc::new(inner), @@ -537,6 +561,8 @@ impl LogScannerInner { connections: Arc, config: &Config, projected_fields: Option>, + fixed_schema: bool, + admin: Arc, ) -> Result { let log_scanner_status = Arc::new(LogScannerStatus::new()); @@ -553,6 +579,15 @@ impl LogScannerInner { None => to_arrow_schema(full_row_type)?, }; + // Create schema getter for schema evolution support + let latest_schema = + SchemaInfo::new(table_info.get_schema().clone(), table_info.get_schema_id()); + let schema_getter = Arc::new(ClientSchemaGetter::new( + table_info.table_path.clone(), + admin, + latest_schema, + )); + let metrics = Arc::new(ScannerMetrics::new(&table_info.table_path)); let last_poll_unix_ms = Arc::new(AtomicI64::new(0)); let last_poll_seconds_ago_task = spawn_last_poll_seconds_ago_ticker( @@ -572,7 +607,9 @@ impl LogScannerInner { log_scanner_status.clone(), config, projected_fields, + fixed_schema, Arc::clone(&metrics), + schema_getter, )?, arrow_schema, reader_active: std::sync::atomic::AtomicBool::new(false), @@ -868,7 +905,7 @@ impl LogScannerInner { } async fn poll_for_fetches(&self) -> Result>> { - let result = self.log_fetcher.collect_fetches()?; + let result = self.log_fetcher.collect_fetches().await?; if !result.is_empty() { return Ok(result); } @@ -877,7 +914,7 @@ impl LogScannerInner { self.log_fetcher.send_fetches().await?; // Collect completed fetches from buffer - self.log_fetcher.collect_fetches() + self.log_fetcher.collect_fetches().await } async fn poll_batches(&self, timeout: Duration) -> Result> { @@ -912,13 +949,13 @@ impl LogScannerInner { } async fn poll_for_batches(&self) -> Result> { - let result = self.log_fetcher.collect_batches()?; + let result = self.log_fetcher.collect_batches().await?; if !result.is_empty() { return Ok(result); } self.log_fetcher.send_fetches().await?; - self.log_fetcher.collect_batches() + self.log_fetcher.collect_batches().await } } @@ -1122,8 +1159,7 @@ struct LogFetcher { table_path: TablePath, is_partitioned: bool, log_scanner_status: Arc, - read_context: ReadContext, - remote_read_context: ReadContext, + resolver: Arc, remote_log_downloader: Arc, /// Background security token manager for remote filesystem access. /// Kept alive to run the background refresh task; stopped on drop. @@ -1145,8 +1181,7 @@ struct FetchResponseContext { metadata: Arc, log_fetch_buffer: Arc, log_scanner_status: Arc, - read_context: ReadContext, - remote_read_context: ReadContext, + resolver: Arc, remote_log_downloader: Arc, /// Per-table scanner metric handles for `scanner.fetch_*` recording. metrics: Arc, @@ -1156,6 +1191,7 @@ struct FetchResponseContext { } impl LogFetcher { + #[allow(clippy::too_many_arguments)] pub fn new( table_info: TableInfo, conns: Arc, @@ -1163,7 +1199,9 @@ impl LogFetcher { log_scanner_status: Arc, config: &Config, projected_fields: Option>, + fixed_schema: bool, metrics: Arc, + schema_getter: Arc, ) -> Result { let full_row_type = table_info.get_row_type(); let full_arrow_schema = to_arrow_schema(full_row_type)?; @@ -1176,23 +1214,39 @@ impl LogFetcher { .collect(), )), }; - let read_context = Self::create_read_context( - full_arrow_schema.clone(), - projected_row_type.clone(), - projected_fields.clone(), - false, - )? - .with_fluss_row_type(projected_row_type.clone()); - let remote_read_context = Self::create_read_context( - full_arrow_schema, - projected_row_type.clone(), - projected_fields.clone(), - true, - )? - .with_fluss_row_type(projected_row_type); + let read_context = Arc::new( + Self::create_read_context( + full_arrow_schema.clone(), + projected_row_type.clone(), + projected_fields.clone(), + false, + )? + .with_fluss_row_type(projected_row_type.clone()), + ); + let remote_read_context = Arc::new( + Self::create_read_context( + full_arrow_schema, + projected_row_type.clone(), + projected_fields.clone(), + true, + )? + .with_fluss_row_type(projected_row_type), + ); + + 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 tmp_dir = TempDir::with_prefix("fluss-remote-logs")?; - let log_fetch_buffer = Arc::new(LogFetchBuffer::new(read_context.clone())); + let log_fetch_buffer = Arc::new(LogFetchBuffer::new(Arc::clone(&resolver))); // Create security token manager for background token refresh let security_token_manager = @@ -1219,8 +1273,7 @@ impl LogFetcher { table_path: table_info.table_path.clone(), is_partitioned: table_info.is_partitioned(), log_scanner_status, - read_context, - remote_read_context, + resolver, remote_log_downloader, security_token_manager, log_fetch_buffer, @@ -1393,8 +1446,7 @@ impl LogFetcher { let conns = Arc::clone(&self.conns); let log_fetch_buffer = self.log_fetch_buffer.clone(); let log_scanner_status = self.log_scanner_status.clone(); - let read_context = self.read_context.clone(); - let remote_read_context = self.remote_read_context.clone(); + let resolver = Arc::clone(&self.resolver); let remote_log_downloader = Arc::clone(&self.remote_log_downloader); let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone(); let metadata = self.metadata.clone(); @@ -1456,8 +1508,7 @@ impl LogFetcher { metadata: metadata.clone(), log_fetch_buffer, log_scanner_status, - read_context, - remote_read_context, + resolver, remote_log_downloader, metrics, request_start_time, @@ -1487,8 +1538,7 @@ impl LogFetcher { metadata, log_fetch_buffer, log_scanner_status, - read_context, - remote_read_context, + resolver, remote_log_downloader, metrics, request_start_time, @@ -1581,7 +1631,7 @@ impl LogFetcher { Self::pending_remote_fetches( remote_log_downloader.clone(), log_fetch_buffer.clone(), - remote_read_context.clone(), + Arc::clone(&resolver), &table_bucket, remote_fetch_info, fetch_offset, @@ -1592,13 +1642,14 @@ impl LogFetcher { let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1); let records = fetch_log_for_bucket.records.unwrap_or(vec![]); let size_in_bytes = records.len(); - let log_record_batch = LogRecordsBatches::new(records); + let log_record_batch = LogRecordsBatches::new(records); let completed_fetch = DefaultCompletedFetch::new( table_bucket.clone(), log_record_batch, size_in_bytes, - read_context.clone(), + Arc::clone(&resolver), + false, // is_remote fetch_offset, high_watermark, ); @@ -1611,7 +1662,7 @@ impl LogFetcher { fn pending_remote_fetches( remote_log_downloader: Arc, log_fetch_buffer: Arc, - read_context: ReadContext, + resolver: Arc, table_bucket: &TableBucket, remote_fetch_info: RemoteLogFetchInfo, fetch_offset: i64, @@ -1647,7 +1698,7 @@ impl LogFetcher { pos_in_log_segment, current_fetch_offset, high_watermark, - read_context.clone(), + Arc::clone(&resolver), ); // Add to pending fetches in buffer (similar to Java's logFetchBuffer.pend) log_fetch_buffer.pend(Box::new(pending_fetch)); @@ -1656,7 +1707,7 @@ impl LogFetcher { /// Collect completed fetches from buffer /// Reference: LogFetchCollector.collectFetch in Java - fn collect_fetches(&self) -> Result>> { + async fn collect_fetches(&self) -> Result>> { let mut result: HashMap> = HashMap::new(); let mut records_remaining = self.max_poll_records; @@ -1700,10 +1751,10 @@ impl LogFetcher { } else { // Fetch records from next_in_line if let Some(mut next_fetch) = next_in_line { - let records = match self + let fetch_result = match self .fetch_records_from_fetch(&mut next_fetch, records_remaining) { - Ok(records) => records, + Ok(fetch_result) => fetch_result, Err(e) => { if !next_fetch.is_consumed() { self.log_fetch_buffer @@ -1713,14 +1764,34 @@ impl LogFetcher { } }; - if !records.is_empty() { - let table_bucket = next_fetch.table_bucket().clone(); - // Merge with existing records for this bucket - let existing = result.entry(table_bucket).or_default(); - let records_count = records.len(); - existing.extend(records); + match fetch_result { + FetchResult::Data(records) => { + if !records.is_empty() { + let table_bucket = next_fetch.table_bucket().clone(); + // Merge with existing records for this bucket + let existing = result.entry(table_bucket).or_default(); + let records_count = records.len(); + existing.extend(records); + + records_remaining = + records_remaining.saturating_sub(records_count); + } + } + FetchResult::SchemaRequired(schema_id) => { + // Put the fetch back before awaiting. If this poll future is + // cancelled, the exact raw batch remains available for retry. + self.log_fetch_buffer + .set_next_in_line_fetch(Some(next_fetch)); + + // Never await after collecting user-visible records: dropping + // the future at that point would otherwise lose the local result. + if !result.is_empty() { + return Ok(result); + } - records_remaining = records_remaining.saturating_sub(records_count); + self.resolver.fetch_and_register(schema_id).await?; + continue; + } } // If the fetch is not fully consumed, put it back for the next round @@ -1839,7 +1910,7 @@ impl LogFetcher { &self, next_in_line_fetch: &mut Box, max_records: usize, - ) -> Result> { + ) -> Result>> { let table_bucket = next_in_line_fetch.table_bucket().clone(); let current_offset = self.log_scanner_status.get_bucket_offset(&table_bucket); @@ -1848,7 +1919,7 @@ impl LogFetcher { "Ignoring fetched records for {table_bucket:?} since the bucket has been unsubscribed" ); next_in_line_fetch.drain(); - return Ok(Vec::new()); + return Ok(FetchResult::Data(Vec::new())); } let current_offset = current_offset.unwrap(); @@ -1856,32 +1927,34 @@ impl LogFetcher { // Check if this fetch is next in line if fetch_offset == current_offset { - let records = next_in_line_fetch.fetch_records(max_records)?; - let next_fetch_offset = next_in_line_fetch.next_fetch_offset(); + let fetch_result = next_in_line_fetch.fetch_records(max_records)?; + if matches!(fetch_result, FetchResult::Data(_)) { + let next_fetch_offset = next_in_line_fetch.next_fetch_offset(); - if next_fetch_offset > current_offset { - self.log_scanner_status - .update_offset(&table_bucket, next_fetch_offset); - } + if next_fetch_offset > current_offset { + self.log_scanner_status + .update_offset(&table_bucket, next_fetch_offset); + } - if next_in_line_fetch.is_consumed() && next_in_line_fetch.records_read() > 0 { - self.log_scanner_status - .move_bucket_to_end(table_bucket.clone()); + if next_in_line_fetch.is_consumed() && next_in_line_fetch.records_read() > 0 { + self.log_scanner_status + .move_bucket_to_end(table_bucket.clone()); + } } - Ok(records) + Ok(fetch_result) } else { // These records aren't next in line, ignore them warn!( "Ignoring fetched records for {table_bucket:?} at offset {fetch_offset} since the current offset is {current_offset}" ); next_in_line_fetch.drain(); - Ok(Vec::new()) + Ok(FetchResult::Data(Vec::new())) } } /// Collect completed fetches as ScanBatches (with bucket and offset metadata) - fn collect_batches(&self) -> Result> { + async fn collect_batches(&self) -> Result> { // Limit memory usage with both batch count and byte size constraints. // Max 100 batches per poll, but also check total bytes (soft cap ~64MB). const MAX_BATCHES: usize = 100; @@ -1896,20 +1969,39 @@ impl LogFetcher { match next_in_line { Some(mut next_fetch) if !next_fetch.is_consumed() => { - let scan_batches = + let fetch_result = self.fetch_batches_from_fetch(&mut next_fetch, batches_remaining)?; - let batch_count = scan_batches.len(); - - if !scan_batches.is_empty() { - // Track bytes consumed (soft cap - may exceed by one fetch) - let batch_bytes: usize = scan_batches - .iter() - .map(|sb| sb.batch().get_array_memory_size()) - .sum(); - bytes_consumed += batch_bytes; - - result.extend(scan_batches); - batches_remaining = batches_remaining.saturating_sub(batch_count); + match fetch_result { + FetchResult::Data(scan_batches) => { + let batch_count = scan_batches.len(); + + if !scan_batches.is_empty() { + // Track bytes consumed (soft cap - may exceed by one fetch) + let batch_bytes: usize = scan_batches + .iter() + .map(|sb| sb.batch().get_array_memory_size()) + .sum(); + bytes_consumed += batch_bytes; + + result.extend(scan_batches); + batches_remaining = + batches_remaining.saturating_sub(batch_count); + } + } + FetchResult::SchemaRequired(schema_id) => { + // Preserve the current file-backed batch across await/cancel. + self.log_fetch_buffer + .set_next_in_line_fetch(Some(next_fetch)); + + // Return already decoded batches before doing another async RPC, + // keeping cancellation from discarding user-visible progress. + if !result.is_empty() { + return Ok(result); + } + + self.resolver.fetch_and_register(schema_id).await?; + continue; + } } if !next_fetch.is_consumed() { @@ -1962,7 +2054,7 @@ impl LogFetcher { &self, next_in_line_fetch: &mut Box, max_batches: usize, - ) -> Result> { + ) -> Result>> { let table_bucket = next_in_line_fetch.table_bucket().clone(); let current_offset = self.log_scanner_status.get_bucket_offset(&table_bucket); @@ -1971,34 +2063,42 @@ impl LogFetcher { "Ignoring fetched batches for {table_bucket:?} since the bucket has been unsubscribed" ); next_in_line_fetch.drain(); - return Ok(Vec::new()); + return Ok(FetchResult::Data(Vec::new())); } let current_offset = current_offset.unwrap(); let fetch_offset = next_in_line_fetch.next_fetch_offset(); if fetch_offset == current_offset { - let batches_with_offsets = next_in_line_fetch.fetch_batches(max_batches)?; - let next_fetch_offset = next_in_line_fetch.next_fetch_offset(); + match next_in_line_fetch.fetch_batches(max_batches)? { + FetchResult::Data(batches_with_offsets) => { + let next_fetch_offset = next_in_line_fetch.next_fetch_offset(); - if next_fetch_offset > current_offset { - self.log_scanner_status - .update_offset(&table_bucket, next_fetch_offset); - } + if next_fetch_offset > current_offset { + self.log_scanner_status + .update_offset(&table_bucket, next_fetch_offset); + } - // Convert to ScanBatch with bucket info - Ok(batches_with_offsets - .into_iter() - .map(|(batch, base_offset)| { - ScanBatch::new(table_bucket.clone(), batch, base_offset) - }) - .collect()) + // Convert to ScanBatch with bucket info + Ok(FetchResult::Data( + batches_with_offsets + .into_iter() + .map(|(batch, base_offset)| { + ScanBatch::new(table_bucket.clone(), batch, base_offset) + }) + .collect(), + )) + } + FetchResult::SchemaRequired(schema_id) => { + Ok(FetchResult::SchemaRequired(schema_id)) + } + } } else { warn!( "Ignoring fetched batches for {table_bucket:?} at offset {fetch_offset} since the current offset is {current_offset}" ); next_in_line_fetch.drain(); - Ok(Vec::new()) + Ok(FetchResult::Data(Vec::new())) } } @@ -2057,11 +2157,15 @@ impl LogFetcher { if ready_for_fetch_count == 0 { HashMap::new() } else { - let (projection_enabled, projected_fields) = - match self.read_context.project_fields_in_order() { - None => (false, vec![]), - Some(fields) => (true, fields.iter().map(|&i| i as i32).collect()), - }; + let initial_ctx = self + .resolver + .resolve(self.resolver.initial_schema_id(), false) + .expect("initial ReadContext must exist"); + let (projection_enabled, projected_fields) = match initial_ctx.project_fields_in_order() + { + None => (false, vec![]), + Some(fields) => (true, fields.iter().map(|&i| i as i32).collect()), + }; fetch_log_req_for_buckets .into_iter() @@ -2280,7 +2384,9 @@ fn validate_scan_support_inner( mod tests { use super::*; use crate::client::WriteRecord; + use crate::client::admin::FlussAdmin; use crate::client::metadata::Metadata; + use crate::client::table::read_context_resolver::ReadContextResolver; use crate::compression::{ ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, @@ -2294,6 +2400,45 @@ mod tests { assert_scanner_entries_labeled, build_cluster_arc, build_table_info, test_scanner_metrics, }; + fn test_admin(metadata: &Arc) -> Arc { + Arc::new(FlussAdmin::new( + Arc::new(RpcClient::new()), + metadata.clone(), + )) + } + + fn test_schema_getter( + table_info: &TableInfo, + metadata: &Arc, + ) -> Arc { + let latest = SchemaInfo::new(table_info.get_schema().clone(), table_info.get_schema_id()); + Arc::new(ClientSchemaGetter::new( + table_info.table_path.clone(), + test_admin(metadata), + latest, + )) + } + + fn test_resolver(table_info: &TableInfo) -> Arc { + let row_type = table_info.get_row_type(); + let arrow_schema = to_arrow_schema(row_type).unwrap(); + let row_type_arc = Arc::new(row_type.clone()); + let local_ctx = Arc::new( + ReadContext::new(arrow_schema.clone(), row_type_arc.clone(), false) + .with_fluss_row_type(row_type_arc.clone()), + ); + let remote_ctx = Arc::new( + ReadContext::new(arrow_schema, row_type_arc.clone(), true) + .with_fluss_row_type(row_type_arc), + ); + Arc::new(ReadContextResolver::new( + table_info.get_schema_id() as i16, + local_ctx, + remote_ctx, + None, + )) + } + fn build_records(table_info: &TableInfo, table_path: Arc) -> Result> { let mut builder = MemoryLogRecordsArrowBuilder::new( 1, @@ -2326,11 +2471,13 @@ mod tests { let fetcher = LogFetcher::new( table_info.clone(), Arc::new(RpcClient::new()), - metadata, + metadata.clone(), status.clone(), &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; let bucket = TableBucket::new(1, 0); @@ -2338,13 +2485,19 @@ mod tests { let data = build_records(&table_info, Arc::new(table_path))?; let log_records = LogRecordsBatches::new(data.clone()); - let row_type = Arc::new(table_info.get_row_type().clone()); - let read_context = ReadContext::new(to_arrow_schema(&row_type)?, row_type, false); - let completed = - DefaultCompletedFetch::new(bucket.clone(), log_records, data.len(), read_context, 0, 0); + let resolver = test_resolver(&table_info); + let completed = DefaultCompletedFetch::new( + bucket.clone(), + log_records, + data.len(), + resolver, + false, + 0, + 0, + ); fetcher.log_fetch_buffer.add(Box::new(completed)); - let fetched = fetcher.collect_fetches()?; + let fetched = fetcher.collect_fetches().await?; assert_eq!(fetched.get(&bucket).unwrap().len(), 1); assert_eq!(status.get_bucket_offset(&bucket), Some(1)); Ok(()) @@ -2360,29 +2513,31 @@ mod tests { let fetcher = LogFetcher::new( table_info.clone(), Arc::new(RpcClient::new()), - metadata, + metadata.clone(), status, &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; let bucket = TableBucket::new(1, 0); let data = build_records(&table_info, Arc::new(table_path))?; let log_records = LogRecordsBatches::new(data.clone()); - let row_type = Arc::new(table_info.get_row_type().clone()); - let read_context = ReadContext::new(to_arrow_schema(&row_type)?, row_type, false); + let resolver = test_resolver(&table_info); let mut completed: Box = Box::new(DefaultCompletedFetch::new( bucket, log_records, data.len(), - read_context, + resolver, + false, 0, 0, )); let records = fetcher.fetch_records_from_fetch(&mut completed, 10)?; - assert!(records.is_empty()); + assert!(matches!(records, FetchResult::Data(records) if records.is_empty())); assert!(completed.is_consumed()); Ok(()) } @@ -2396,13 +2551,15 @@ mod tests { let status = Arc::new(LogScannerStatus::new()); status.assign_scan_bucket(TableBucket::new(1, 0), 0); let fetcher = LogFetcher::new( - table_info, + table_info.clone(), Arc::new(RpcClient::new()), - metadata, + metadata.clone(), status, &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; fetcher.nodes_with_pending_fetch_requests.lock().insert(1); @@ -2427,7 +2584,9 @@ mod tests { status.clone(), &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; let response = FetchLogResponse { @@ -2451,8 +2610,7 @@ mod tests { metadata: metadata.clone(), log_fetch_buffer: fetcher.log_fetch_buffer.clone(), log_scanner_status: fetcher.log_scanner_status.clone(), - read_context: fetcher.read_context.clone(), - remote_read_context: fetcher.remote_read_context.clone(), + resolver: Arc::clone(&fetcher.resolver), remote_log_downloader: fetcher.remote_log_downloader.clone(), metrics: Arc::clone(&fetcher.metrics), request_start_time: Instant::now(), @@ -2481,7 +2639,9 @@ mod tests { status.clone(), &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; let bucket = TableBucket::new(1, 0); @@ -2508,8 +2668,7 @@ mod tests { metadata: metadata.clone(), log_fetch_buffer: fetcher.log_fetch_buffer.clone(), log_scanner_status: fetcher.log_scanner_status.clone(), - read_context: fetcher.read_context.clone(), - remote_read_context: fetcher.remote_read_context.clone(), + resolver: Arc::clone(&fetcher.resolver), remote_log_downloader: fetcher.remote_log_downloader.clone(), metrics: Arc::clone(&fetcher.metrics), request_start_time: Instant::now(), @@ -2608,13 +2767,15 @@ mod tests { }; let fetcher = LogFetcher::new( - table_info, + table_info.clone(), Arc::new(RpcClient::new()), - metadata, + metadata.clone(), status, &config, None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), )?; let requests = fetcher.prepare_fetch_log_requests().await; @@ -2648,12 +2809,19 @@ mod tests { let table_info = build_table_info(table_path.clone(), 1, 1); let cluster = build_cluster_arc(&table_path, 1, 1); let metadata = Arc::new(Metadata::new_for_test(cluster)); + let rpc_client = Arc::new(RpcClient::new()); + let admin = Arc::new(crate::client::admin::FlussAdmin::new( + rpc_client.clone(), + metadata.clone(), + )); let inner = LogScannerInner::new( &table_info, metadata, - Arc::new(RpcClient::new()), + rpc_client, &Config::default(), None, + false, + admin, ) .expect("build LogScannerInner"); body(&inner); @@ -2813,13 +2981,15 @@ mod tests { let status = Arc::new(LogScannerStatus::new()); status.assign_scan_bucket(TableBucket::new(1, 0), 5); let fetcher = LogFetcher::new( - table_info, + table_info.clone(), Arc::new(RpcClient::new()), metadata.clone(), status, &Config::default(), None, + false, test_scanner_metrics(&table_path), + test_schema_getter(&table_info, &metadata), ) .expect("build LogFetcher"); @@ -2844,8 +3014,7 @@ mod tests { metadata: metadata.clone(), log_fetch_buffer: fetcher.log_fetch_buffer.clone(), log_scanner_status: fetcher.log_scanner_status.clone(), - read_context: fetcher.read_context.clone(), - remote_read_context: fetcher.remote_read_context.clone(), + resolver: Arc::clone(&fetcher.resolver), remote_log_downloader: fetcher.remote_log_downloader.clone(), metrics: Arc::clone(&fetcher.metrics), request_start_time: Instant::now(), @@ -3156,12 +3325,19 @@ mod tests { let table_info = build_table_info(table_path.clone(), 1, 1); let cluster = build_cluster_arc(&table_path, 1, 1); let metadata = Arc::new(Metadata::new_for_test(cluster)); + let rpc_client = Arc::new(RpcClient::new()); + let admin = Arc::new(crate::client::admin::FlussAdmin::new( + rpc_client.clone(), + metadata.clone(), + )); let inner = LogScannerInner::new( &table_info, metadata, - Arc::new(RpcClient::new()), + rpc_client, &Config::default(), None, + false, + admin, ) .expect("build LogScannerInner"); diff --git a/fluss-rust/crates/fluss/src/record/arrow.rs b/fluss-rust/crates/fluss/src/record/arrow.rs index 025fe0d625..dbbbcc991c 100644 --- a/fluss-rust/crates/fluss/src/record/arrow.rs +++ b/fluss-rust/crates/fluss/src/record/arrow.rs @@ -25,7 +25,7 @@ use crate::record::{ChangeType, ScanRecord}; use crate::row::column_vector::TypedBatch; use crate::row::column_writer::{ColumnWriter, round_up_to_8}; use crate::row::{ColumnarRow, InternalRow}; -use arrow::array::{ArrayBuilder, ArrayRef}; +use arrow::array::{ArrayBuilder, ArrayRef, new_null_array}; use arrow::{ array::RecordBatch, buffer::Buffer, @@ -1440,6 +1440,7 @@ pub struct ReadContext { projection: Option, is_from_remote: bool, fluss_row_type: Option>, + align_to_target_schema: bool, } #[derive(Clone)] @@ -1465,6 +1466,7 @@ impl ReadContext { projection: None, is_from_remote, fluss_row_type: None, + align_to_target_schema: false, } } @@ -1477,6 +1479,24 @@ impl ReadContext { self.fluss_row_type.as_ref() } + pub(crate) fn target_schema(&self) -> SchemaRef { + self.target_schema.clone() + } + + pub(crate) fn row_type_arc(&self) -> Arc { + self.row_type.clone() + } + + pub(crate) fn with_target_schema_alignment(mut self, target_schema: SchemaRef) -> ReadContext { + debug_assert!( + self.projection.is_none(), + "target schema alignment is not supported with projection" + ); + self.target_schema = target_schema; + self.align_to_target_schema = true; + self + } + pub fn with_projection_pushdown( arrow_schema: SchemaRef, row_type: Arc, @@ -1548,6 +1568,7 @@ impl ReadContext { projection: Some(project), is_from_remote, fluss_row_type: None, + align_to_target_schema: false, }) } @@ -1590,7 +1611,7 @@ impl ReadContext { let resolve_schema = { // if from remote, no projection, need to use full schema - if self.is_from_remote { + if self.is_from_remote || self.align_to_target_schema { self.full_schema.clone() } else { // the record batch from server must be ordered by field pos, @@ -1643,6 +1664,11 @@ 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 + }; Ok(record_batch) } @@ -1669,10 +1695,51 @@ 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 + }; Ok(Some(record_batch)) } } +fn align_record_batch_to_schema( + record_batch: RecordBatch, + target_schema: SchemaRef, +) -> Result { + 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)), + } + } + + Ok(RecordBatch::try_new(target_schema, columns)?) +} + pub enum LogRecordIterator { Empty, Arrow(ArrowLogRecordIterator), diff --git a/fluss-rust/crates/fluss/tests/integration/log_table.rs b/fluss-rust/crates/fluss/tests/integration/log_table.rs index 625f48ac40..d50d8bc58b 100644 --- a/fluss-rust/crates/fluss/tests/integration/log_table.rs +++ b/fluss-rust/crates/fluss/tests/integration/log_table.rs @@ -26,13 +26,16 @@ mod table_test { }; use arrow::array::record_batch; use fluss::client::{EARLIEST_OFFSET, FlussTable, TableScan}; - use fluss::metadata::{DataField, DataTypes, Schema, TableDescriptor, TablePath}; + use fluss::metadata::{ + AddColumn, AlterTableChanges, ColumnPositionType, DataField, DataTypes, JsonSerde, Schema, + TableDescriptor, TablePath, + }; use fluss::record::ScanRecord; use fluss::row::binary_array::FlussArrayWriter; use fluss::row::binary_map::FlussMapWriter; use fluss::row::{ - DataGetters, Date, Datum, Decimal, GenericRow, InternalArray, InternalMap, Time, - TimestampLtz, TimestampNtz, + DataGetters, Date, Datum, Decimal, GenericRow, InternalArray, InternalMap, InternalRow, + Time, TimestampLtz, TimestampNtz, }; use fluss::rpc::message::OffsetSpec; use std::collections::HashMap; @@ -1949,4 +1952,187 @@ mod table_test { admin.drop_table(&table_path, false).await.expect("drop"); } + + #[tokio::test] + async fn schema_evolution_add_column_log_scanner_dynamic_schema() { + run_schema_evolution_add_column_log_scanner(false).await; + } + + #[tokio::test] + async fn schema_evolution_add_column_log_scanner_fixed_schema() { + run_schema_evolution_add_column_log_scanner(true).await; + } + + /// Test that a single log scanner can read records across a schema change + /// (add column) in both dynamic-schema and fixed-schema modes. + async fn run_schema_evolution_add_column_log_scanner(fixed_schema: bool) { + let cluster = get_shared_cluster(); + let connection = cluster.get_fluss_connection().await; + let admin = connection.get_admin().expect("Failed to get admin"); + + let table_path = TablePath::new( + "fluss", + format!("test_schema_evolution_log_scanner_fixed_{fixed_schema}"), + ); + + // 1. Create table with initial schema: (id INT, name STRING) + let table_descriptor = TableDescriptor::builder() + .schema( + Schema::builder() + .column("id", DataTypes::int()) + .column("name", DataTypes::string()) + .build() + .expect("Failed to build schema"), + ) + .build() + .expect("Failed to build table"); + + create_table(&admin, &table_path, &table_descriptor).await; + wait_for_table_ready(&admin, &table_path).await; + + // 2. Get table handle and create scanner + subscribe from EARLIEST + let table = connection + .get_table(&table_path) + .await + .expect("Failed to get table"); + let log_scanner = table + .new_scan() + .with_fixed_schema(fixed_schema) + .create_log_scanner() + .expect("Failed to create log scanner"); + let num_buckets = table.get_table_info().get_num_buckets(); + for bucket_id in 0..num_buckets { + log_scanner + .subscribe(bucket_id, EARLIEST_OFFSET) + .await + .expect("Failed to subscribe"); + } + + // 3. Write records with old schema + let writer_v0 = table + .new_append() + .expect("append") + .create_writer() + .expect("writer"); + writer_v0 + .append_arrow_batch( + record_batch!( + ("id", Int32, [1, 2, 3]), + ("name", Utf8, ["alice", "bob", "charlie"]) + ) + .unwrap(), + ) + .expect("Failed to append old-schema batch"); + writer_v0.flush().await.expect("flush"); + + // 4. Poll old-schema records + let mut old_records: Vec<(i32, String)> = Vec::new(); + let start = std::time::Instant::now(); + while old_records.len() < 3 && start.elapsed() < Duration::from_secs(10) { + let recs = log_scanner + .poll(Duration::from_millis(500)) + .await + .expect("poll"); + for rec in recs { + let row = rec.row(); + old_records.push(( + row.get_int(0).unwrap(), + row.get_string(1).unwrap().to_string(), + )); + } + } + assert_eq!(old_records.len(), 3, "Expected 3 old-schema records"); + old_records.sort_by_key(|r| r.0); + assert_eq!( + old_records, + vec![ + (1, "alice".to_string()), + (2, "bob".to_string()), + (3, "charlie".to_string()), + ] + ); + + // 5. Alter table: add column "age INT" + let age_type_json = serde_json::to_vec( + &DataTypes::int() + .serialize_json() + .expect("serialize INT type"), + ) + .expect("to_vec"); + admin + .alter_table( + &table_path, + false, + AlterTableChanges { + add_columns: vec![AddColumn { + column_name: "age".to_string(), + data_type_json: age_type_json, + comment: None, + position: ColumnPositionType::Last, + }], + ..Default::default() + }, + ) + .await + .expect("Failed to alter table"); + + // 6. Get a new table handle with the updated schema and write new records + let table_v1 = connection + .get_table(&table_path) + .await + .expect("Failed to get updated table"); + let writer_v1 = table_v1 + .new_append() + .expect("append") + .create_writer() + .expect("writer"); + writer_v1 + .append_arrow_batch( + record_batch!( + ("id", Int32, [4, 5, 6]), + ("name", Utf8, ["dave", "eve", "frank"]), + ("age", Int32, [30, 25, 40]) + ) + .unwrap(), + ) + .expect("Failed to append new-schema batch"); + writer_v1.flush().await.expect("flush"); + + // 7. Continue reading from the SAME scanner — it should handle schema + // evolution and return the new-schema records with the extra column. + let mut new_records: Vec<(i32, String, Option)> = Vec::new(); + let start = std::time::Instant::now(); + while new_records.len() < 3 && start.elapsed() < Duration::from_secs(10) { + let recs = log_scanner + .poll(Duration::from_millis(500)) + .await + .expect("poll"); + for rec in recs { + let row = rec.row(); + let id = row.get_int(0).unwrap(); + let name = row.get_string(1).unwrap().to_string(); + let age = if row.get_field_count() > 2 && !row.is_null_at(2).unwrap_or(true) { + Some(row.get_int(2).unwrap()) + } else { + None + }; + new_records.push((id, name, age)); + } + } + assert_eq!(new_records.len(), 3, "Expected 3 new-schema records"); + new_records.sort_by_key(|r| r.0); + assert_eq!( + new_records, + vec![ + (4, "dave".to_string(), (!fixed_schema).then_some(30),), + (5, "eve".to_string(), (!fixed_schema).then_some(25)), + (6, "frank".to_string(), (!fixed_schema).then_some(40),), + ] + ); + + admin + .drop_table(&table_path, false) + .await + .expect("Failed to drop table"); + } }