chore: return a bool on snapshot update to indicate whether files were read

This makes it easier to understand whether or not the function has
actually _read_ the ReplayStream

Signed-off-by: R. Tyler Croy <[email protected]>

Author: rtyler

crates/core/src/kernel/snapshot/mod.rs

@@ -432,13 +432,18 @@ impl EagerSnapshot {
     }
 
     /// Update the snapshot to the given version
+    ///
+    /// This will return a true value if the [LogStore] was read from. This can be helpful for
+    /// understanding whether the snapshot loaded data or not
     pub async fn update(
         &mut self,
         log_store: &dyn LogStore,
         target_version: Option<i64>,
-    ) -> DeltaResult<()> {
+    ) -> DeltaResult<bool> {
+        // Whether or not data has been read by this function
+        let mut read_data = false;
         if Some(self.version()) == target_version {
-            return Ok(());
+            return Ok(read_data);
         }
 
         let new_slice = self
@@ -447,7 +452,7 @@ impl EagerSnapshot {
             .await?;
 
         if new_slice.is_none() {
-            return Ok(());
+            return Ok(read_data);
         }
         let new_slice = new_slice.unwrap();
 
@@ -457,9 +462,15 @@ impl EagerSnapshot {
             .flat_map(get_visitor)
             .collect::<Vec<_>>();
 
+        // If files is `None` then this can exit early because the snapshot has intentionally been
+        // loaded _without_ files
+        if self.files.is_none() {
+            self.process_visitors(visitors)?;
+            return Ok(read_data);
+        }
+
         let mut schema_actions: HashSet<_> =
             visitors.iter().flat_map(|v| v.required_actions()).collect();
-        let require_files = self.files.is_some();
         let files = std::mem::take(&mut self.files);
 
         schema_actions.insert(ActionType::Add);
@@ -479,23 +490,17 @@ impl EagerSnapshot {
         let log_stream = new_slice.commit_stream(log_store, &read_schema, &self.snapshot.config)?;
 
         let mapper = LogMapper::try_new(&self.snapshot, None)?;
+        let files =
+            ReplayStream::try_new(log_stream, checkpoint_stream, &self.snapshot, &mut visitors)?
+                .map(|batch| batch.and_then(|b| mapper.map_batch(b)))
+                .try_collect()
+                .await?;
 
-        if require_files {
-            let files = ReplayStream::try_new(
-                log_stream,
-                checkpoint_stream,
-                &self.snapshot,
-                &mut visitors,
-            )?
-            .map(|batch| batch.and_then(|b| mapper.map_batch(b)))
-            .try_collect()
-            .await?;
-
-            self.files = Some(files);
-        }
+        self.files = Some(files);
+        read_data = true;
         self.process_visitors(visitors)?;
 
-        Ok(())
+        Ok(read_data)
     }
 
     /// Get the underlying snapshot
@@ -550,7 +555,7 @@ impl EagerSnapshot {
     pub fn log_data(&self) -> LogDataHandler<'_> {
         static EMPTY: Vec<RecordBatch> = vec![];
         LogDataHandler::new(
-            &self.files.as_ref().unwrap_or(&EMPTY),
+            self.files.as_ref().unwrap_or(&EMPTY),
             self.metadata(),
             self.schema(),
         )

crates/core/src/lib.rs

@@ -231,9 +231,11 @@ mod tests {
         let table_newest_version = crate::open_table(path).await.unwrap();
         let mut table_to_update = crate::open_table_with_version(path, 0).await.unwrap();
         // calling update several times should not produce any duplicates
-        table_to_update.update().await.unwrap();
-        table_to_update.update().await.unwrap();
-        table_to_update.update().await.unwrap();
+        // The first call should have read some data
+        assert_eq!(true, table_to_update.update().await.unwrap());
+        // Subsequent calls should not
+        assert_eq!(false, table_to_update.update().await.unwrap());
+        assert_eq!(false, table_to_update.update().await.unwrap());
 
         assert_eq!(
             table_newest_version.get_files_iter().unwrap().collect_vec(),

crates/core/src/table/builder.rs

@@ -290,7 +290,7 @@ impl DeltaTableBuilder {
             DeltaVersion::Newest => table.load().await?,
             DeltaVersion::Version(v) => table.load_version(v).await?,
             DeltaVersion::Timestamp(ts) => table.load_with_datetime(ts).await?,
-        }
+        };
         Ok(table)
     }
 }

crates/core/src/table/mod.rs

@@ -195,13 +195,13 @@ impl DeltaTable {
     }
 
     /// Load DeltaTable with data from latest checkpoint
-    pub async fn load(&mut self) -> Result<(), DeltaTableError> {
+    pub async fn load(&mut self) -> Result<bool, DeltaTableError> {
         self.update_incremental(None).await
     }
 
     /// Updates the DeltaTable to the most recent state committed to the transaction log by
     /// loading the last checkpoint and incrementally applying each version since.
-    pub async fn update(&mut self) -> Result<(), DeltaTableError> {
+    pub async fn update(&mut self) -> Result<bool, DeltaTableError> {
         self.update_incremental(None).await
     }
 
@@ -210,21 +210,21 @@ impl DeltaTable {
     pub async fn update_incremental(
         &mut self,
         max_version: Option<i64>,
-    ) -> Result<(), DeltaTableError> {
+    ) -> Result<bool, DeltaTableError> {
         match self.state.as_mut() {
             Some(state) => state.update(&self.log_store, max_version).await,
             _ => {
                 let state =
                     DeltaTableState::try_new(&self.log_store, self.config.clone(), max_version)
                         .await?;
                 self.state = Some(state);
-                Ok(())
+                Ok(true)
             }
         }
     }
 
     /// Loads the DeltaTable state for the given version.
-    pub async fn load_version(&mut self, version: i64) -> Result<(), DeltaTableError> {
+    pub async fn load_version(&mut self, version: i64) -> Result<bool, DeltaTableError> {
         if let Some(snapshot) = &self.state {
             if snapshot.version() > version {
                 self.state = None;
@@ -379,7 +379,7 @@ impl DeltaTable {
     pub async fn load_with_datetime(
         &mut self,
         datetime: DateTime<Utc>,
-    ) -> Result<(), DeltaTableError> {
+    ) -> Result<bool, DeltaTableError> {
         let mut min_version: i64 = -1;
         let log_store = self.log_store();
         let prefix = log_store.log_path();

crates/core/src/table/state.rs

@@ -204,9 +204,8 @@ impl DeltaTableState {
         &mut self,
         log_store: &dyn LogStore,
         version: Option<i64>,
-    ) -> Result<(), DeltaTableError> {
-        self.snapshot.update(log_store, version).await?;
-        Ok(())
+    ) -> Result<bool, DeltaTableError> {
+        self.snapshot.update(log_store, version).await
     }
 
     /// Obtain Add actions for files that match the filter

python/src/lib.rs

@@ -306,7 +306,7 @@ impl RawDeltaTable {
     /// Load the internal [RawDeltaTable] with the table state from the specified `version`
     ///
     /// This will acquire the internal lock since it is a mutating operation!
-    pub fn load_version(&self, py: Python, version: i64) -> PyResult<()> {
+    pub fn load_version(&self, py: Python, version: i64) -> PyResult<bool> {
         py.allow_threads(|| {
             #[allow(clippy::await_holding_lock)]
             rt().block_on(async {
@@ -359,7 +359,7 @@ impl RawDeltaTable {
         })
     }
 
-    pub fn load_with_datetime(&self, py: Python, ds: &str) -> PyResult<()> {
+    pub fn load_with_datetime(&self, py: Python, ds: &str) -> PyResult<bool> {
         py.allow_threads(|| {
             let datetime =
                 DateTime::<Utc>::from(DateTime::<FixedOffset>::parse_from_rfc3339(ds).map_err(
@@ -1014,7 +1014,7 @@ impl RawDeltaTable {
             .collect())
     }
 
-    pub fn update_incremental(&self) -> PyResult<()> {
+    pub fn update_incremental(&self) -> PyResult<bool> {
         #[allow(clippy::await_holding_lock)]
         #[allow(deprecated)]
         Ok(rt()