feat: allow iterating over logical files

Signed-off-by: Robert Pack <[email protected]>

Author: roeap

crates/core/src/kernel/snapshot_next/iterators.rs

@@ -1,5 +1,5 @@
 use std::collections::HashSet;
-use std::sync::{Arc, LazyLock};
+use std::sync::Arc;
 
 use arrow_array::cast::AsArray;
 use arrow_array::types::Int64Type;
@@ -14,7 +14,6 @@ use delta_kernel::engine::arrow_data::ArrowEngineData;
 use delta_kernel::engine::arrow_expression::ProvidesColumnByName;
 use delta_kernel::engine_data::{GetData, RowVisitor};
 use delta_kernel::expressions::{Scalar, StructData};
-use delta_kernel::scan::scan_row_schema;
 
 use crate::kernel::scalars::ScalarExt;
 use crate::{DeltaResult, DeltaTableError};
@@ -202,23 +201,6 @@ impl LogicalFileView {
     }
 }
 
-impl Iterator for LogicalFileView {
-    type Item = LogicalFileView;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        if self.index < self.files.num_rows() {
-            let file = LogicalFileView {
-                files: self.files.clone(),
-                index: self.index,
-            };
-            self.index += 1;
-            Some(file)
-        } else {
-            None
-        }
-    }
-}
-
 pub struct LogicalFileViewIterator<I>
 where
     I: IntoIterator<Item = Result<RecordBatch, DeltaTableError>>,
@@ -244,43 +226,43 @@ where
     }
 }
 
-// impl<I> Iterator for LogicalFileViewIterator<I>
-// where
-//     I: IntoIterator<Item = DeltaResult<RecordBatch>>,
-// {
-//     type Item = DeltaResult<LogicalFileView>;
-//
-//     fn next(&mut self) -> Option<Self::Item> {
-//         if let Some(batch) = &self.batch {
-//             if self.current < batch.num_rows() {
-//                 let item = LogicalFileView {
-//                     files: batch.clone(),
-//                     index: self.current,
-//                 };
-//                 self.current += 1;
-//                 return Some(Ok(item));
-//             }
-//         }
-//         match self.inner.next() {
-//             Some(Ok(batch)) => {
-//                 if validate_logical_file(&batch).is_err() {
-//                     return Some(Err(DeltaTableError::generic(
-//                         "Invalid logical file data encountered.",
-//                     )));
-//                 }
-//                 self.batch = Some(batch);
-//                 self.current = 0;
-//                 self.next()
-//             }
-//             Some(Err(e)) => Some(Err(e)),
-//             None => None,
-//         }
-//     }
-//
-//     fn size_hint(&self) -> (usize, Option<usize>) {
-//         self.inner.size_hint()
-//     }
-// }
+impl<I> Iterator for LogicalFileViewIterator<I>
+where
+    I: IntoIterator<Item = DeltaResult<RecordBatch>>,
+{
+    type Item = DeltaResult<LogicalFileView>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if let Some(batch) = &self.batch {
+            if self.current < batch.num_rows() {
+                let item = LogicalFileView {
+                    files: batch.clone(),
+                    index: self.current,
+                };
+                self.current += 1;
+                return Some(Ok(item));
+            }
+        }
+        match self.inner.next() {
+            Some(Ok(batch)) => {
+                if validate_logical_file(&batch).is_err() {
+                    return Some(Err(DeltaTableError::generic(
+                        "Invalid logical file data encountered.",
+                    )));
+                }
+                self.batch = Some(batch);
+                self.current = 0;
+                self.next()
+            }
+            Some(Err(e)) => Some(Err(e)),
+            None => None,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
+}
 
 pub struct AddViewIterator<I>
 where
@@ -353,6 +335,15 @@ pub(crate) fn validate_add(batch: &RecordBatch) -> DeltaResult<()> {
     Ok(())
 }
 
+fn validate_logical_file(batch: &RecordBatch) -> DeltaResult<()> {
+    validate_column::<StringArray>(batch, &["path"])?;
+    validate_column::<Int64Array>(batch, &["size"])?;
+    validate_column::<Int64Array>(batch, &["modificationTime"])?;
+    // validate_column::<StructArray>(batch, &["deletionVector"])?;
+    // validate_column::<StructArray>(batch, &["fileConstantValues"])?;
+    Ok(())
+}
+
 fn validate_column<'a, T: Array + 'static>(
     actions: &'a RecordBatch,
     col: &'a [impl AsRef<str>],

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

@@ -17,7 +17,7 @@ use delta_kernel::scan::scan_row_schema;
 use delta_kernel::schema::{DataType, Schema};
 use delta_kernel::table_properties::TableProperties;
 use delta_kernel::{EngineData, ExpressionRef, Version};
-use iterators::{AddView, AddViewIterator};
+use iterators::{AddView, AddViewIterator, LogicalFileView, LogicalFileViewIterator};
 use itertools::Itertools;
 use url::Url;
 
@@ -109,6 +109,16 @@ pub trait Snapshot {
         predicate: Option<ExpressionRef>,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<RecordBatch>>>>;
 
+    fn logical_files_view(
+        &self,
+        predicate: Option<ExpressionRef>,
+    ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<LogicalFileView>>>> {
+        #[allow(deprecated)]
+        Ok(Box::new(LogicalFileViewIterator::new(
+            self.logical_files(predicate)?,
+        )))
+    }
+
     /// Get all currently active files in the table.
     ///
     /// # Parameters
@@ -125,6 +135,10 @@ pub trait Snapshot {
         predicate: Option<ExpressionRef>,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<RecordBatch>>>>;
 
+    #[deprecated(
+        since = "0.25.0",
+        note = "Use `logical_files_view` instead, which returns a more focussed dataset and avoids computational overhead."
+    )]
     fn files_view(
         &self,
         predicate: Option<ExpressionRef>,
@@ -423,6 +437,7 @@ mod tests {
             test_files_view(snapshot.as_ref())?;
             test_commit_infos(snapshot.as_ref())?;
             test_logical_files(snapshot.as_ref())?;
+            test_logical_files_view(snapshot.as_ref())?;
         }
 
         let mut snapshot = get_snapshot(ctx, TestTables::Checkpoints, Some(0))?.await?;
@@ -434,6 +449,7 @@ mod tests {
             test_files_view(snapshot.as_ref())?;
             test_commit_infos(snapshot.as_ref())?;
             test_logical_files(snapshot.as_ref())?;
+            test_logical_files_view(snapshot.as_ref())?;
         }
 
         Ok(())
@@ -451,6 +467,15 @@ mod tests {
         Ok(())
     }
 
+    fn test_logical_files_view(snapshot: &dyn Snapshot) -> TestResult<()> {
+        let num_files_view = snapshot
+            .logical_files_view(None)?
+            .map(|f| f.unwrap().path().to_string())
+            .count() as u64;
+        assert_eq!(num_files_view, snapshot.version());
+        Ok(())
+    }
+
     fn test_files(snapshot: &dyn Snapshot) -> TestResult<()> {
         #[allow(deprecated)]
         let batches = snapshot.files(None)?.collect::<Result<Vec<_>, _>>()?;
@@ -460,6 +485,7 @@ mod tests {
     }
 
     fn test_files_view(snapshot: &dyn Snapshot) -> TestResult<()> {
+        #[allow(deprecated)]
         let num_files_view = snapshot
             .files_view(None)?
             .map(|f| f.unwrap().path().to_string())