feat(java): support row lineage and cdf apis

java/lance-jni/src/delta.rs

@@ -0,0 +1,206 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use crate::blocking_dataset::{BlockingDataset, NATIVE_DATASET};
+use crate::error::Result;
+use crate::ffi::JNIEnvExt;
+use crate::transaction::convert_to_java_transaction;
+use crate::RT;
+use arrow::ffi_stream::FFI_ArrowArrayStream;
+use jni::objects::{JObject, JValue};
+use jni::sys::jlong;
+use jni::JNIEnv;
+use lance::dataset::delta::DatasetDelta as RustDatasetDelta;
+use lance::dataset::scanner::DatasetRecordBatchStream;
+use lance::dataset::transaction::Transaction;
+use lance_io::ffi::to_ffi_arrow_array_stream;
+
+pub const NATIVE_DELTA: &str = "nativeDeltaHandle";
+
+pub struct BlockingDatasetDelta {
+    pub(crate) inner: RustDatasetDelta,
+}
+
+fn attach_native_delta<'local>(
+    env: &mut JNIEnv<'local>,
+    delta: BlockingDatasetDelta,
+    java_dataset: &JObject<'local>,
+) -> Result<JObject<'local>> {
+    let j_delta = env.new_object("org/lance/delta/DatasetDelta", "()V", &[])?;
+
+    unsafe { env.set_rust_field(&j_delta, NATIVE_DELTA, delta) }?;
+
+    env.set_field(
+        &j_delta,
+        "dataset",
+        "Lorg/lance/Dataset;",
+        JValue::Object(java_dataset),
+    )?;
+    Ok(j_delta)
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_delta_DatasetDeltaBuilder_nativeBuild<'local>(
+    mut env: JNIEnv<'local>,
+    _obj: JObject<'local>,
+    java_dataset: JObject<'local>,
+    compared_against_obj: JObject<'local>,
+    begin_version_obj: JObject<'local>,
+    end_version_obj: JObject<'local>,
+) -> JObject<'local> {
+    ok_or_throw!(
+        env,
+        inner_native_build(
+            &mut env,
+            java_dataset,
+            compared_against_obj,
+            begin_version_obj,
+            end_version_obj
+        )
+    )
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_Dataset_nativeBuildDelta<'local>(
+    mut env: JNIEnv<'local>,
+    java_dataset: JObject<'local>,
+    compared_against_obj: JObject<'local>,
+    begin_version_obj: JObject<'local>,
+    end_version_obj: JObject<'local>,
+) -> JObject<'local> {
+    ok_or_throw!(
+        env,
+        inner_native_build(
+            &mut env,
+            java_dataset,
+            compared_against_obj,
+            begin_version_obj,
+            end_version_obj
+        )
+    )
+}
+
+fn inner_native_build<'local>(
+    env: &mut JNIEnv<'local>,
+    java_dataset: JObject<'local>,
+    compared_against_obj: JObject<'local>,
+    begin_version_obj: JObject<'local>,
+    end_version_obj: JObject<'local>,
+) -> Result<JObject<'local>> {
+    let compared_against = env.get_u64_opt(&compared_against_obj)?;
+    let begin_version = env.get_u64_opt(&begin_version_obj)?;
+    let end_version = env.get_u64_opt(&end_version_obj)?;
+
+    let delta = {
+        let dataset_guard =
+            unsafe { env.get_rust_field::<_, _, BlockingDataset>(&java_dataset, NATIVE_DATASET)? };
+
+        let mut builder = dataset_guard.inner.delta();
+        if let Some(compared) = compared_against {
+            builder = builder.compared_against_version(compared);
+        } else if let (Some(begin), Some(end)) = (begin_version, end_version) {
+            builder = builder.with_begin_version(begin).with_end_version(end);
+        }
+        builder.build()?
+    };
+
+    let blocking_delta = BlockingDatasetDelta { inner: delta };
+    attach_native_delta(env, blocking_delta, &java_dataset)
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_delta_DatasetDelta_listTransactions<'local>(
+    mut env: JNIEnv<'local>,
+    j_delta: JObject<'local>,
+) -> JObject<'local> {
+    ok_or_throw!(env, inner_list_transactions(&mut env, j_delta))
+}
+
+fn inner_list_transactions<'local>(
+    env: &mut JNIEnv<'local>,
+    j_delta: JObject<'local>,
+) -> Result<JObject<'local>> {
+    let txs: Vec<Transaction> = {
+        let delta_guard =
+            unsafe { env.get_rust_field::<_, _, BlockingDatasetDelta>(&j_delta, NATIVE_DELTA) }?;
+        RT.block_on(delta_guard.inner.list_transactions())?
+    };
+
+    let java_dataset = env
+        .get_field(&j_delta, "dataset", "Lorg/lance/Dataset;")?
+        .l()?;
+
+    let array_list = env.new_object("java/util/ArrayList", "()V", &[])?;
+    for tx in txs.into_iter() {
+        let jtx = convert_to_java_transaction(env, tx, &java_dataset)?;
+        env.call_method(
+            &array_list,
+            "add",
+            "(Ljava/lang/Object;)Z",
+            &[JValue::Object(&jtx)],
+        )?;
+    }
+    Ok(array_list)
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_delta_DatasetDelta_getInsertedRows<'local>(
+    mut env: JNIEnv<'local>,
+    j_delta: JObject<'local>,
+    stream_addr: jlong,
+) {
+    ok_or_throw_without_return!(env, inner_get_inserted_rows(&mut env, j_delta, stream_addr))
+}
+
+fn inner_get_inserted_rows<'local>(
+    env: &mut JNIEnv,
+    j_delta: JObject<'local>,
+    stream_addr: jlong,
+) -> Result<()> {
+    let delta_guard =
+        unsafe { env.get_rust_field::<_, _, BlockingDatasetDelta>(&j_delta, NATIVE_DELTA) }?;
+
+    let stream: DatasetRecordBatchStream = RT.block_on(delta_guard.inner.get_inserted_rows())?;
+    let ffi_stream = to_ffi_arrow_array_stream(stream, RT.handle().clone())?;
+
+    unsafe { std::ptr::write_unaligned(stream_addr as *mut FFI_ArrowArrayStream, ffi_stream) }
+    Ok(())
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_delta_DatasetDelta_getUpdatedRows<'local>(
+    mut env: JNIEnv<'local>,
+    j_delta: JObject<'local>,
+    stream_addr: jlong,
+) {
+    ok_or_throw_without_return!(env, inner_get_updated_rows(&mut env, j_delta, stream_addr))
+}
+
+fn inner_get_updated_rows<'local>(
+    env: &mut JNIEnv,
+    j_delta: JObject<'local>,
+    stream_addr: jlong,
+) -> Result<()> {
+    let delta_guard =
+        unsafe { env.get_rust_field::<_, _, BlockingDatasetDelta>(&j_delta, NATIVE_DELTA) }?;
+
+    let stream: DatasetRecordBatchStream = RT.block_on(delta_guard.inner.get_updated_rows())?;
+    let ffi_stream = to_ffi_arrow_array_stream(stream, RT.handle().clone())?;
+
+    unsafe { std::ptr::write_unaligned(stream_addr as *mut FFI_ArrowArrayStream, ffi_stream) }
+    Ok(())
+}
+
+#[no_mangle]
+pub extern "system" fn Java_org_lance_delta_DatasetDelta_releaseNativeDelta(
+    mut env: JNIEnv,
+    obj: JObject,
+    handle: jlong,
+) {
+    ok_or_throw_without_return!(env, inner_release_native_delta(&mut env, obj, handle));
+}
+
+fn inner_release_native_delta(env: &mut JNIEnv, obj: JObject, _handle: jlong) -> Result<()> {
+    let _: BlockingDatasetDelta = unsafe { env.take_rust_field(obj, NATIVE_DELTA) }?;
+    Ok(())
+}

java/lance-jni/src/lib.rs

@@ -42,6 +42,7 @@ macro_rules! ok_or_throw_with_return {
 mod blocking_blob;
 mod blocking_dataset;
 mod blocking_scanner;
+mod delta;
 pub mod error;
 pub mod ffi;
 mod file_reader;

java/lance-jni/src/transaction.rs

@@ -452,7 +452,7 @@ fn inner_read_transaction<'local>(
     Ok(transaction)
 }
 
-fn convert_to_java_transaction<'local>(
+pub(crate) fn convert_to_java_transaction<'local>(
     env: &mut JNIEnv<'local>,
     transaction: Transaction,
     java_dataset: &JObject,
@@ -479,7 +479,7 @@ fn convert_to_java_transaction<'local>(
     Ok(java_transaction)
 }
 
-fn convert_to_java_operation<'local>(
+pub(crate) fn convert_to_java_operation<'local>(
     env: &mut JNIEnv<'local>,
     operation: Option<Operation>,
 ) -> Result<JObject<'local>> {
@@ -707,7 +707,7 @@ fn convert_to_java_operation_inner<'local>(
     }
 }
 
-fn convert_to_java_schema<'local>(
+pub(crate) fn convert_to_java_schema<'local>(
     env: &mut JNIEnv<'local>,
     schema: LanceSchema,
 ) -> Result<JObject<'local>> {

java/src/main/java/org/lance/Dataset.java

@@ -16,6 +16,7 @@
 import org.lance.cleanup.CleanupPolicy;
 import org.lance.cleanup.RemovalStats;
 import org.lance.compaction.CompactionOptions;
+import org.lance.delta.DatasetDelta;
 import org.lance.index.Index;
 import org.lance.index.IndexOptions;
 import org.lance.index.IndexParams;
@@ -1371,6 +1372,39 @@ public SqlQuery sql(String sql) {
     return new SqlQuery(this, sql);
   }
 
+  /**
+   * Compute the delta between current version and this version.
+   *
+   * @param comparedAgainst the version to compare the current dataset against
+   * @return a DatasetDelta view
+   * @throws IllegalArgumentException if mutual exclusivity or completeness rules are violated
+   */
+  public DatasetDelta delta(long comparedAgainst) {
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed");
+      return nativeBuildDelta(Optional.of(comparedAgainst), Optional.empty(), Optional.empty());
+    }
+  }
+
+  /**
+   * Compute the delta between both {@code beginVersion} (exclusive) and {@code endVersion}
+   * (inclusive).
+   *
+   * @param beginVersion the beginning version (exclusive) for explicit range
+   * @param endVersion the ending version (inclusive) for explicit range
+   * @return a DatasetDelta view
+   * @throws IllegalArgumentException if mutual exclusivity or completeness rules are violated
+   */
+  public DatasetDelta delta(long beginVersion, long endVersion) {
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed");
+      return nativeBuildDelta(Optional.empty(), Optional.of(beginVersion), Optional.of(endVersion));
+    }
+  }
+
+  private native DatasetDelta nativeBuildDelta(
+      Optional<Long> comparedAgainst, Optional<Long> beginVersion, Optional<Long> endVersion);
+
   /**
    * Merge source data with the existing target data.
    *

java/src/main/java/org/lance/delta/DatasetDelta.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.lance.delta;
+
+import org.lance.Dataset;
+import org.lance.JniLoader;
+import org.lance.LockManager;
+import org.lance.Transaction;
+
+import org.apache.arrow.c.ArrowArrayStream;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * A view of differences between two versions of a dataset.
+ *
+ * <p>Created by {@link DatasetDeltaBuilder}. Provides methods to list transactions and stream
+ * inserted/updated rows between two versions.
+ */
+public class DatasetDelta implements Closeable {
+  static {
+    JniLoader.ensureLoaded();
+  }
+
+  /** Native handle to the Rust DatasetDelta. */
+  private long nativeDeltaHandle;
+
+  /** Base dataset used to compute the delta. Also used for Transaction conversion. */
+  private Dataset dataset;
+
+  private final LockManager lockManager = new LockManager();
+
+  private DatasetDelta() {}
+
+  /**
+   * List transactions between begin_version + 1 and end_version (inclusive).
+   *
+   * @return list of transactions
+   */
+  public List<Transaction> listTransactions() {
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeDeltaHandle != 0, "DatasetDelta is closed");
+      return nativeListTransactions();
+    }
+  }
+
+  private native List<Transaction> nativeListTransactions();
+
+  /** Return a streaming ArrowReader for inserted rows. */
+  public ArrowReader getInsertedRows() throws IOException {
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeDeltaHandle != 0, "DatasetDelta is closed");
+      BufferAllocator allocator = dataset.allocator();
+      try (ArrowArrayStream s = ArrowArrayStream.allocateNew(allocator)) {
+        nativeGetInsertedRows(s.memoryAddress());
+        return Data.importArrayStream(allocator, s);
+      }
+    }
+  }
+
+  private native void nativeGetInsertedRows(long streamAddress) throws IOException;
+
+  /** Return a streaming ArrowReader for updated rows. */
+  public ArrowReader getUpdatedRows() throws IOException {
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeDeltaHandle != 0, "DatasetDelta is closed");
+      BufferAllocator allocator = dataset.allocator();
+      try (ArrowArrayStream s = ArrowArrayStream.allocateNew(allocator)) {
+        nativeGetUpdatedRows(s.memoryAddress());
+        return Data.importArrayStream(allocator, s);
+      }
+    }
+  }
+
+  private native void nativeGetUpdatedRows(long streamAddress) throws IOException;
+
+  @Override
+  public void close() {
+    try (LockManager.WriteLock writeLock = lockManager.acquireWriteLock()) {
+      if (nativeDeltaHandle != 0) {
+        releaseNativeDelta(nativeDeltaHandle);
+        nativeDeltaHandle = 0;
+      }
+    }
+  }
+
+  private native void releaseNativeDelta(long handle);
+}