fix!: get prefix from offset path (#699)

## What changes are proposed in this pull request?

Our `list_from` implementation for the object_store based filesystem
client is currently broken, since it does not behave as documented /
required for that function. Specifically we should list all files in the
parent folder for using the path as offset to list from.

In a follow up PR we then need to lift the assumption that all URLs will
always be under the same store to get proper URL handling.

### This PR affects the following public APIs

- `DefaultEngine::new` no longer requires a `table_root` parameter.
- `list_from` consistently returns keys greater than (`>`) the offset,
previously the `sync-engines` client returned all keys (`>=`)

## How was this change tested?

Additional tests asserting consistent `list_from` behavior for all our
file system client implementations.

---------

Signed-off-by: Robert Pack <[email protected]>
Co-authored-by: Ryan Johnson <[email protected]>
Co-authored-by: Zach Schuermann <[email protected]>

Author: 3 people

ffi/src/lib.rs

@@ -765,7 +765,7 @@ impl<T> Default for ReferenceSet<T> {
 #[cfg(test)]
 mod tests {
     use delta_kernel::engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine};
-    use object_store::{memory::InMemory, path::Path};
+    use object_store::memory::InMemory;
     use test_utils::{actions_to_string, actions_to_string_partitioned, add_commit, TestAction};
 
     use super::*;
@@ -839,11 +839,7 @@ mod tests {
             actions_to_string(vec![TestAction::Metadata]),
         )
         .await?;
-        let engine = DefaultEngine::new(
-            storage.clone(),
-            Path::from("/"),
-            Arc::new(TokioBackgroundExecutor::new()),
-        );
+        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
         let path = "memory:///";
 
@@ -872,11 +868,7 @@ mod tests {
             actions_to_string_partitioned(vec![TestAction::Metadata]),
         )
         .await?;
-        let engine = DefaultEngine::new(
-            storage.clone(),
-            Path::from("/"),
-            Arc::new(TokioBackgroundExecutor::new()),
-        );
+        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
         let engine = engine_to_handle(Arc::new(engine), allocate_err);
         let path = "memory:///";
 

kernel/src/engine/default/filesystem.rs

@@ -14,7 +14,6 @@ use crate::{DeltaResult, Error, FileMeta, FileSlice, FileSystemClient};
 pub struct ObjectStoreFileSystemClient<E: TaskExecutor> {
     inner: Arc<DynObjectStore>,
     has_ordered_listing: bool,
-    table_root: Path,
     task_executor: Arc<E>,
     readahead: usize,
 }
@@ -23,13 +22,11 @@ impl<E: TaskExecutor> ObjectStoreFileSystemClient<E> {
     pub(crate) fn new(
         store: Arc<DynObjectStore>,
         has_ordered_listing: bool,
-        table_root: Path,
         task_executor: Arc<E>,
     ) -> Self {
         Self {
             inner: store,
             has_ordered_listing,
-            table_root,
             task_executor,
             readahead: 10,
         }
@@ -47,16 +44,28 @@ impl<E: TaskExecutor> FileSystemClient for ObjectStoreFileSystemClient<E> {
         &self,
         path: &Url,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<FileMeta>>>> {
-        let url = path.clone();
-        let offset = Path::from(path.path());
-        // TODO properly handle table prefix
-        let prefix = self.table_root.child("_delta_log");
+        // The offset is used for list-after; the prefix is used to restrict the listing to a specific directory.
+        // Unfortunately, `Path` provides no easy way to check whether a name is directory-like,
+        // because it strips trailing /, so we're reduced to manually checking the original URL.
+        let offset = Path::from_url_path(path.path())?;
+        let prefix = if path.path().ends_with('/') {
+            offset.clone()
+        } else {
+            let mut parts = offset.parts().collect_vec();
+            if parts.pop().is_none() {
+                return Err(Error::Generic(format!(
+                    "Offset path must not be a root directory. Got: '{}'",
+                    path.as_str()
+                )));
+            }
+            Path::from_iter(parts)
+        };
 
         let store = self.inner.clone();
 
         // This channel will become the iterator
         let (sender, receiver) = std::sync::mpsc::sync_channel(4_000);
-
+        let url = path.clone();
         self.task_executor.spawn(async move {
             let mut stream = store.list_with_offset(Some(&prefix), &offset);
 
@@ -192,11 +201,9 @@ mod tests {
         let mut url = Url::from_directory_path(tmp.path()).unwrap();
 
         let store = Arc::new(LocalFileSystem::new());
-        let prefix = Path::from(url.path());
         let client = ObjectStoreFileSystemClient::new(
             store,
             false, // don't have ordered listing
-            prefix,
             Arc::new(TokioBackgroundExecutor::new()),
         );
 
@@ -229,11 +236,10 @@ mod tests {
         store.put(&name, data.clone().into()).await.unwrap();
 
         let table_root = Url::parse("memory:///").expect("valid url");
-        let prefix = Path::from_url_path(table_root.path()).expect("Couldn't get path");
-        let engine = DefaultEngine::new(store, prefix, Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
         let files: Vec<_> = engine
             .get_file_system_client()
-            .list_from(&table_root)
+            .list_from(&table_root.join("_delta_log").unwrap().join("0").unwrap())
             .unwrap()
             .try_collect()
             .unwrap();
@@ -260,11 +266,12 @@ mod tests {
 
         let url = Url::from_directory_path(tmp.path()).unwrap();
         let store = Arc::new(LocalFileSystem::new());
-        let prefix = Path::from_url_path(url.path()).expect("Couldn't get path");
-        let engine = DefaultEngine::new(store, prefix, Arc::new(TokioBackgroundExecutor::new()));
+        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
         let client = engine.get_file_system_client();
 
-        let files = client.list_from(&Url::parse("file://").unwrap()).unwrap();
+        let files = client
+            .list_from(&url.join("_delta_log").unwrap().join("0").unwrap())
+            .unwrap();
         let mut len = 0;
         for (file, expected) in files.zip(expected_names.iter()) {
             assert!(

kernel/src/engine/default/json.rs

@@ -142,7 +142,7 @@ impl<E: TaskExecutor> JsonHandler for DefaultJsonHandler<E> {
         let buffer = to_json_bytes(data)?;
         // Put if absent
         let store = self.store.clone(); // cheap Arc
-        let path = Path::from(path.path());
+        let path = Path::from_url_path(path.path())?;
         let path_str = path.to_string();
         self.task_executor
             .block_on(async move {

kernel/src/engine/default/mod.rs

@@ -10,7 +10,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use self::storage::parse_url_opts;
-use object_store::{path::Path, DynObjectStore};
+use object_store::DynObjectStore;
 use url::Url;
 
 use self::executor::TaskExecutor;
@@ -60,8 +60,8 @@ impl<E: TaskExecutor> DefaultEngine<E> {
         V: Into<String>,
     {
         // table root is the path of the table in the ObjectStore
-        let (store, table_root) = parse_url_opts(table_root, options)?;
-        Ok(Self::new(Arc::new(store), table_root, task_executor))
+        let (store, _table_root) = parse_url_opts(table_root, options)?;
+        Ok(Self::new(Arc::new(store), task_executor))
     }
 
     /// Create a new [`DefaultEngine`] instance
@@ -71,7 +71,7 @@ impl<E: TaskExecutor> DefaultEngine<E> {
     /// - `store`: The object store to use.
     /// - `table_root_path`: The root path of the table within storage.
     /// - `task_executor`: Used to spawn async IO tasks. See [executor::TaskExecutor].
-    pub fn new(store: Arc<DynObjectStore>, table_root: Path, task_executor: Arc<E>) -> Self {
+    pub fn new(store: Arc<DynObjectStore>, task_executor: Arc<E>) -> Self {
         // HACK to check if we're using a LocalFileSystem from ObjectStore. We need this because
         // local filesystem doesn't return a sorted list by default. Although the `object_store`
         // crate explicitly says it _does not_ return a sorted listing, in practice all the cloud
@@ -97,7 +97,6 @@ impl<E: TaskExecutor> DefaultEngine<E> {
             file_system: Arc::new(ObjectStoreFileSystemClient::new(
                 store.clone(),
                 !is_local,
-                table_root,
                 task_executor.clone(),
             )),
             json: Arc::new(DefaultJsonHandler::new(
@@ -158,3 +157,20 @@ impl<E: TaskExecutor> Engine for DefaultEngine<E> {
         self.parquet.clone()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::executor::tokio::TokioBackgroundExecutor;
+    use super::*;
+    use crate::engine::tests::test_arrow_engine;
+    use object_store::local::LocalFileSystem;
+
+    #[test]
+    fn test_default_engine() {
+        let tmp = tempfile::tempdir().unwrap();
+        let url = Url::from_directory_path(tmp.path()).unwrap();
+        let store = Arc::new(LocalFileSystem::new());
+        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
+        test_arrow_engine(&engine, &url);
+    }
+}

kernel/src/engine/mod.rs

@@ -27,3 +27,80 @@ pub(crate) mod arrow_get_data;
 pub(crate) mod ensure_data_types;
 #[cfg(any(feature = "default-engine-base", feature = "sync-engine"))]
 pub mod parquet_row_group_skipping;
+
+#[cfg(test)]
+mod tests {
+    use itertools::Itertools;
+    use object_store::path::Path;
+    use std::sync::Arc;
+    use url::Url;
+
+    use crate::arrow::array::{RecordBatch, StringArray};
+    use crate::arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+    use crate::engine::arrow_data::ArrowEngineData;
+    use crate::{Engine, EngineData};
+
+    use test_utils::delta_path_for_version;
+
+    fn test_list_from_should_sort_and_filter(
+        engine: &dyn Engine,
+        base_url: &Url,
+        engine_data: impl Fn() -> Box<dyn EngineData>,
+    ) {
+        let json = engine.get_json_handler();
+        let get_data = || Box::new(std::iter::once(Ok(engine_data())));
+
+        let expected_names: Vec<Path> = (1..4)
+            .map(|i| delta_path_for_version(i, "json"))
+            .collect_vec();
+
+        for i in expected_names.iter().rev() {
+            let path = base_url.join(i.as_ref()).unwrap();
+            json.write_json_file(&path, get_data(), false).unwrap();
+        }
+        let path = base_url.join("other").unwrap();
+        json.write_json_file(&path, get_data(), false).unwrap();
+
+        let fs = engine.get_file_system_client();
+
+        // list files after an offset
+        let test_url = base_url.join(expected_names[0].as_ref()).unwrap();
+        let files: Vec<_> = fs.list_from(&test_url).unwrap().try_collect().unwrap();
+        assert_eq!(files.len(), expected_names.len() - 1);
+        for (file, expected) in files.iter().zip(expected_names.iter().skip(1)) {
+            assert_eq!(file.location, base_url.join(expected.as_ref()).unwrap());
+        }
+
+        let test_url = base_url
+            .join(delta_path_for_version(0, "json").as_ref())
+            .unwrap();
+        let files: Vec<_> = fs.list_from(&test_url).unwrap().try_collect().unwrap();
+        assert_eq!(files.len(), expected_names.len());
+
+        // list files inside a directory / key prefix
+        let test_url = base_url.join("_delta_log/").unwrap();
+        let files: Vec<_> = fs.list_from(&test_url).unwrap().try_collect().unwrap();
+        assert_eq!(files.len(), expected_names.len());
+        for (file, expected) in files.iter().zip(expected_names.iter()) {
+            assert_eq!(file.location, base_url.join(expected.as_ref()).unwrap());
+        }
+    }
+
+    fn get_arrow_data() -> Box<dyn EngineData> {
+        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "dog",
+            ArrowDataType::Utf8,
+            true,
+        )]));
+        let data = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(StringArray::from(vec!["remi", "wilson"]))],
+        )
+        .unwrap();
+        Box::new(ArrowEngineData::new(data))
+    }
+
+    pub(crate) fn test_arrow_engine(engine: &dyn Engine, base_url: &Url) {
+        test_list_from_should_sort_and_filter(engine, base_url, get_arrow_data);
+    }
+}

kernel/src/engine/sync/fs_client.rs

@@ -39,7 +39,7 @@ impl FileSystemClient for SyncFilesystemClient {
             let all_ents: Vec<_> = std::fs::read_dir(path_to_read)?
                 .filter(|ent_res| {
                     match (ent_res, min_file_name) {
-                        (Ok(ent), Some(min_file_name)) => ent.file_name() >= *min_file_name,
+                        (Ok(ent), Some(min_file_name)) => ent.file_name() > *min_file_name,
                         _ => true, // Keep unfiltered and/or error entries
                     }
                 })
@@ -106,7 +106,7 @@ mod tests {
         writeln!(f, "null")?;
         f.flush()?;
 
-        let url_path = tmp_dir.path().join(get_json_filename(1));
+        let url_path = tmp_dir.path().join(get_json_filename(0));
         let url = Url::from_file_path(url_path).unwrap();
         let files: Vec<_> = client.list_from(&url)?.try_collect()?;
 
@@ -137,11 +137,11 @@ mod tests {
             // i+1 in index because we started at 0001 in the listing
             assert_eq!(
                 file?.location.to_file_path().unwrap().to_str().unwrap(),
-                expected[i + 1].to_str().unwrap()
+                expected[i + 2].to_str().unwrap()
             );
             file_count += 1;
         }
-        assert_eq!(file_count, 2);
+        assert_eq!(file_count, 1);
 
         let url_path = tmp_dir.path().join("");
         let url = Url::from_file_path(url_path).unwrap();

kernel/src/engine/sync/json.rs

@@ -66,6 +66,10 @@ impl JsonHandler for SyncJsonHandler {
             )));
         };
 
+        if !parent.exists() {
+            std::fs::create_dir_all(parent)?;
+        }
+
         // write data to tmp file
         let mut tmp_file = NamedTempFile::new_in(parent)?;
         let buf = to_json_bytes(data)?;

kernel/src/engine/sync/mod.rs

@@ -97,3 +97,17 @@ where
         .map(|data| Ok(Box::new(ArrowEngineData::new(data??.into())) as _));
     Ok(Box::new(result))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::engine::tests::test_arrow_engine;
+
+    #[test]
+    fn test_sync_engine() {
+        let tmp = tempfile::tempdir().unwrap();
+        let url = url::Url::from_directory_path(tmp.path()).unwrap();
+        let engine = SyncEngine::new();
+        test_arrow_engine(&engine, &url);
+    }
+}

kernel/src/lib.rs

@@ -43,7 +43,7 @@
 //!
 //! Delta Kernel needs to perform some basic operations against file systems like listing and
 //! reading files. These interactions are encapsulated in the [`FileSystemClient`] trait.
-//! Implementors must take care that all assumptions on the behavior if the functions - like sorted
+//! Implementers must take care that all assumptions on the behavior if the functions - like sorted
 //! results - are respected.
 //!
 //! ## Reading log and data files
@@ -348,8 +348,11 @@ pub trait ExpressionHandler: AsAny {
 /// file system where the Delta table is present. Connector implementation of
 /// this trait can hide filesystem specific details from Delta Kernel.
 pub trait FileSystemClient: AsAny {
-    /// List the paths in the same directory that are lexicographically greater or equal to
+    /// List the paths in the same directory that are lexicographically greater than
     /// (UTF-8 sorting) the given `path`. The result should also be sorted by the file name.
+    ///
+    /// If the path is directory-like (ends with '/'), the result should contain
+    /// all the files in the directory.
     fn list_from(&self, path: &Url)
         -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<FileMeta>>>>;