feat(reader): support execution-time predicates in ArrowReader

crates/iceberg/public-api.txt

@@ -79,6 +79,7 @@ pub fn iceberg::arrow::ArrowReaderBuilder::with_range_coalesce_bytes(self, range
 pub fn iceberg::arrow::ArrowReaderBuilder::with_range_fetch_concurrency(self, range_fetch_concurrency: usize) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_row_group_filtering_enabled(self, row_group_filtering_enabled: bool) -> Self
 pub fn iceberg::arrow::ArrowReaderBuilder::with_row_selection_enabled(self, row_selection_enabled: bool) -> Self
+pub fn iceberg::arrow::ArrowReaderBuilder::with_runtime_predicate_provider(self, runtime_predicate_provider: alloc::sync::Arc<dyn iceberg::arrow::RuntimePredicateProvider>) -> Self
 pub struct iceberg::arrow::PartitionValueCalculator
 impl iceberg::arrow::PartitionValueCalculator
 pub fn iceberg::arrow::PartitionValueCalculator::calculate(&self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<arrow_array::array::ArrayRef>
@@ -93,6 +94,13 @@ pub fn iceberg::arrow::RecordBatchPartitionSplitter::split(&self, batch: &arrow_
 pub fn iceberg::arrow::RecordBatchPartitionSplitter::try_new(iceberg_schema: iceberg::spec::SchemaRef, partition_spec: iceberg::spec::PartitionSpecRef, calculator: core::option::Option<iceberg::arrow::PartitionValueCalculator>) -> iceberg::Result<Self>
 pub fn iceberg::arrow::RecordBatchPartitionSplitter::try_new_with_computed_values(iceberg_schema: iceberg::spec::SchemaRef, partition_spec: iceberg::spec::PartitionSpecRef) -> iceberg::Result<Self>
 pub fn iceberg::arrow::RecordBatchPartitionSplitter::try_new_with_precomputed_values(iceberg_schema: iceberg::spec::SchemaRef, partition_spec: iceberg::spec::PartitionSpecRef) -> iceberg::Result<Self>
+pub struct iceberg::arrow::RuntimePredicateSnapshot
+impl iceberg::arrow::RuntimePredicateSnapshot
+pub fn iceberg::arrow::RuntimePredicateSnapshot::generation(&self) -> u64
+pub fn iceberg::arrow::RuntimePredicateSnapshot::into_predicate(self) -> core::option::Option<iceberg::expr::Predicate>
+pub fn iceberg::arrow::RuntimePredicateSnapshot::new(predicate: core::option::Option<iceberg::expr::Predicate>, generation: u64) -> Self
+impl core::fmt::Debug for iceberg::arrow::RuntimePredicateSnapshot
+pub fn iceberg::arrow::RuntimePredicateSnapshot::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg::arrow::ScanMetrics
 impl iceberg::scan::ScanMetrics
 pub fn iceberg::scan::ScanMetrics::bytes_read(&self) -> u64
@@ -124,6 +132,9 @@ pub fn iceberg::arrow::ArrowSchemaVisitor::primitive(&mut self, p: &arrow_schema
 pub fn iceberg::arrow::ArrowSchemaVisitor::schema(&mut self, schema: &arrow_schema::schema::Schema, values: alloc::vec::Vec<Self::T>) -> iceberg::Result<Self::U>
 pub fn iceberg::arrow::ArrowSchemaVisitor::struct(&mut self, fields: &arrow_schema::fields::Fields, results: alloc::vec::Vec<Self::T>) -> iceberg::Result<Self::T>
 pub fn iceberg::arrow::ArrowSchemaVisitor::variant(&mut self, field: &arrow_schema::field::FieldRef) -> iceberg::Result<Self::T> where Self: core::marker::Sized
+pub trait iceberg::arrow::RuntimePredicateProvider: core::marker::Send + core::marker::Sync
+pub fn iceberg::arrow::RuntimePredicateProvider::generation(&self) -> u64
+pub fn iceberg::arrow::RuntimePredicateProvider::snapshot(&self) -> iceberg::Result<iceberg::arrow::RuntimePredicateSnapshot>
 pub fn iceberg::arrow::arrow_primitive_to_literal(primitive_array: &arrow_array::array::ArrayRef, ty: &iceberg::spec::Type) -> iceberg::Result<alloc::vec::Vec<core::option::Option<iceberg::spec::Literal>>>
 pub fn iceberg::arrow::arrow_schema_to_schema(schema: &arrow_schema::schema::Schema) -> iceberg::Result<iceberg::spec::Schema>
 pub fn iceberg::arrow::arrow_schema_to_schema_auto_assign_ids(schema: &arrow_schema::schema::Schema) -> iceberg::Result<iceberg::spec::Schema>

crates/iceberg/src/arrow/reader/mod.rs

@@ -17,6 +17,8 @@
 
 //! Parquet file data reader
 
+use std::sync::Arc;
+
 use crate::arrow::caching_delete_file_loader::CachingDeleteFileLoader;
 use crate::io::FileIO;
 use crate::runtime::Runtime;
@@ -42,13 +44,17 @@ mod predicate_visitor;
 mod projection;
 mod row_filter;
 mod row_lineage;
+mod runtime_predicate;
+#[cfg(test)]
+mod runtime_predicate_tests;
 pub use file_reader::ArrowFileReader;
 pub(crate) use options::ParquetReadOptions;
 use predicate_visitor::{CollectFieldIdVisitor, PredicateConverter};
 use projection::{
     add_fallback_field_ids_to_arrow_schema, apply_name_mapping_to_arrow_schema,
     find_leaf_by_field_id,
 };
+pub use runtime_predicate::{RuntimePredicateProvider, RuntimePredicateSnapshot};
 
 /// Builder to create ArrowReader
 pub struct ArrowReaderBuilder {
@@ -58,6 +64,7 @@ pub struct ArrowReaderBuilder {
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
     bloom_filter_enabled: bool,
+    runtime_predicate_provider: Option<Arc<dyn RuntimePredicateProvider>>,
     parquet_read_options: ParquetReadOptions,
     runtime: Runtime,
 }
@@ -74,6 +81,7 @@ impl ArrowReaderBuilder {
             row_group_filtering_enabled: true,
             row_selection_enabled: false,
             bloom_filter_enabled: false,
+            runtime_predicate_provider: None,
             parquet_read_options: ParquetReadOptions::builder().build(),
             runtime,
         }
@@ -119,6 +127,21 @@ impl ArrowReaderBuilder {
         self
     }
 
+    /// Supplies execution-time predicates, sampled when each data-file task
+    /// starts and combined with the task's filters using AND. See
+    /// [`RuntimePredicateProvider`].
+    ///
+    /// Only the reader accepts a provider. `TableScan::to_arrow` cannot pass
+    /// one, so wiring it through `TableScan` and execution engines is a
+    /// follow-up.
+    pub fn with_runtime_predicate_provider(
+        mut self,
+        runtime_predicate_provider: Arc<dyn RuntimePredicateProvider>,
+    ) -> Self {
+        self.runtime_predicate_provider = Some(runtime_predicate_provider);
+        self
+    }
+
     /// Provide a hint as to the number of bytes to prefetch for parsing the Parquet metadata
     ///
     /// This hint can help reduce the number of fetch requests. For more details see the
@@ -159,6 +182,7 @@ impl ArrowReaderBuilder {
             row_group_filtering_enabled: self.row_group_filtering_enabled,
             row_selection_enabled: self.row_selection_enabled,
             bloom_filter_enabled: self.bloom_filter_enabled,
+            runtime_predicate_provider: self.runtime_predicate_provider,
             parquet_read_options: self.parquet_read_options,
         }
     }
@@ -177,5 +201,6 @@ pub struct ArrowReader {
     row_group_filtering_enabled: bool,
     row_selection_enabled: bool,
     bloom_filter_enabled: bool,
+    runtime_predicate_provider: Option<Arc<dyn RuntimePredicateProvider>>,
     parquet_read_options: ParquetReadOptions,
 }