wip: rearrange traits

rust/lance-table/src/io/manifest.rs

@@ -6,6 +6,7 @@ use std::{ops::Range, sync::Arc};
 use async_trait::async_trait;
 use byteorder::{ByteOrder, LittleEndian};
 use bytes::{Bytes, BytesMut};
+use futures::future::BoxFuture;
 use lance_arrow::DataTypeExt;
 use lance_file::{version::LanceFileVersion, writer::ManifestProvider};
 use object_store::path::Path;
@@ -22,7 +23,7 @@ use lance_io::{
     utils::read_message,
 };
 
-use crate::format::{pb, DataStorageFormat, Index, Manifest, MAGIC};
+use crate::format::{pb, DataStorageFormat, Index, Manifest, MAGIC,  MAJOR_VERSION, MINOR_VERSION};
 
 /// Read Manifest on URI.
 ///
@@ -116,10 +117,10 @@ pub async fn read_manifest_indexes(
 async fn do_write_manifest(
     writer: &mut dyn Writer,
     manifest: &mut Manifest,
-    indices: Option<Vec<Index>>,
+    indices: &[Index],
 ) -> Result<usize> {
     // Write indices if presented.
-    if let Some(indices) = indices.as_ref() {
+    if !indices.is_empty() {
         let section = pb::IndexSection {
             indices: indices.iter().map(|i| i.into()).collect(),
         };
@@ -134,7 +135,7 @@ async fn do_write_manifest(
 pub async fn write_manifest(
     writer: &mut dyn Writer,
     manifest: &mut Manifest,
-    indices: Option<Vec<Index>>,
+    indices: &[Index],
 ) -> Result<usize> {
     // Write dictionary values.
     let max_field_id = manifest.schema.max_field_id().unwrap_or(-1);
@@ -187,6 +188,23 @@ pub async fn write_manifest(
     do_write_manifest(writer, manifest, indices).await
 }
 
+pub fn write_manifest_file_to_path<'a>(
+    object_store: &'a ObjectStore,
+    manifest: &'a mut Manifest,
+    indices: &'a [Index],
+    path: &'a Path,
+) -> BoxFuture<'a, Result<()>> {
+    Box::pin(async {
+        let mut object_writer = ObjectWriter::new(object_store, path).await?;
+        let pos = write_manifest(&mut object_writer, manifest, indices).await?;
+        object_writer
+            .write_magics(pos, MAJOR_VERSION, MINOR_VERSION, MAGIC)
+            .await?;
+        object_writer.shutdown().await?;
+        Ok(())
+    })
+}
+
 /// Implementation of ManifestProvider that describes a Lance file by writing
 /// a manifest that contains nothing but default fields and the schema
 pub struct ManifestDescribing {}
@@ -202,7 +220,7 @@ impl ManifestProvider for ManifestDescribing {
             Arc::new(vec![]),
             DataStorageFormat::new(LanceFileVersion::Legacy),
         );
-        let pos = do_write_manifest(object_writer, &mut manifest, None).await?;
+        let pos = do_write_manifest(object_writer, &mut manifest, &[]).await?;
         Ok(Some(pos))
     }
 }
@@ -244,7 +262,7 @@ mod test {
             ArrowSchema::new(vec![ArrowField::new(long_name, DataType::Int64, false)]);
         let schema = Schema::try_from(&arrow_schema).unwrap();
         let mut manifest = Manifest::new(schema, Arc::new(vec![]), DataStorageFormat::default());
-        let pos = write_manifest(&mut writer, &mut manifest, None)
+        let pos = write_manifest(&mut writer, &mut manifest, &[])
             .await
             .unwrap();
         writer

rust/lance-table/src/lib.rs

@@ -4,5 +4,6 @@
 pub mod feature_flags;
 pub mod format;
 pub mod io;
+pub mod manifest_store;
 pub mod rowids;
 pub mod utils;

rust/lance-table/src/manifest_store.rs

@@ -0,0 +1,97 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! [ManifestStore] encapsulates the logic for reading, writing and managing
+//! manifest files. Different implementations may be used, depending on the
+//! capabilities of the underlying storage system.
+
+use futures::future::Future;
+
+use futures::stream::BoxStream;
+use lance_core::{Error, Result};
+use lance_io::traits::Reader;
+
+use crate::format::{Index, Manifest};
+
+pub mod legacy;
+#[cfg(feature = "dynamodb")]
+pub mod dynamodb;
+
+const MANIFEST_EXTENSION: &str = "manifest";
+
+/// A store of manifests. This provides fast access to the latest version
+/// of the dataset and allows for listing and opening older versions.
+pub trait ManifestStore {
+    /// Get the latest version of the dataset.
+    fn latest_version(&self) -> impl Future<Output = Result<ManifestVersion>>;
+
+    /// Open the latest manifest file.
+    fn open_latest_manifest(&self) -> impl Future<Output = Result<Box<dyn Reader>>>;
+
+    /// Open the manifest file for the given version.
+    ///
+    /// Should use the provided size if available to avoid an extra HEAD request.
+    fn open_manifest(&self, version: impl Into<ManifestVersion>) -> impl Future<Output = Result<Box<dyn Reader>>>;
+
+    /// List all the versions of the dataset.
+    ///
+    /// This should return them in descending order.
+    fn list_versions(&self) -> BoxStream<Result<ManifestVersion>>;
+
+    /// Try to commit the given manifest as the given version.
+    ///
+    /// If the version already exists, this should return an error, even if
+    /// the version was created by a concurrent process.
+    ///
+    /// Any temporary files created during the commit should be cleaned up
+    /// if the commit fails.
+    /// 
+    /// The `manifest` is mutable because the offsets to certain internal
+    /// structures are updated during the writing process.
+    fn try_commit(
+        &self,
+        manifest: &mut Manifest,
+        indices: &[Index],
+    ) -> impl Future<Output = std::result::Result<(), CommitError>>;
+
+    // TODO: what about cleanup?
+}
+
+pub struct ManifestVersion {
+    version: u64,
+    known_size: Option<u64>,
+}
+
+impl From<u64> for ManifestVersion {
+    fn from(version: u64) -> Self {
+        Self {
+            version,
+            known_size: None,
+        }
+    }
+}
+
+
+/// Errors that can occur when committing a manifest.
+#[derive(Debug)]
+pub enum CommitError {
+    /// Another transaction has already been written to the path
+    CommitConflict,
+    /// Something else went wrong
+    OtherError(Error),
+}
+
+impl From<Error> for CommitError {
+    fn from(e: Error) -> Self {
+        Self::OtherError(e)
+    }
+}
+
+// Goal: make the paths opaque, so that the store implementation can choose how 
+// the paths are represented.
+
+// Goal 2: separate idea of commit handler (what happens when we write the manifest)
+// from the idea of the store (how we read the manifests). Allow customizing both.
+
+// This is really just a cleaned up version of CommitHandler. We can provide
+// an adapter for now.

rust/lance-table/src/manifest_store/dynamodb.rs

@@ -0,0 +1,3 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+