refactor!: migrate Snapshot::try_new_from into SnapshotBuilder (delta-io#1289)

> [!IMPORTANT]
> This PR is split into two commits: (1) material changes, and (2)
callsite updates

This PR integrates `Snapshot::try_new_from` into `SnapshotBuilder`. We
move `Snapshot::try_new_from` into `Snapshot::builder_from` and modify
`SnapshotBuilder` to return an `Arc<Snapshot>` instead of a `Snapshot`.

Also, `Snapshot::builder(table_root)` is renamed to
`Snapshot::builder_for(table_root)`

1. Replaces `Snapshot::try_new_from(existing)` with
`Snapshot::build_from(existing).build()`
2. `SnapshotBuilder::build()` now returns an `Arc<Snapshot>` instead of
a `Snapshot`
3. Renames `Snapshot::builder` to `Snapshot::builder_for`

refactor, updated tests

Author: zachschuermann

acceptance/src/data.rs

@@ -115,8 +115,8 @@ pub async fn assert_scan_metadata(
     test_case: &TestCaseInfo,
 ) -> TestResult<()> {
     let table_root = test_case.table_root()?;
-    let snapshot = Snapshot::builder(table_root).build(engine.as_ref())?;
-    let scan = snapshot.into_scan_builder().build()?;
+    let snapshot = Snapshot::builder_for(table_root).build(engine.as_ref())?;
+    let scan = snapshot.scan_builder().build()?;
     let mut schema = None;
     let batches: Vec<RecordBatch> = scan
         .execute(engine)?

acceptance/src/meta.rs

@@ -103,11 +103,11 @@ impl TestCaseInfo {
         let engine = engine.as_ref();
         let (latest, versions) = self.versions().await?;
 
-        let snapshot = Snapshot::builder(self.table_root()?).build(engine)?;
+        let snapshot = Snapshot::builder_for(self.table_root()?).build(engine)?;
         self.assert_snapshot_meta(&latest, &snapshot)?;
 
         for table_version in versions {
-            let snapshot = Snapshot::builder(self.table_root()?)
+            let snapshot = Snapshot::builder_for(self.table_root()?)
                 .at_version(table_version.version)
                 .build(engine)?;
             self.assert_snapshot_meta(&table_version, &snapshot)?;

ffi/src/lib.rs

@@ -601,15 +601,15 @@ fn snapshot_impl(
     extern_engine: &dyn ExternEngine,
     version: Option<Version>,
 ) -> DeltaResult<Handle<SharedSnapshot>> {
-    let builder = Snapshot::builder(url?);
+    let builder = Snapshot::builder_for(url?);
     let builder = if let Some(v) = version {
         // TODO: should we include a `with_version_opt` method for the builder?
         builder.at_version(v)
     } else {
         builder
     };
     let snapshot = builder.build(extern_engine.engine().as_ref())?;
-    Ok(Arc::new(snapshot).into())
+    Ok(snapshot.into())
 }
 
 /// # Safety

ffi/src/transaction/mod.rs

@@ -9,7 +9,6 @@ use crate::{DeltaResult, ExternEngine, Snapshot, Url};
 use crate::{ExclusiveEngineData, SharedExternEngine};
 use delta_kernel::transaction::{CommitResult, Transaction};
 use delta_kernel_ffi_macros::handle_descriptor;
-use std::sync::Arc;
 
 /// A handle representing an exclusive transaction on a Delta table. (Similar to a Box<_>)
 ///
@@ -38,7 +37,7 @@ fn transaction_impl(
     url: DeltaResult<Url>,
     extern_engine: &dyn ExternEngine,
 ) -> DeltaResult<Handle<ExclusiveTransaction>> {
-    let snapshot = Arc::new(Snapshot::builder(url?).build(extern_engine.engine().as_ref())?);
+    let snapshot = Snapshot::builder_for(url?).build(extern_engine.engine().as_ref())?;
     let transaction = snapshot.transaction();
     Ok(Box::new(transaction?).into())
 }

kernel/benches/metadata_bench.rs

@@ -53,7 +53,7 @@ fn create_snapshot_benchmark(c: &mut Criterion) {
 
     c.bench_function("create_snapshot", |b| {
         b.iter(|| {
-            Snapshot::builder(url.clone())
+            Snapshot::builder_for(url.clone())
                 .build(engine.as_ref())
                 .expect("Failed to create snapshot")
         })
@@ -63,11 +63,9 @@ fn create_snapshot_benchmark(c: &mut Criterion) {
 fn scan_metadata_benchmark(c: &mut Criterion) {
     let (_tempdir, url, engine) = setup();
 
-    let snapshot = Arc::new(
-        Snapshot::builder(url.clone())
-            .build(engine.as_ref())
-            .expect("Failed to create snapshot"),
-    );
+    let snapshot = Snapshot::builder_for(url.clone())
+        .build(engine.as_ref())
+        .expect("Failed to create snapshot");
 
     let mut group = c.benchmark_group("scan_metadata");
     group.sample_size(SCAN_METADATA_BENCH_SAMPLE_SIZE);

kernel/examples/common/src/lib.rs

@@ -62,7 +62,7 @@ pub fn get_engine(
 
 /// Construct a scan at the latest snapshot. This is over the specified table and using the passed
 /// engine. Parameters of the scan are controlled by the specified `ScanArgs`
-pub fn get_scan(snapshot: Snapshot, args: &ScanArgs) -> DeltaResult<Option<Scan>> {
+pub fn get_scan(snapshot: Arc<Snapshot>, args: &ScanArgs) -> DeltaResult<Option<Scan>> {
     if args.schema_only {
         println!("{:#?}", snapshot.schema());
         return Ok(None);
@@ -86,7 +86,7 @@ pub fn get_scan(snapshot: Snapshot, args: &ScanArgs) -> DeltaResult<Option<Scan>
         .transpose()?;
     Ok(Some(
         snapshot
-            .into_scan_builder()
+            .scan_builder()
             .with_schema_opt(read_schema_opt)
             .build()?,
     ))

kernel/examples/inspect-table/src/main.rs

@@ -182,7 +182,7 @@ fn try_main() -> DeltaResult<()> {
 
     let url = delta_kernel::try_parse_uri(&cli.location_args.path)?;
     let engine = common::get_engine(&url, &cli.location_args)?;
-    let snapshot = Snapshot::builder(url).build(&engine)?;
+    let snapshot = Snapshot::builder_for(url).build(&engine)?;
 
     match cli.command {
         Commands::TableVersion => {

kernel/examples/read-table-multi-threaded/src/main.rs

@@ -99,7 +99,7 @@ fn try_main() -> DeltaResult<()> {
     let url = delta_kernel::try_parse_uri(&cli.location_args.path)?;
     println!("Reading {url}");
     let engine = common::get_engine(&url, &cli.location_args)?;
-    let snapshot = Snapshot::builder(url).build(&engine)?;
+    let snapshot = Snapshot::builder_for(url).build(&engine)?;
     let Some(scan) = common::get_scan(snapshot, &cli.scan_args)? else {
         return Ok(());
     };

kernel/examples/read-table-single-threaded/src/main.rs

@@ -43,7 +43,7 @@ fn try_main() -> DeltaResult<()> {
     let url = delta_kernel::try_parse_uri(&cli.location_args.path)?;
     println!("Reading {url}");
     let engine = common::get_engine(&url, &cli.location_args)?;
-    let snapshot = Snapshot::builder(url).build(&engine)?;
+    let snapshot = Snapshot::builder_for(url).build(&engine)?;
     let Some(scan) = common::get_scan(snapshot, &cli.scan_args)? else {
         return Ok(());
     };

kernel/examples/write-table/src/main.rs

@@ -79,7 +79,7 @@ async fn try_main() -> DeltaResult<()> {
     )?;
 
     // Create or get the table
-    let snapshot = Arc::new(create_or_get_base_snapshot(&url, &engine, &cli.schema).await?);
+    let snapshot = create_or_get_base_snapshot(&url, &engine, &cli.schema).await?;
 
     // Create sample data based on the schema
     let sample_data = create_sample_data(&snapshot.schema(), cli.num_rows)?;
@@ -123,9 +123,9 @@ async fn create_or_get_base_snapshot(
     url: &Url,
     engine: &dyn Engine,
     schema_str: &str,
-) -> DeltaResult<Snapshot> {
+) -> DeltaResult<Arc<Snapshot>> {
     // Check if table already exists
-    match Snapshot::builder(url.clone()).build(engine) {
+    match Snapshot::builder_for(url.clone()).build(engine) {
         Ok(snapshot) => {
             println!("✓ Found existing table at version {}", snapshot.version());
             Ok(snapshot)
@@ -135,7 +135,7 @@ async fn create_or_get_base_snapshot(
             println!("Creating new Delta table...");
             let schema = parse_schema(schema_str)?;
             create_table(url, &schema).await?;
-            Snapshot::builder(url.clone()).build(engine)
+            Snapshot::builder_for(url.clone()).build(engine)
         }
     }
 }
@@ -294,8 +294,8 @@ async fn read_and_display_data(
     table_url: &Url,
     engine: DefaultEngine<TokioBackgroundExecutor>,
 ) -> DeltaResult<()> {
-    let snapshot = Snapshot::builder(table_url.clone()).build(&engine)?;
-    let scan = snapshot.into_scan_builder().build()?;
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(&engine)?;
+    let scan = snapshot.scan_builder().build()?;
 
     let batches: Vec<RecordBatch> = scan
         .execute(Arc::new(engine))?