Adding fresh table loading on scan start

crates/integrations/datafusion/src/extensions.rs

@@ -4,6 +4,7 @@
 use std::sync;
 
 use datafusion::arrow::datatypes;
+use datafusion::error;
 use iceberg::table;
 
 impl crate::IcebergTableProvider {
@@ -28,4 +29,16 @@ impl crate::IcebergTableProvider {
             catalog: Some(catalog),
         }
     }
+
+    /// Returns table to be used in operations.
+    /// If the catalog implementer is provided, loads a fresh table from it, otherwise clones the inner value.
+    pub(crate) fn table_to_use(&self) -> error::Result<table::Table> {
+        match self.catalog {
+            Some(ref catalog) => {
+                futures::executor::block_on(catalog.load_table(self.table.identifier()))
+                    .map_err(|e| error::DataFusionError::Internal(e.to_string()))
+            }
+            None => Ok(self.table.clone()),
+        }
+    }
 }

crates/integrations/datafusion/src/table/mod.rs

@@ -154,7 +154,7 @@ impl TableProvider for IcebergTableProvider {
         _limit: Option<usize>,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
         Ok(Arc::new(IcebergTableScan::new(
-            self.table.clone(),
+            self.table_to_use()?, // This automatically updates available snapshots when starting a scan.
             self.snapshot_id,
             self.schema.clone(),
             projection,