diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 67b8cde4d8..e41c36e2a3 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -23,13 +23,14 @@ use futures::{SinkExt, TryFutureExt}; use crate::delete_file_index::DeleteFileIndex; use crate::expr::{Bind, BoundPredicate, Predicate}; use crate::io::object_cache::ObjectCache; +use crate::scan::plan::BoundPredicates; +use crate::scan::source::ManifestSource; use crate::scan::{ - BoundPredicates, ExpressionEvaluatorCache, FileScanTask, ManifestEvaluatorCache, - PartitionFilterCache, + ExpressionEvaluatorCache, FileScanTask, ManifestEvaluatorCache, PartitionFilterCache, }; use crate::spec::{ - ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping, - PartitionSpecRef, SchemaRef, SnapshotRef, StructType, TableMetadataRef, + ManifestContentType, ManifestEntryRef, ManifestFile, NameMapping, PartitionSpecRef, SchemaRef, + StructType, TableMetadataRef, }; use crate::{Error, ErrorKind, Result}; @@ -154,14 +155,14 @@ impl ManifestEntryContext { } } -/// PlanContext wraps a [`SnapshotRef`] alongside all the other -/// objects that are required to perform a scan file plan. +/// PlanContext holds everything required to perform a scan file plan: the +/// manifests to read, and how to project, filter and evaluate their entries. #[derive(Debug)] pub(crate) struct PlanContext { - pub snapshot: SnapshotRef, + pub manifest_source: Arc, pub table_metadata: TableMetadataRef, - pub snapshot_schema: SchemaRef, + pub plan_schema: SchemaRef, pub case_sensitive: bool, pub predicate: Option>, pub snapshot_bound_predicate: Option>, @@ -177,44 +178,16 @@ pub(crate) struct PlanContext { } impl PlanContext { - pub(crate) async fn get_manifest_list(&self) -> Result> { - self.object_cache - .as_ref() - .get_manifest_list(&self.snapshot, &self.table_metadata) - .await - } - - /// Returns the partition filter for a manifest. See [`PartitionFilterCache::get`] for the - /// always-true fallback when the manifest's spec cannot be resolved against the scan schema. - fn get_partition_filter(&self, manifest_file: &ManifestFile) -> Result> { - let partition_spec_id = manifest_file.partition_spec_id; - - let partition_filter = self.partition_filter_cache.get( - partition_spec_id, - &self.table_metadata, - &self.snapshot_schema, - self.case_sensitive, - self.predicate - .as_ref() - .ok_or(Error::new( - ErrorKind::Unexpected, - "Expected a predicate but none present", - ))? - .as_ref() - .bind(self.snapshot_schema.clone(), self.case_sensitive)?, - )?; - - Ok(partition_filter) - } - - pub(crate) fn build_manifest_file_contexts( + pub(crate) async fn build_manifest_file_contexts( &self, - manifest_list: Arc, tx_data: Sender, delete_file_idx: DeleteFileIndex, delete_file_tx: Sender, ) -> Result> + 'static>> { - let mut manifest_files = manifest_list.entries().iter().collect::>(); + let mut manifest_files = self + .manifest_source + .manifest_files(&self.object_cache, &self.table_metadata) + .await?; // Sort manifest files to process delete manifests first. // This avoids a deadlock where the producer blocks on sending data manifest entries // (because the data channel is full) while the delete manifest consumer is waiting @@ -236,7 +209,7 @@ impl PlanContext { }; let partition_bound_predicate = if self.predicate.is_some() { - let partition_bound_predicate = self.get_partition_filter(manifest_file)?; + let partition_bound_predicate = self.get_partition_filter(&manifest_file)?; // evaluate the ManifestFile against the partition filter. Skip // if it cannot contain any matching rows @@ -246,7 +219,7 @@ impl PlanContext { manifest_file.partition_spec_id, partition_bound_predicate.clone(), ) - .eval(manifest_file)? + .eval(&manifest_file)? { continue; } @@ -271,7 +244,7 @@ impl PlanContext { fn create_manifest_file_context( &self, - manifest_file: &ManifestFile, + manifest_file: ManifestFile, partition_filter: Option>, sender: Sender, delete_file_index: DeleteFileIndex, @@ -288,22 +261,47 @@ impl PlanContext { None }; + let partition_spec = self + .table_metadata + .partition_spec_by_id(manifest_file.partition_spec_id) + .cloned(); + ManifestFileContext { - manifest_file: manifest_file.clone(), + manifest_file, bound_predicates, sender, object_cache: self.object_cache.clone(), - snapshot_schema: self.snapshot_schema.clone(), + snapshot_schema: self.plan_schema.clone(), field_ids: self.field_ids.clone(), expression_evaluator_cache: self.expression_evaluator_cache.clone(), delete_file_index, name_mapping: self.name_mapping.clone(), case_sensitive: self.case_sensitive, - partition_spec: self - .table_metadata - .partition_spec_by_id(manifest_file.partition_spec_id) - .cloned(), + partition_spec, unified_partition_type: self.unified_partition_type.clone(), } } + + /// Returns the partition filter for a manifest. See [`PartitionFilterCache::get`] for the + /// always-true fallback when the manifest's spec cannot be resolved against the scan schema. + fn get_partition_filter(&self, manifest_file: &ManifestFile) -> Result> { + let partition_spec_id = manifest_file.partition_spec_id; + + let partition_filter = self.partition_filter_cache.get( + partition_spec_id, + &self.table_metadata, + &self.plan_schema, + self.case_sensitive, + self.predicate + .as_ref() + .ok_or(Error::new( + ErrorKind::Unexpected, + "Expected a predicate but none present", + ))? + .as_ref() + .bind(self.plan_schema.clone(), self.case_sensitive)?, + )?; + + Ok(partition_filter) + } } diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 7abbf2dddf..fe7bdbc0e1 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -21,28 +21,28 @@ mod cache; use cache::*; mod context; use context::*; +mod plan; +mod source; mod task; use std::sync::Arc; use arrow_array::RecordBatch; -use futures::channel::mpsc::{Sender, channel}; use futures::stream::BoxStream; -use futures::{SinkExt, StreamExt, TryStreamExt}; pub use task::*; use crate::arrow::ArrowReaderBuilder; pub use crate::arrow::{ScanMetrics, ScanResult}; -use crate::delete_file_index::DeleteFileIndex; -use crate::expr::visitors::inclusive_metrics_evaluator::InclusiveMetricsEvaluator; -use crate::expr::{Bind, BoundPredicate, Predicate}; +use crate::expr::{Bind, Predicate}; use crate::io::FileIO; use crate::metadata_columns::{ RESERVED_FIELD_ID_PARTITION, get_metadata_field_id, is_metadata_column_name, }; use crate::partitioning::compute_unified_partition_type; use crate::runtime::Runtime; -use crate::spec::{DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, Schema, SnapshotRef}; +use crate::scan::plan::plan_tasks; +use crate::scan::source::{EmptySource, ManifestSource, SnapshotSource}; +use crate::spec::{DEFAULT_SCHEMA_NAME_MAPPING, NameMapping, Schema, SnapshotRef}; use crate::table::Table; use crate::util::available_parallelism; use crate::{Error, ErrorKind, Result}; @@ -201,38 +201,34 @@ impl<'a> TableScanBuilder<'a> { /// Build the table scan. pub fn build(self) -> Result { + // A table with no snapshots scans no files, but is still projected and + // filtered against its current schema so that an invalid column or + // predicate is rejected either way. let snapshot = match self.snapshot_id { - Some(snapshot_id) => self - .table - .metadata() - .snapshot_by_id(snapshot_id) - .ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("Snapshot with id {snapshot_id} not found"), - ) - })? - .clone(), - None => { - let Some(current_snapshot_id) = self.table.metadata().current_snapshot() else { - return Ok(TableScan { - batch_size: self.batch_size, - column_names: self.column_names, - file_io: self.table.file_io().clone(), - plan_context: None, - concurrency_limit_data_files: self.concurrency_limit_data_files, - concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries, - concurrency_limit_manifest_files: self.concurrency_limit_manifest_files, - row_group_filtering_enabled: self.row_group_filtering_enabled, - row_selection_enabled: self.row_selection_enabled, - runtime: self.table.runtime().clone(), - }); - }; - current_snapshot_id.clone() - } + Some(snapshot_id) => Some( + self.table + .metadata() + .snapshot_by_id(snapshot_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Snapshot with id {snapshot_id} not found"), + ) + })? + .clone(), + ), + None => self.table.metadata().current_snapshot().cloned(), }; - let schema = snapshot.schema(self.table.metadata())?; + let manifest_source: Arc = match snapshot { + Some(ref snapshot) => Arc::new(SnapshotSource::new(snapshot.clone())), + None => Arc::new(EmptySource), + }; + + let schema = match snapshot { + Some(ref snapshot) => snapshot.schema(self.table.metadata())?, + None => self.table.metadata().current_schema().clone(), + }; let mut field_ids = vec![]; let column_names = self.column_names.clone().unwrap_or_else(|| { @@ -314,9 +310,9 @@ impl<'a> TableScanBuilder<'a> { }; let plan_context = PlanContext { - snapshot, + manifest_source, table_metadata: self.table.metadata_ref(), - snapshot_schema: schema, + plan_schema: schema, case_sensitive: self.case_sensitive, predicate: self.filter.map(Arc::new), snapshot_bound_predicate: snapshot_bound_predicate.map(Arc::new), @@ -333,7 +329,8 @@ impl<'a> TableScanBuilder<'a> { batch_size: self.batch_size, column_names: self.column_names, file_io: self.table.file_io().clone(), - plan_context: Some(plan_context), + snapshot, + plan_context, concurrency_limit_data_files: self.concurrency_limit_data_files, concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries, concurrency_limit_manifest_files: self.concurrency_limit_manifest_files, @@ -347,10 +344,10 @@ impl<'a> TableScanBuilder<'a> { /// Table scan. #[derive(Debug)] pub struct TableScan { - /// A [PlanContext], if this table has at least one snapshot, otherwise None. - /// - /// If this is None, then the scan contains no rows. - plan_context: Option, + /// The snapshot being scanned, if this table has at least one snapshot. + /// When absent the scan yields no rows. + snapshot: Option, + plan_context: PlanContext, batch_size: Option, file_io: FileIO, column_names: Option>, @@ -375,123 +372,13 @@ pub struct TableScan { impl TableScan { /// Returns a stream of [`FileScanTask`]s. pub async fn plan_files(&self) -> Result { - let Some(plan_context) = self.plan_context.as_ref() else { - return Ok(Box::pin(futures::stream::empty())); - }; - - let concurrency_limit_manifest_files = self.concurrency_limit_manifest_files; - let concurrency_limit_manifest_entries = self.concurrency_limit_manifest_entries; - - // used to stream ManifestEntryContexts between stages of the file plan operation - let (manifest_entry_data_ctx_tx, manifest_entry_data_ctx_rx) = - channel(concurrency_limit_manifest_files); - let (manifest_entry_delete_ctx_tx, manifest_entry_delete_ctx_rx) = - channel(concurrency_limit_manifest_files); - - // used to stream the results back to the caller - let (file_scan_task_tx, file_scan_task_rx) = channel(concurrency_limit_manifest_entries); - - let (delete_file_idx, delete_file_tx) = DeleteFileIndex::new(self.runtime.clone()); - - let manifest_list = plan_context.get_manifest_list().await?; - - // get the [`ManifestFile`]s from the [`ManifestList`], filtering out any - // whose partitions cannot match this - // scan's filter - let manifest_file_contexts = plan_context.build_manifest_file_contexts( - manifest_list, - manifest_entry_data_ctx_tx, - delete_file_idx.clone(), - manifest_entry_delete_ctx_tx, - )?; - - let mut channel_for_manifest_error = file_scan_task_tx.clone(); - let mut channel_for_data_manifest_entry_error = file_scan_task_tx.clone(); - let mut channel_for_delete_manifest_entry_error = file_scan_task_tx.clone(); - - let rt = self.runtime.clone(); - - // Concurrently load all [`Manifest`]s and stream their [`ManifestEntry`]s - rt.io().spawn(async move { - let result = futures::stream::iter(manifest_file_contexts) - .try_for_each_concurrent(concurrency_limit_manifest_files, |ctx| async move { - ctx.fetch_manifest_and_stream_manifest_entries().await - }) - .await; - - if let Err(error) = result { - let _ = channel_for_manifest_error.send(Err(error)).await; - } - }); - - // Process the delete file [`ManifestEntry`] stream in parallel - { - let rt = rt.clone(); - let rt_inner = rt.clone(); - rt.cpu().spawn(async move { - let result = manifest_entry_delete_ctx_rx - .map(|me_ctx| Ok((me_ctx, delete_file_tx.clone()))) - .try_for_each_concurrent( - concurrency_limit_manifest_entries, - |(manifest_entry_context, tx)| { - let rt_inner = rt_inner.clone(); - async move { - rt_inner - .cpu() - .spawn(async move { - Self::process_delete_manifest_entry( - manifest_entry_context, - tx, - ) - .await - }) - .await? - } - }, - ) - .await; - - if let Err(error) = result { - let _ = channel_for_delete_manifest_entry_error - .send(Err(error)) - .await; - } - }); - } - - // Process the data file [`ManifestEntry`] stream in parallel - { - let rt_inner = rt.clone(); - rt.cpu().spawn(async move { - let result = manifest_entry_data_ctx_rx - .map(|me_ctx| Ok((me_ctx, file_scan_task_tx.clone()))) - .try_for_each_concurrent( - concurrency_limit_manifest_entries, - |(manifest_entry_context, tx)| { - let rt_inner = rt_inner.clone(); - async move { - rt_inner - .cpu() - .spawn(async move { - Self::process_data_manifest_entry( - manifest_entry_context, - tx, - ) - .await - }) - .await? - } - }, - ) - .await; - - if let Err(error) = result { - let _ = channel_for_data_manifest_entry_error.send(Err(error)).await; - } - }); - } - - Ok(file_scan_task_rx.boxed()) + plan_tasks( + &self.plan_context, + &self.runtime, + self.concurrency_limit_manifest_files, + self.concurrency_limit_manifest_entries, + ) + .await } /// Returns an [`ArrowRecordBatchStream`]. @@ -519,115 +406,10 @@ impl TableScan { /// Returns a reference to the snapshot of the table scan. pub fn snapshot(&self) -> Option<&SnapshotRef> { - self.plan_context.as_ref().map(|x| &x.snapshot) - } - - async fn process_data_manifest_entry( - manifest_entry_context: ManifestEntryContext, - mut file_scan_task_tx: Sender>, - ) -> Result<()> { - // skip processing this manifest entry if it has been marked as deleted - if !manifest_entry_context.manifest_entry.is_alive() { - return Ok(()); - } - - // abort the plan if we encounter a manifest entry for a delete file - if manifest_entry_context.manifest_entry.content_type() != DataContentType::Data { - return Err(Error::new( - ErrorKind::FeatureUnsupported, - "Encountered an entry for a delete file in a data file manifest", - )); - } - - if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates { - let BoundPredicates { - snapshot_bound_predicate, - partition_bound_predicate, - } = bound_predicates.as_ref(); - - let expression_evaluator_cache = - manifest_entry_context.expression_evaluator_cache.as_ref(); - - let expression_evaluator = expression_evaluator_cache.get( - manifest_entry_context.partition_spec_id, - partition_bound_predicate, - )?; - - // skip any data file whose partition data indicates that it can't contain - // any data that matches this scan's filter - if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? { - return Ok(()); - } - - // skip any data file whose metrics don't match this scan's filter - if !InclusiveMetricsEvaluator::eval( - snapshot_bound_predicate, - manifest_entry_context.manifest_entry.data_file(), - false, - )? { - return Ok(()); - } - } - - // congratulations! the manifest entry has made its way through the - // entire plan without getting filtered out. Create a corresponding - // FileScanTask and push it to the result stream - file_scan_task_tx - .send(Ok(manifest_entry_context.into_file_scan_task().await?)) - .await?; - - Ok(()) - } - - async fn process_delete_manifest_entry( - manifest_entry_context: ManifestEntryContext, - mut delete_file_ctx_tx: Sender, - ) -> Result<()> { - // skip processing this manifest entry if it has been marked as deleted - if !manifest_entry_context.manifest_entry.is_alive() { - return Ok(()); - } - - // abort the plan if we encounter a manifest entry that is not for a delete file - if manifest_entry_context.manifest_entry.content_type() == DataContentType::Data { - return Err(Error::new( - ErrorKind::FeatureUnsupported, - "Encountered an entry for a data file in a delete manifest", - )); - } - - if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates { - let expression_evaluator_cache = - manifest_entry_context.expression_evaluator_cache.as_ref(); - - let expression_evaluator = expression_evaluator_cache.get( - manifest_entry_context.partition_spec_id, - &bound_predicates.partition_bound_predicate, - )?; - - // skip any data file whose partition data indicates that it can't contain - // any data that matches this scan's filter - if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? { - return Ok(()); - } - } - - delete_file_ctx_tx - .send(DeleteFileContext { - manifest_entry: manifest_entry_context.manifest_entry.clone(), - partition_spec_id: manifest_entry_context.partition_spec_id, - }) - .await?; - - Ok(()) + self.snapshot.as_ref() } } -pub(crate) struct BoundPredicates { - partition_bound_predicate: BoundPredicate, - snapshot_bound_predicate: BoundPredicate, -} - #[cfg(test)] pub mod tests { //! shared tests for the table scan API @@ -1883,14 +1665,7 @@ pub mod tests { let table = TableTestFixture::new().table; let table_scan = table.scan().build().unwrap(); - assert!( - table_scan - .plan_context - .as_ref() - .unwrap() - .name_mapping - .is_none() - ); + assert!(table_scan.plan_context.name_mapping.is_none()); } #[test] @@ -1901,8 +1676,6 @@ pub mod tests { let table_scan = table.scan().build().unwrap(); let mapping = table_scan .plan_context - .as_ref() - .unwrap() .name_mapping .as_ref() .expect("name_mapping should be parsed from the table property"); @@ -1976,6 +1749,24 @@ pub mod tests { assert!(batches.is_empty()); } + #[test] + fn test_scan_without_any_snapshots_still_validates_projection() { + let table = TableTestFixture::new_empty().table; + + table + .scan() + .select(["x"]) + .build() + .expect("a column of the current schema should be projectable"); + + let error = table + .scan() + .select(["nonexistent"]) + .build() + .expect_err("an absent column should be rejected even with no snapshots"); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + } + #[tokio::test] async fn test_plan_files_no_deletions() { let mut fixture = TableTestFixture::new(); @@ -3097,23 +2888,12 @@ pub mod tests { .build() .unwrap_or_else(|e| panic!("scan of data column `{column_name}` failed: {e}")); - assert_eq!( - table_scan.plan_context.as_ref().unwrap().field_ids.as_ref(), - &[2] - ); + assert_eq!(table_scan.plan_context.field_ids.as_ref(), &[2]); // The default projection (all columns) must resolve to the real field ids // too, not shadow the data column with a reserved delete-file id. let default_scan = table.scan().build().unwrap(); - assert_eq!( - default_scan - .plan_context - .as_ref() - .unwrap() - .field_ids - .as_ref(), - &[1, 2] - ); + assert_eq!(default_scan.plan_context.field_ids.as_ref(), &[1, 2]); } } diff --git a/crates/iceberg/src/scan/plan.rs b/crates/iceberg/src/scan/plan.rs new file mode 100644 index 0000000000..8b9d5da06a --- /dev/null +++ b/crates/iceberg/src/scan/plan.rs @@ -0,0 +1,243 @@ +// 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. + +//! Drives a scan plan: fans manifests out to their entries and turns the +//! surviving entries into [`FileScanTask`]s. + +use futures::channel::mpsc::{Sender, channel}; +use futures::{SinkExt, StreamExt, TryStreamExt}; + +use crate::delete_file_index::DeleteFileIndex; +use crate::expr::BoundPredicate; +use crate::expr::visitors::inclusive_metrics_evaluator::InclusiveMetricsEvaluator; +use crate::runtime::Runtime; +use crate::scan::context::{ManifestEntryContext, PlanContext}; +use crate::scan::{DeleteFileContext, FileScanTask, FileScanTaskStream}; +use crate::spec::DataContentType; +use crate::{Error, ErrorKind, Result}; + +pub(crate) async fn plan_tasks( + plan_context: &PlanContext, + runtime: &Runtime, + concurrency_limit_manifest_files: usize, + concurrency_limit_manifest_entries: usize, +) -> Result { + // used to stream ManifestEntryContexts between stages of the file plan operation + let (manifest_entry_data_ctx_tx, manifest_entry_data_ctx_rx) = + channel(concurrency_limit_manifest_files); + let (manifest_entry_delete_ctx_tx, manifest_entry_delete_ctx_rx) = + channel(concurrency_limit_manifest_files); + + // used to stream the results back to the caller + let (file_scan_task_tx, file_scan_task_rx) = channel(concurrency_limit_manifest_entries); + + let (delete_file_idx, delete_file_tx) = DeleteFileIndex::new(runtime.clone()); + + // get the [`ManifestFile`]s from the [`ManifestList`], filtering out any + // whose partitions cannot match this + // scan's filter + let manifest_file_contexts = plan_context + .build_manifest_file_contexts( + manifest_entry_data_ctx_tx, + delete_file_idx.clone(), + manifest_entry_delete_ctx_tx, + ) + .await?; + + let mut channel_for_manifest_error = file_scan_task_tx.clone(); + let mut channel_for_data_manifest_entry_error = file_scan_task_tx.clone(); + let mut channel_for_delete_manifest_entry_error = file_scan_task_tx.clone(); + + let rt = runtime.clone(); + + // Concurrently load all [`Manifest`]s and stream their [`ManifestEntry`]s + rt.io().spawn(async move { + let result = futures::stream::iter(manifest_file_contexts) + .try_for_each_concurrent(concurrency_limit_manifest_files, |ctx| async move { + ctx.fetch_manifest_and_stream_manifest_entries().await + }) + .await; + + if let Err(error) = result { + let _ = channel_for_manifest_error.send(Err(error)).await; + } + }); + + // Process the delete file [`ManifestEntry`] stream in parallel + { + let rt = rt.clone(); + let rt_inner = rt.clone(); + rt.cpu().spawn(async move { + let result = manifest_entry_delete_ctx_rx + .map(|me_ctx| Ok((me_ctx, delete_file_tx.clone()))) + .try_for_each_concurrent( + concurrency_limit_manifest_entries, + |(manifest_entry_context, tx)| { + let rt_inner = rt_inner.clone(); + async move { + rt_inner + .cpu() + .spawn(async move { + process_delete_manifest_entry(manifest_entry_context, tx).await + }) + .await? + } + }, + ) + .await; + + if let Err(error) = result { + let _ = channel_for_delete_manifest_entry_error + .send(Err(error)) + .await; + } + }); + } + + // Process the data file [`ManifestEntry`] stream in parallel + { + let rt_inner = rt.clone(); + rt.cpu().spawn(async move { + let result = manifest_entry_data_ctx_rx + .map(|me_ctx| Ok((me_ctx, file_scan_task_tx.clone()))) + .try_for_each_concurrent( + concurrency_limit_manifest_entries, + |(manifest_entry_context, tx)| { + let rt_inner = rt_inner.clone(); + async move { + rt_inner + .cpu() + .spawn(async move { + process_data_manifest_entry(manifest_entry_context, tx).await + }) + .await? + } + }, + ) + .await; + + if let Err(error) = result { + let _ = channel_for_data_manifest_entry_error.send(Err(error)).await; + } + }); + } + + Ok(file_scan_task_rx.boxed()) +} + +async fn process_delete_manifest_entry( + manifest_entry_context: ManifestEntryContext, + mut delete_file_ctx_tx: Sender, +) -> Result<()> { + // skip processing this manifest entry if it has been marked as deleted + if !manifest_entry_context.manifest_entry.is_alive() { + return Ok(()); + } + + // abort the plan if we encounter a manifest entry that is not for a delete file + if manifest_entry_context.manifest_entry.content_type() == DataContentType::Data { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + "Encountered an entry for a data file in a delete manifest", + )); + } + + if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates { + let expression_evaluator_cache = manifest_entry_context.expression_evaluator_cache.as_ref(); + + let expression_evaluator = expression_evaluator_cache.get( + manifest_entry_context.partition_spec_id, + &bound_predicates.partition_bound_predicate, + )?; + + // skip any data file whose partition data indicates that it can't contain + // any data that matches this scan's filter + if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? { + return Ok(()); + } + } + + delete_file_ctx_tx + .send(DeleteFileContext { + manifest_entry: manifest_entry_context.manifest_entry.clone(), + partition_spec_id: manifest_entry_context.partition_spec_id, + }) + .await?; + + Ok(()) +} + +async fn process_data_manifest_entry( + manifest_entry_context: ManifestEntryContext, + mut file_scan_task_tx: Sender>, +) -> Result<()> { + // skip processing this manifest entry if it has been marked as deleted + if !manifest_entry_context.manifest_entry.is_alive() { + return Ok(()); + } + + // abort the plan if we encounter a manifest entry for a delete file + if manifest_entry_context.manifest_entry.content_type() != DataContentType::Data { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + "Encountered an entry for a delete file in a data file manifest", + )); + } + + if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates { + let BoundPredicates { + snapshot_bound_predicate, + partition_bound_predicate, + } = bound_predicates.as_ref(); + + let expression_evaluator_cache = manifest_entry_context.expression_evaluator_cache.as_ref(); + + let expression_evaluator = expression_evaluator_cache.get( + manifest_entry_context.partition_spec_id, + partition_bound_predicate, + )?; + + // skip any data file whose partition data indicates that it can't contain + // any data that matches this scan's filter + if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? { + return Ok(()); + } + + // skip any data file whose metrics don't match this scan's filter + if !InclusiveMetricsEvaluator::eval( + snapshot_bound_predicate, + manifest_entry_context.manifest_entry.data_file(), + false, + )? { + return Ok(()); + } + } + + // congratulations! the manifest entry has made its way through the + // entire plan without getting filtered out. Create a corresponding + // FileScanTask and push it to the result stream + file_scan_task_tx + .send(Ok(manifest_entry_context.into_file_scan_task().await?)) + .await?; + + Ok(()) +} + +pub(crate) struct BoundPredicates { + pub(crate) partition_bound_predicate: BoundPredicate, + pub(crate) snapshot_bound_predicate: BoundPredicate, +} diff --git a/crates/iceberg/src/scan/source.rs b/crates/iceberg/src/scan/source.rs new file mode 100644 index 0000000000..8249efabe0 --- /dev/null +++ b/crates/iceberg/src/scan/source.rs @@ -0,0 +1,88 @@ +// 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. + +//! Where a scan's manifests come from. +//! +//! A [`PlanContext`](crate::scan::PlanContext) says how to project, filter and +//! evaluate a scan; a [`ManifestSource`] says which manifests it reads. Keeping +//! the two apart lets one planning path serve a single snapshot and, later, a +//! range of them. + +use std::fmt::Debug; + +use futures::future::BoxFuture; + +use crate::Result; +use crate::io::object_cache::ObjectCache; +use crate::spec::{ManifestFile, SnapshotRef, TableMetadataRef}; + +/// Resolves the manifests a scan should read. +/// +/// The returned future borrows both the source and the objects it reads +/// through, so a source is reusable: [`TableScan::plan_files`] takes `&self` +/// and may be called more than once. +/// +/// [`TableScan::plan_files`]: crate::scan::TableScan::plan_files +pub(crate) trait ManifestSource: Debug + Send + Sync { + fn manifest_files<'a>( + &'a self, + object_cache: &'a ObjectCache, + table_metadata: &'a TableMetadataRef, + ) -> BoxFuture<'a, Result>>; +} + +/// No manifests at all, for a table that has no snapshots yet. +#[derive(Debug)] +pub(crate) struct EmptySource; + +impl ManifestSource for EmptySource { + fn manifest_files<'a>( + &'a self, + _object_cache: &'a ObjectCache, + _table_metadata: &'a TableMetadataRef, + ) -> BoxFuture<'a, Result>> { + Box::pin(async { Ok(vec![]) }) + } +} + +/// Every manifest listed by a single snapshot. +#[derive(Debug)] +pub(crate) struct SnapshotSource { + snapshot: SnapshotRef, +} + +impl SnapshotSource { + pub(crate) fn new(snapshot: SnapshotRef) -> Self { + Self { snapshot } + } +} + +impl ManifestSource for SnapshotSource { + fn manifest_files<'a>( + &'a self, + object_cache: &'a ObjectCache, + table_metadata: &'a TableMetadataRef, + ) -> BoxFuture<'a, Result>> { + Box::pin(async move { + let manifest_list = object_cache + .get_manifest_list(&self.snapshot, table_metadata) + .await?; + + Ok(manifest_list.entries().to_vec()) + }) + } +}