fix: reject cloning onto an existing local table (#4416)

Local `clone_table` passed existing targets to Lance's clone
transaction, which entered the manifest-update path and failed with
`Clone operation should not enter build_manifest`.

Check the target dataset using the connection's storage options and
session before cloning, and return `TableAlreadyExists` when it exists.
Propagate other read errors. Add regression coverage for cloning onto
the source itself and onto a different existing table, with both
manifest layouts, verifying that rows and version histories remain
unchanged.

Validation:

- All four regression cases reproduced the reported internal error
before the fix.
- `cargo test --locked --quiet --profile ci --features remote -p lancedb
--lib test_clone -j 12`: 24 passed.
- `cargo fmt --all -- --check` and `git diff --check` passed.
- `cargo clippy --locked --quiet --profile ci --features remote -p
lancedb --tests --examples -j 12` passed.

Fixes #4387

<!-- lance-gatekeeper-fix:v1 agent=522c80785eeaaef146e8179c848b903e
generation=1 -->

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>

rust/lancedb/src/connection.rs

@@ -568,6 +568,7 @@ impl Connection {
     /// Creates a new table by cloning from an existing source table.
     /// By default, this performs a shallow clone where the new table shares
     /// the underlying data files with the source table.
+    /// The target table name must be unused. An existing table is left unchanged.
     ///
     /// # Parameters
     /// - `target_table_name`: The name of the new table to create
@@ -2193,6 +2194,63 @@ mod tests {
         assert_eq!(tables.len(), 0);
     }
 
+    #[rstest::rstest]
+    #[tokio::test]
+    async fn test_clone_table_target_already_exists(
+        #[values("source", "target")] target: &str,
+        #[values(false, true)] enable_v2_manifest_paths: bool,
+    ) {
+        use crate::query::ExecutableQuery;
+        use futures::TryStreamExt;
+
+        let tmp_dir = tempdir().unwrap();
+        let options = ListingDatabaseOptions::builder()
+            .enable_v2_manifest_paths(enable_v2_manifest_paths)
+            .build();
+        let db = connect(tmp_dir.path().to_str().unwrap())
+            .database_options(&options)
+            .execute()
+            .await
+            .unwrap();
+        let source_data = arrow_array::record_batch!(("id", Int32, [0, 1, 2])).unwrap();
+        let target_data = arrow_array::record_batch!(("id", Int32, [10, 11])).unwrap();
+        db.create_table("source", source_data.clone())
+            .execute()
+            .await
+            .unwrap();
+        db.create_table("target", target_data.clone())
+            .execute()
+            .await
+            .unwrap();
+
+        let source_uri = tmp_dir.path().join("source.lance");
+        let err = db
+            .clone_table(target, source_uri.to_str().unwrap())
+            .execute()
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(&err, Error::TableAlreadyExists { name } if name == target),
+            "unexpected clone error: {err:?}"
+        );
+
+        // Reopen both tables to verify that the failed clone left their data and versions intact.
+        for (name, expected_data) in [("source", source_data), ("target", target_data)] {
+            let table = db.open_table(name).execute().await.unwrap();
+            assert_eq!(table.version().await.unwrap(), 1);
+            assert_eq!(table.list_versions().await.unwrap().len(), 1);
+            let batches = table
+                .query()
+                .execute()
+                .await
+                .unwrap()
+                .try_collect::<Vec<_>>()
+                .await
+                .unwrap();
+            assert_eq!(batches, vec![expected_data]);
+        }
+    }
+
     #[tokio::test]
     async fn test_clone_table() {
         let tmp_dir = tempdir().unwrap();

rust/lancedb/src/database/listing.rs

@@ -1107,6 +1107,21 @@ impl Database for ListingDatabase {
             ..Default::default()
         };
 
+        let target_uri = self.table_uri(&request.target_table_name)?;
+        match DatasetBuilder::from_uri(&target_uri)
+            .with_read_params(read_params.clone())
+            .load()
+            .await
+        {
+            Ok(_) => {
+                return Err(Error::TableAlreadyExists {
+                    name: request.target_table_name,
+                });
+            }
+            Err(lance::Error::DatasetNotFound { .. }) => {}
+            Err(err) => return Err(err.into()),
+        }
+
         let mut source_dataset = DatasetBuilder::from_uri(&request.source_uri)
             .with_read_params(read_params.clone())
             .load()
@@ -1122,7 +1137,6 @@ impl Database for ListingDatabase {
             }),
         }?;
 
-        let target_uri = self.table_uri(&request.target_table_name)?;
         source_dataset
             .shallow_clone(&target_uri, version_ref, Some(storage_params))
             .await