diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index 4a6971583cf..fa762da93e3 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -923,9 +923,9 @@ message PbFetchLogRespForBucket { optional int64 log_start_offset = 6; // TODO now we don't introduce log start offset, but remain it in protobuf optional PbRemoteLogFetchInfo remote_log_fetch_info = 7; optional bytes records = 8; - // The next fetch offset when all batches in this bucket are filtered out by server-side - // filter. When set (>= 0), the records field MUST be empty — the two fields are mutually - // exclusive. The client should use this offset as the starting offset for the next fetch. + // The offset the server scanned up to while filtering out batches, which may accompany a + // non-empty records field when only trailing batches were filtered. The client should start + // its next fetch from the later of this offset and the end of the records it received. optional int64 filtered_end_offset = 9; } 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 b057612d6e3..6d3a2ad4a62 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 @@ -354,6 +354,9 @@ impl PendingFetch for CompletedPendingFetch { } } +/// Sentinel for a fetch whose response carried no `filtered_end_offset`. +pub(crate) const NO_FILTERED_END_OFFSET: i64 = -1; + /// Default implementation of CompletedFetch for in-memory log records /// Used for local fetches from tablet server pub struct DefaultCompletedFetch { @@ -365,6 +368,9 @@ pub struct DefaultCompletedFetch { resolver: Arc, is_remote: bool, next_fetch_offset: i64, + /// Offset the server scanned up to while pruning batches by their + /// statistics, or [`NO_FILTERED_END_OFFSET`] when it sent no filtered range. + filtered_end_offset: i64, high_watermark: i64, size_in_bytes: usize, consumed: bool, @@ -405,6 +411,7 @@ impl DefaultCompletedFetch { resolver, is_remote, next_fetch_offset: fetch_offset, + filtered_end_offset: NO_FILTERED_END_OFFSET, high_watermark, size_in_bytes, consumed: false, @@ -435,6 +442,7 @@ impl DefaultCompletedFetch { resolver, is_remote: false, next_fetch_offset: fetch_offset, + filtered_end_offset: NO_FILTERED_END_OFFSET, high_watermark: -1, size_in_bytes: 0, consumed: false, @@ -466,6 +474,7 @@ impl DefaultCompletedFetch { resolver, is_remote: false, next_fetch_offset: fetch_offset, + filtered_end_offset: NO_FILTERED_END_OFFSET, high_watermark: -1, size_in_bytes: 0, consumed: false, @@ -481,6 +490,23 @@ impl DefaultCompletedFetch { } } + /// Records how far the server scanned while pruning batches, which it + /// reports both for a fetch it emptied and for one whose tail it pruned. + /// + /// `next_fetch_offset` is left alone so the fetch still passes the + /// next-in-line check; it jumps past the range once drained. + pub(crate) fn with_filtered_end_offset(mut self, filtered_end_offset: i64) -> Self { + debug_assert!( + filtered_end_offset == NO_FILTERED_END_OFFSET + || filtered_end_offset >= self.next_fetch_offset, + "filtered_end_offset ({filtered_end_offset}) must be >= fetch offset ({}) for bucket {}", + self.next_fetch_offset, + self.table_bucket + ); + self.filtered_end_offset = filtered_end_offset; + self + } + /// Get the next fetched record, handling batch iteration and record skipping fn next_fetched_record(&mut self) -> Result> { loop { @@ -498,6 +524,7 @@ impl DefaultCompletedFetch { if let Some(batch) = self.current_record_batch.take() { self.next_fetch_offset = batch.next_log_offset(); } + self.skip_filtered_range(); self.drain(); return Ok(FetchStep::End); }; @@ -588,6 +615,7 @@ impl DefaultCompletedFetch { loop { if self.pending_record_batch.is_none() { let Some(log_batch_result) = self.log_record_batch.next() else { + self.skip_filtered_range(); self.drain(); return Ok(FetchStep::End); }; @@ -633,6 +661,14 @@ impl DefaultCompletedFetch { } } + /// Advances past the range the server scanned and pruned, guarding on the + /// sentinel since an unfetched bucket sits at `EARLIEST_OFFSET` (-2) < -1. + fn skip_filtered_range(&mut self) { + if self.filtered_end_offset != NO_FILTERED_END_OFFSET { + self.next_fetch_offset = self.next_fetch_offset.max(self.filtered_end_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(); @@ -978,8 +1014,8 @@ impl PendingFetch for RemotePendingFetch { #[cfg(test)] mod tests { use super::*; - use crate::client::WriteRecord; use crate::client::table::read_context_resolver::ReadContextResolver; + use crate::client::{EARLIEST_OFFSET, WriteRecord}; use crate::compression::{ ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, @@ -1237,6 +1273,151 @@ mod tests { Ok(()) } + /// An empty fetch carrying only a filtered range, as the server sends when + /// it prunes every batch it scanned. + fn filtered_empty_fetch( + fetch_offset: i64, + filtered_end_offset: i64, + ) -> Result { + Ok(DefaultCompletedFetch::new( + TableBucket::new(1, 0), + LogRecordsBatches::new(Vec::new()), + 0, + test_resolver()?, + false, + fetch_offset, + 9, + ) + .with_filtered_end_offset(filtered_end_offset)) + } + + #[test] + fn filtered_empty_fetch_advances_past_the_filtered_range() -> Result<()> { + let mut fetch = filtered_empty_fetch(4, 7)?; + + // Stays put until drained so the fetch still passes the next-in-line check. + assert_eq!(fetch.next_fetch_offset(), 4); + assert_eq!(fetch.high_watermark(), 9); + + let records = expect_data(fetch.fetch_records(10)?); + assert!(records.is_empty()); + assert_eq!(fetch.next_fetch_offset(), 7); + assert!(fetch.is_consumed()); + assert_eq!(fetch.records_read(), 0); + Ok(()) + } + + #[test] + fn filtered_empty_fetch_advances_the_batch_path_too() -> Result<()> { + let mut fetch = filtered_empty_fetch(4, 7)?; + + let batches = expect_data(fetch.fetch_batches(10)?); + assert!(batches.is_empty()); + assert_eq!(fetch.next_fetch_offset(), 7); + assert!(fetch.is_consumed()); + Ok(()) + } + + /// The server also reports a filtered range alongside records when it prunes + /// only the tail of what it scanned. + #[test] + fn fetch_with_records_still_skips_a_pruned_tail() -> Result<()> { + let row_type = RowType::new(vec![ + DataField::new("id", DataTypes::int(), None), + DataField::new("name", DataTypes::string(), None), + ]); + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1)); + let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); + + let mut builder = MemoryLogRecordsArrowBuilder::new( + 1, + &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(2); + row.set_field(0, 1_i32); + row.set_field(1, "alice"); + builder.append(&WriteRecord::for_append( + table_info, + physical_table_path, + 1, + &row, + ))?; + let data = builder.build()?; + + 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), + LogRecordsBatches::new(data.clone()), + data.len(), + resolver, + false, + 0, + 9, + ) + .with_filtered_end_offset(5); + + let records = expect_data(fetch.fetch_records(10)?); + assert_eq!(records.len(), 1); + assert_eq!(records[0].offset(), 0); + // The record ends at offset 1, but the server already scanned through 5. + assert_eq!(fetch.next_fetch_offset(), 5); + Ok(()) + } + + #[test] + fn fetch_without_a_filtered_range_keeps_its_offset() -> Result<()> { + let mut fetch = DefaultCompletedFetch::new( + TableBucket::new(1, 0), + LogRecordsBatches::new(Vec::new()), + 0, + test_resolver()?, + false, + 4, + 9, + ); + + let records = expect_data(fetch.fetch_records(10)?); + assert!(records.is_empty()); + assert_eq!(fetch.next_fetch_offset(), 4); + Ok(()) + } + + /// A bucket subscribed at [`EARLIEST_OFFSET`] (-2) still holds that sentinel + /// when its first fetch is drained, and -1 must not outrank it. + #[test] + fn fetch_without_a_filtered_range_keeps_a_sentinel_offset() -> Result<()> { + let mut fetch = DefaultCompletedFetch::new( + TableBucket::new(1, 0), + LogRecordsBatches::new(Vec::new()), + 0, + test_resolver()?, + false, + EARLIEST_OFFSET, + 9, + ); + + let records = expect_data(fetch.fetch_records(10)?); + assert!(records.is_empty()); + assert_eq!(fetch.next_fetch_offset(), EARLIEST_OFFSET); + Ok(()) + } + #[test] fn fixed_schema_fetch_batches_pads_missing_columns() -> Result<()> { let table_path = TablePath::new("db".to_string(), "tbl".to_string()); diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs index 6e4d046e7cb..eaa978982cc 100644 --- a/fluss-rust/crates/fluss/src/client/table/scanner.rs +++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs @@ -22,7 +22,7 @@ use crate::client::metadata::Metadata; use crate::client::table::batch_scanner::LimitBatchScanner; use crate::client::table::log_fetch_buffer::{ CompletedFetch, DefaultCompletedFetch, FetchErrorAction, FetchErrorContext, FetchErrorLogLevel, - FetchResult, LogFetchBuffer, RemotePendingFetch, + FetchResult, LogFetchBuffer, NO_FILTERED_END_OFFSET, RemotePendingFetch, }; use crate::client::table::read_context_resolver::ReadContextResolver; use crate::client::table::remote_log::{RemoteLogDownloader, RemoteLogFetchInfo}; @@ -33,8 +33,10 @@ use crate::metadata::{ LogFormat, PhysicalTablePath, RowType, SchemaInfo, TableBucket, TableInfo, TablePath, }; use crate::metrics::ScannerMetrics; +use crate::predicate::{Predicate, to_pb_predicate}; use crate::proto::{ - ErrorResponse, FetchLogRequest, FetchLogResponse, PbFetchLogReqForBucket, PbFetchLogReqForTable, + ErrorResponse, FetchLogRequest, FetchLogResponse, PbFetchLogReqForBucket, + PbFetchLogReqForTable, PbPredicate, }; use crate::record::{ LogRecordsBatches, ReadContext, ScanBatch, ScanRecord, ScanRecords, to_arrow_schema, @@ -69,6 +71,9 @@ pub struct TableScan<'a> { fixed_schema: bool, /// Optional row limit. When set, callers may construct a [`BatchScanner`] for a one-shot bounded scan. limit: Option, + /// Filter pushed down to the server, encoded eagerly so that an unresolvable + /// column is reported by [`Self::filter`] rather than at scanner creation. + filter: Option, } impl<'a> TableScan<'a> { @@ -80,6 +85,7 @@ impl<'a> TableScan<'a> { projected_fields: None, fixed_schema: true, limit: None, + filter: None, } } @@ -111,6 +117,36 @@ impl<'a> TableScan<'a> { Ok(self) } + /// Pushes `predicate` down to the log scanners, which skip whole record + /// batches whose statistics cannot match. + /// + /// This only reduces what is fetched, so a scan still returns a superset of + /// the matching rows and callers needing exact results must filter again. + /// + /// # Errors + /// Returns an error if a column is missing from the table, has no schema + /// field id, or holds a literal its declared type cannot represent exactly. + pub fn filter(mut self, predicate: Predicate) -> Result { + // Resolve against the full row type: the server evaluates the filter + // before projection, so projected indices would name the wrong columns. + self.filter = Some(to_pb_predicate(&predicate, self.table_info.get_row_type())?); + Ok(self) + } + + /// Batch scanners have no predicate field in their request; reject a + /// configured filter rather than silently ignoring it. + fn reject_filter(&self, scanner: &str) -> Result<()> { + if self.filter.is_some() { + return Err(Error::UnsupportedOperation { + message: format!( + "{scanner} doesn't support filter pushdown. Table: {}", + self.table_info.table_path + ), + }); + } + Ok(()) + } + /// Log scanners don't support limit pushdown; reject a configured limit /// rather than silently ignoring it. fn reject_limit(&self, scanner: &str) -> Result<()> { @@ -134,6 +170,7 @@ impl<'a> TableScan<'a> { self, table_bucket: TableBucket, ) -> Result { + self.reject_filter("BatchScanner")?; let limit = self.limit.ok_or_else(|| Error::IllegalArgument { message: "create_bucket_batch_scanner requires a limit configured via .limit(n)" .to_string(), @@ -352,6 +389,7 @@ impl<'a> TableScan<'a> { self.conn.config(), self.projected_fields, self.fixed_schema, + self.filter, admin, )?; Ok(LogScanner { @@ -376,6 +414,7 @@ impl<'a> TableScan<'a> { self.conn.config(), self.projected_fields, self.fixed_schema, + self.filter, admin, )?; Ok(RecordBatchLogScanner { @@ -555,6 +594,7 @@ impl Drop for LogScannerInner { } impl LogScannerInner { + #[allow(clippy::too_many_arguments)] fn new( table_info: &TableInfo, metadata: Arc, @@ -562,6 +602,7 @@ impl LogScannerInner { config: &Config, projected_fields: Option>, fixed_schema: bool, + filter: Option, admin: Arc, ) -> Result { let log_scanner_status = Arc::new(LogScannerStatus::new()); @@ -608,6 +649,7 @@ impl LogScannerInner { config, projected_fields, fixed_schema, + filter, Arc::clone(&metrics), schema_getter, )?, @@ -1170,6 +1212,9 @@ struct LogFetcher { /// Per-table scanner metric handles shared with the owning /// `LogScannerInner` and `RemoteLogDownloader`. metrics: Arc, + /// Encoded filter sent on every fetch request, paired with the schema id it + /// was compiled against so the server can resolve its field ids. + filter: Option<(PbPredicate, i32)>, max_poll_records: usize, fetch_max_bytes: i32, fetch_min_bytes: i32, @@ -1200,6 +1245,7 @@ impl LogFetcher { config: &Config, projected_fields: Option>, fixed_schema: bool, + filter: Option, metrics: Arc, schema_getter: Arc, ) -> Result { @@ -1280,6 +1326,7 @@ impl LogFetcher { log_fetch_buffer, nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())), metrics, + filter: filter.map(|predicate| (predicate, table_info.get_schema_id())), max_poll_records: config.scanner_log_max_poll_records, fetch_max_bytes: config.scanner_log_fetch_max_bytes, fetch_min_bytes: config.scanner_log_fetch_min_bytes, @@ -1638,9 +1685,19 @@ impl LogFetcher { fetch_offset, high_watermark, ); - } else if fetch_log_for_bucket.records.is_some() { - // Handle regular in-memory records - create completed fetch directly + } else if fetch_log_for_bucket.records.is_some() + || fetch_log_for_bucket.filtered_end_offset.is_some() + { + // Handle regular in-memory records - create completed fetch directly. + // A filtered response may arrive empty, or carry records with a + // pruned tail; either way the end offset is how far the server + // scanned, so the client skips that range instead of re-fetching it. let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1); + let filtered_end_offset = Self::validate_filtered_end_offset( + fetch_log_for_bucket.filtered_end_offset, + fetch_offset, + &table_bucket, + ); let records = fetch_log_for_bucket.records.unwrap_or(vec![]); let size_in_bytes = records.len(); @@ -1653,13 +1710,33 @@ impl LogFetcher { false, // is_remote fetch_offset, high_watermark, - ); + ) + .with_filtered_end_offset(filtered_end_offset); log_fetch_buffer.add(Box::new(completed_fetch)); } } } } + /// Drops a filtered end offset that would move the bucket backwards, since + /// the server is only ever meant to report a range it has already scanned. + fn validate_filtered_end_offset( + filtered_end_offset: Option, + fetch_offset: i64, + table_bucket: &TableBucket, + ) -> i64 { + match filtered_end_offset { + Some(end) if end >= fetch_offset => end, + Some(end) => { + warn!( + "Ignoring filtered end offset {end} for bucket {table_bucket} because it precedes the fetch offset {fetch_offset}" + ); + NO_FILTERED_END_OFFSET + } + None => NO_FILTERED_END_OFFSET, + } + } + fn pending_remote_fetches( remote_log_downloader: Arc, log_fetch_buffer: Arc, @@ -2176,8 +2253,9 @@ impl LogFetcher { projection_pushdown_enabled: projection_enabled, projected_fields: projected_fields.clone(), buckets_req: feq_for_buckets, - filter_predicate: None, - filter_schema_id: None, + // The proto requires both filter fields to be set together. + filter_predicate: self.filter.as_ref().map(|(p, _)| p.clone()), + filter_schema_id: self.filter.as_ref().map(|&(_, id)| id), }; let fetch_log_request = FetchLogRequest { @@ -2478,6 +2556,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2520,6 +2599,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2560,6 +2640,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2571,6 +2652,220 @@ mod tests { Ok(()) } + /// Builds the fetcher used by the filter tests, encoding `predicate` the way + /// `TableScan::filter` does. + fn filtering_fetcher( + table_info: &TableInfo, + metadata: &Arc, + status: Arc, + predicate: Option, + ) -> Result { + let filter = predicate + .map(|p| to_pb_predicate(&p, table_info.get_row_type())) + .transpose()?; + LogFetcher::new( + table_info.clone(), + Arc::new(RpcClient::new()), + metadata.clone(), + status, + &Config::default(), + None, + false, + filter, + test_scanner_metrics(&table_info.table_path), + test_schema_getter(table_info, metadata), + ) + } + + #[tokio::test] + async fn prepare_fetch_log_requests_carries_the_filter() -> Result<()> { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 status = Arc::new(LogScannerStatus::new()); + status.assign_scan_bucket(TableBucket::new(1, 0), 0); + let fetcher = filtering_fetcher( + &table_info, + &metadata, + status, + Some(crate::predicate::col("id").gt(5i32)), + )?; + + let requests = fetcher.prepare_fetch_log_requests().await; + let table_req = &requests.get(&1).expect("request for leader").tables_req[0]; + let predicate = table_req + .filter_predicate + .as_ref() + .expect("filter predicate"); + assert_eq!(predicate.r#type, 0); + assert_eq!(predicate.leaf.as_ref().expect("leaf").field_id, 0); + // Both fields must travel together, and the id pins the field ids. + assert_eq!(table_req.filter_schema_id, Some(table_info.get_schema_id())); + Ok(()) + } + + #[tokio::test] + async fn prepare_fetch_log_requests_omits_an_absent_filter() -> Result<()> { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 status = Arc::new(LogScannerStatus::new()); + status.assign_scan_bucket(TableBucket::new(1, 0), 0); + let fetcher = filtering_fetcher(&table_info, &metadata, status, None)?; + + let requests = fetcher.prepare_fetch_log_requests().await; + let table_req = &requests.get(&1).expect("request for leader").tables_req[0]; + assert!(table_req.filter_predicate.is_none()); + assert!(table_req.filter_schema_id.is_none()); + Ok(()) + } + + #[tokio::test] + async fn unresolvable_filter_column_is_rejected() { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 result = filtering_fetcher( + &table_info, + &metadata, + Arc::new(LogScannerStatus::new()), + Some(crate::predicate::col("nope").gt(5i32)), + ); + assert!(matches!(result.err(), Some(Error::IllegalArgument { .. }))); + } + + /// Without this the bucket offset never advances and the scanner re-requests + /// the same range forever whenever a filter prunes a whole fetch. + #[tokio::test] + async fn handle_fetch_response_advances_past_a_fully_filtered_range() -> Result<()> { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 status = Arc::new(LogScannerStatus::new()); + let bucket = TableBucket::new(1, 0); + status.assign_scan_bucket(bucket.clone(), 2); + let fetcher = filtering_fetcher( + &table_info, + &metadata, + status.clone(), + Some(crate::predicate::col("id").gt(5i32)), + )?; + + LogFetcher::handle_fetch_response( + filtered_response(Some(11), Some(9)), + test_response_context(&fetcher, &metadata), + ) + .await; + + let fetched = fetcher.collect_fetches().await?; + assert!(fetched.is_empty()); + assert_eq!(status.get_bucket_offset(&bucket), Some(11)); + Ok(()) + } + + #[tokio::test] + async fn handle_fetch_response_ignores_a_filtered_range_behind_the_fetch_offset() -> Result<()> + { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 status = Arc::new(LogScannerStatus::new()); + let bucket = TableBucket::new(1, 0); + status.assign_scan_bucket(bucket.clone(), 5); + let fetcher = filtering_fetcher( + &table_info, + &metadata, + status.clone(), + Some(crate::predicate::col("id").gt(5i32)), + )?; + + LogFetcher::handle_fetch_response( + filtered_response(Some(3), None), + test_response_context(&fetcher, &metadata), + ) + .await; + + let fetched = fetcher.collect_fetches().await?; + assert!(fetched.is_empty()); + assert_eq!(status.get_bucket_offset(&bucket), Some(5)); + Ok(()) + } + + /// The server reports a filtered range alongside records when it prunes only + /// the tail of what it scanned, so the offset must clear the whole range. + #[tokio::test] + async fn handle_fetch_response_skips_a_pruned_tail_after_its_records() -> Result<()> { + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + 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 status = Arc::new(LogScannerStatus::new()); + let bucket = TableBucket::new(1, 0); + status.assign_scan_bucket(bucket.clone(), 0); + let fetcher = filtering_fetcher( + &table_info, + &metadata, + status.clone(), + Some(crate::predicate::col("id").gt(5i32)), + )?; + + let mut response = filtered_response(Some(8), Some(9)); + response.tables_resp[0].buckets_resp[0].records = + Some(build_records(&table_info, Arc::new(table_path))?); + LogFetcher::handle_fetch_response(response, test_response_context(&fetcher, &metadata)) + .await; + + let fetched = fetcher.collect_fetches().await?; + assert_eq!(fetched.get(&bucket).expect("records").len(), 1); + // The single record ends at offset 1, but the server scanned through 8. + assert_eq!(status.get_bucket_offset(&bucket), Some(8)); + Ok(()) + } + + /// A response for bucket 0 of table 1 that carries no records, standing in + /// for a fetch whose batches the server pruned entirely. + fn filtered_response( + filtered_end_offset: Option, + high_watermark: Option, + ) -> FetchLogResponse { + FetchLogResponse { + tables_resp: vec![PbFetchLogRespForTable { + table_id: 1, + buckets_resp: vec![PbFetchLogRespForBucket { + partition_id: None, + bucket_id: 0, + error_code: None, + error_message: None, + high_watermark, + log_start_offset: None, + remote_log_fetch_info: None, + records: None, + filtered_end_offset, + }], + }], + } + } + + fn test_response_context( + fetcher: &LogFetcher, + metadata: &Arc, + ) -> FetchResponseContext { + FetchResponseContext { + metadata: metadata.clone(), + log_fetch_buffer: fetcher.log_fetch_buffer.clone(), + log_scanner_status: fetcher.log_scanner_status.clone(), + resolver: Arc::clone(&fetcher.resolver), + remote_log_downloader: fetcher.remote_log_downloader.clone(), + metrics: Arc::clone(&fetcher.metrics), + request_start_time: Instant::now(), + } + } + #[tokio::test] async fn handle_fetch_response_sets_error() -> Result<()> { let table_path = TablePath::new("db".to_string(), "tbl".to_string()); @@ -2587,6 +2882,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2642,6 +2938,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2776,6 +3073,7 @@ mod tests { &config, None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), )?; @@ -2823,6 +3121,7 @@ mod tests { &Config::default(), None, false, + None, admin, ) .expect("build LogScannerInner"); @@ -2990,6 +3289,7 @@ mod tests { &Config::default(), None, false, + None, test_scanner_metrics(&table_path), test_schema_getter(&table_info, &metadata), ) @@ -3339,6 +3639,7 @@ mod tests { &Config::default(), None, false, + None, admin, ) .expect("build LogScannerInner"); diff --git a/fluss-rust/crates/fluss/src/predicate/mod.rs b/fluss-rust/crates/fluss/src/predicate/mod.rs index 82cff8c7fee..feb6aefb934 100644 --- a/fluss-rust/crates/fluss/src/predicate/mod.rs +++ b/fluss-rust/crates/fluss/src/predicate/mod.rs @@ -31,12 +31,8 @@ //! Setting a predicate on a scan is only supported on log scans over tables //! with the ARROW log format. -// Everything below is reachable only once `TableScan::filter` threads a -// predicate into the log fetcher. -#[allow(dead_code, reason = "consumed by the scan plumbing")] mod pb; -#[allow(unused_imports, reason = "consumed by the scan plumbing")] pub(crate) use pb::to_pb_predicate; use crate::row::{Decimal, TimestampLtz, TimestampNtz};