fix: adding namespace support to java SDK CommitBuilder from dataset (#6257)

Currently, the namespace / tableID settings are not available when
initializing a CommitBuilder through the Java SDK (although they are
from the URI variant). This is problematic in scenarios where we need to
update the CommitHandler functionality. For example, when managed
versioning is disabled on a REST namespace implementation we override
the CommitHandler to commit directly to object store rather than route
through the REST namespace.

Author: hamersaw

java/lance-jni/src/blocking_dataset.rs

@@ -341,6 +341,7 @@ impl BlockingDataset {
         storage_format: Option<LanceFileVersion>,
         max_retries: u32,
         skip_auto_cleanup: bool,
+        commit_handler: Option<Arc<dyn CommitHandler>>,
     ) -> Result<Self> {
         let mut builder = CommitBuilder::new(Arc::new(self.clone().inner))
             .with_store_params(store_params)
@@ -358,6 +359,9 @@ impl BlockingDataset {
         if skip_auto_cleanup {
             builder = builder.with_skip_auto_cleanup(true);
         }
+        if let Some(handler) = commit_handler {
+            builder = builder.with_commit_handler(handler);
+        }
         let new_dataset = RT.block_on(builder.execute(transaction))?;
         Ok(BlockingDataset { inner: new_dataset })
     }

java/lance-jni/src/transaction.rs

@@ -623,6 +623,8 @@ pub extern "system" fn Java_org_lance_CommitBuilder_nativeCommitToDataset<'local
     storage_format_obj: JObject,
     max_retries: jint,
     skip_auto_cleanup: jboolean,
+    namespace_obj: JObject,
+    table_id_obj: JObject,
 ) -> JObject<'local> {
     ok_or_throw!(
         env,
@@ -637,6 +639,8 @@ pub extern "system" fn Java_org_lance_CommitBuilder_nativeCommitToDataset<'local
             storage_format_obj,
             max_retries as u32,
             skip_auto_cleanup != 0,
+            namespace_obj,
+            table_id_obj,
         )
     )
 }
@@ -653,6 +657,8 @@ fn inner_commit_to_dataset<'local>(
     storage_format_obj: JObject,
     max_retries: u32,
     skip_auto_cleanup: bool,
+    namespace_obj: JObject,
+    table_id_obj: JObject,
 ) -> Result<JObject<'local>> {
     let write_param = if write_params_obj.is_null() {
         HashMap::new()
@@ -745,6 +751,15 @@ fn inner_commit_to_dataset<'local>(
         Some(&mut java_blocking_ds),
     )?;
 
+    // Set namespace commit handler if provided
+    let namespace_info = extract_namespace_info(env, &namespace_obj, &table_id_obj)?;
+    let commit_handler = namespace_info.map(|(ns, tid)| {
+        let external_store = LanceNamespaceExternalManifestStore::new(ns, tid);
+        Arc::new(ExternalManifestCommitHandler {
+            external_manifest_store: Arc::new(external_store),
+        }) as Arc<dyn CommitHandler>
+    });
+
     let new_blocking_ds = {
         let mut dataset_guard =
             unsafe { env.get_rust_field::<_, _, BlockingDataset>(&java_dataset, NATIVE_DATASET) }?;
@@ -757,6 +772,7 @@ fn inner_commit_to_dataset<'local>(
             storage_format,
             max_retries,
             skip_auto_cleanup,
+            commit_handler,
         )?
     };
     new_blocking_ds.into_java(env)

java/src/main/java/org/lance/CommitBuilder.java

@@ -124,7 +124,9 @@ public CommitBuilder storageOptionsProvider(StorageOptionsProvider provider) {
   }
 
   /**
-   * Set the namespace for managed versioning during URI-based commits.
+   * Set the namespace for managed versioning. When set, commits are routed through the namespace's
+   * {@code createTableVersion} API instead of writing directly to the object store. This is
+   * supported for both dataset-based and URI-based commits.
    *
    * @param namespace the LanceNamespace instance
    * @return this builder instance
@@ -250,7 +252,9 @@ public Dataset execute(Transaction transaction) {
               useStableRowIds,
               storageFormat,
               maxRetries,
-              skipAutoCleanup);
+              skipAutoCleanup,
+              namespace,
+              tableId);
       result.setAllocator(dataset.allocator());
       return result;
     }
@@ -285,7 +289,9 @@ private static native Dataset nativeCommitToDataset(
       Boolean useStableRowIds,
       String storageFormat,
       int maxRetries,
-      boolean skipAutoCleanup);
+      boolean skipAutoCleanup,
+      Object namespace,
+      Object tableId);
 
   private static native Dataset nativeCommitToUri(
       String uri,

java/src/test/java/org/lance/namespace/DirectoryNamespaceTest.java

@@ -13,12 +13,17 @@
  */
 package org.lance.namespace;
 
+import org.lance.CommitBuilder;
 import org.lance.Dataset;
+import org.lance.Fragment;
+import org.lance.FragmentMetadata;
 import org.lance.ReadOptions;
+import org.lance.Transaction;
 import org.lance.WriteParams;
 import org.lance.namespace.model.*;
 import org.lance.namespace.model.DescribeTableVersionRequest;
 import org.lance.namespace.model.DescribeTableVersionResponse;
+import org.lance.operation.Append;
 
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
@@ -673,6 +678,148 @@ public VectorSchemaRoot getVectorSchemaRoot() {
     }
   }
 
+  @Test
+  void testDatasetBasedCommitBuilderWithNamespace(@TempDir Path managedVersioningTempDir)
+      throws Exception {
+    try (BufferAllocator allocator = new RootAllocator()) {
+      TableVersionTrackingNamespace namespace =
+          new TableVersionTrackingNamespace(managedVersioningTempDir);
+      String tableName = "test_table";
+      List<String> tableId = Arrays.asList(tableName);
+
+      Schema schema =
+          new Schema(
+              Arrays.asList(
+                  new Field("id", FieldType.nullable(new ArrowType.Int(32, true)), null),
+                  new Field("name", FieldType.nullable(new ArrowType.Utf8()), null)));
+
+      // Create initial dataset through namespace using WriteDatasetBuilder
+      try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+        IntVector idVector = (IntVector) root.getVector("id");
+        VarCharVector nameVector = (VarCharVector) root.getVector("name");
+
+        idVector.allocateNew(2);
+        nameVector.allocateNew(2);
+        idVector.set(0, 1);
+        idVector.set(1, 2);
+        nameVector.set(0, "Alice".getBytes());
+        nameVector.set(1, "Bob".getBytes());
+        idVector.setValueCount(2);
+        nameVector.setValueCount(2);
+        root.setRowCount(2);
+
+        ArrowReader reader =
+            new ArrowReader(allocator) {
+              boolean firstRead = true;
+
+              @Override
+              public boolean loadNextBatch() {
+                if (firstRead) {
+                  firstRead = false;
+                  return true;
+                }
+                return false;
+              }
+
+              @Override
+              public long bytesRead() {
+                return 0;
+              }
+
+              @Override
+              protected void closeReadSource() {}
+
+              @Override
+              protected Schema readSchema() {
+                return schema;
+              }
+
+              @Override
+              public VectorSchemaRoot getVectorSchemaRoot() {
+                return root;
+              }
+            };
+
+        try (Dataset dataset =
+            Dataset.write()
+                .allocator(allocator)
+                .reader(reader)
+                .namespace(namespace)
+                .tableId(tableId)
+                .mode(WriteParams.WriteMode.CREATE)
+                .execute()) {
+          assertEquals(2, dataset.countRows());
+          assertEquals(1, dataset.version());
+        }
+      }
+
+      // Verify initial create used createTableVersion once
+      assertEquals(
+          1,
+          namespace.getCreateTableVersionCount(),
+          "create_table_version should be called once during CREATE");
+
+      // Open dataset through namespace (returns dataset with managed versioning)
+      Dataset existingDataset =
+          Dataset.open().allocator(allocator).namespace(namespace).tableId(tableId).build();
+
+      // Get the dataset URI for Fragment.create()
+      String datasetUri = existingDataset.uri();
+
+      // Create a new fragment independently (simulating Spark worker behavior)
+      List<FragmentMetadata> fragments;
+      try (VectorSchemaRoot appendRoot = VectorSchemaRoot.create(schema, allocator)) {
+        IntVector idVector = (IntVector) appendRoot.getVector("id");
+        VarCharVector nameVector = (VarCharVector) appendRoot.getVector("name");
+
+        idVector.allocateNew(2);
+        nameVector.allocateNew(2);
+        idVector.set(0, 3);
+        idVector.set(1, 4);
+        nameVector.set(0, "Charlie".getBytes());
+        nameVector.set(1, "Diana".getBytes());
+        idVector.setValueCount(2);
+        nameVector.setValueCount(2);
+        appendRoot.setRowCount(2);
+
+        fragments =
+            Fragment.create(datasetUri, allocator, appendRoot, new WriteParams.Builder().build());
+      }
+
+      // Commit using dataset-based CommitBuilder WITH namespace (the new path)
+      int createCountBefore = namespace.getCreateTableVersionCount();
+      try (Transaction txn =
+              new Transaction.Builder()
+                  .readVersion(existingDataset.version())
+                  .operation(Append.builder().fragments(fragments).build())
+                  .build();
+          Dataset committed =
+              new CommitBuilder(existingDataset)
+                  .namespace(namespace)
+                  .tableId(tableId)
+                  .execute(txn)) {
+        assertEquals(2, committed.version());
+        assertEquals(4, committed.countRows());
+      }
+
+      // Verify createTableVersion was called for the dataset-based commit
+      assertEquals(
+          createCountBefore + 1,
+          namespace.getCreateTableVersionCount(),
+          "create_table_version should be called for dataset-based CommitBuilder with namespace");
+
+      // Verify the data is accessible through namespace
+      try (Dataset latestDs =
+          Dataset.open().allocator(allocator).namespace(namespace).tableId(tableId).build()) {
+        assertEquals(4, latestDs.countRows());
+        assertEquals(2, latestDs.version());
+      }
+
+      existingDataset.close();
+      namespace.close();
+    }
+  }
+
   @Test
   void testConcurrentCreateAndDropWithSingleInstance() throws Exception {
     // Initialize namespace first - create parent namespace to ensure __manifest table