feat: report the row ids deleted between two versions

java/lance-jni/src/delta.rs

@@ -187,6 +187,33 @@ fn inner_get_updated_rows<'local>(
     Ok(())
 }
 
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_lance_delta_DatasetDelta_nativeGetDeletedRowIds<'local>(
+    mut env: JNIEnv<'local>,
+    j_delta: JObject<'local>,
+    stream_addr: jlong,
+) {
+    ok_or_throw_without_return!(
+        env,
+        inner_get_deleted_row_ids(&mut env, j_delta, stream_addr)
+    )
+}
+
+fn inner_get_deleted_row_ids<'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 = block_on(delta_guard.inner.get_deleted_row_ids())?;
+    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(())
+}
+
 #[unsafe(no_mangle)]
 pub extern "system" fn Java_org_lance_delta_DatasetDelta_releaseNativeDelta(
     mut env: JNIEnv,

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

@@ -91,6 +91,24 @@ public ArrowReader getUpdatedRows() throws IOException {
 
   private native void nativeGetUpdatedRows(long streamAddress) throws IOException;
 
+  /**
+   * Return a streaming ArrowReader of the row ids deleted in the range.
+   *
+   * <p>The batches carry a single {@code _rowid} column. Requires stable row ids.
+   */
+  public ArrowReader getDeletedRowIds() 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)) {
+        nativeGetDeletedRowIds(s.memoryAddress());
+        return Data.importArrayStream(allocator, s);
+      }
+    }
+  }
+
+  private native void nativeGetDeletedRowIds(long streamAddress) throws IOException;
+
   @Override
   public void close() {
     try (LockManager.WriteLock writeLock = lockManager.acquireWriteLock()) {

java/src/test/java/org/lance/DeltaTest.java

@@ -188,6 +188,57 @@ public void testListTransactionsExplicitRange(@TempDir Path tempDir) throws IOEx
     }
   }
 
+  @Test
+  public void testGetDeletedRowIds(@TempDir Path tempDir) throws IOException {
+    try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+      String uri = tempDir.resolve("delta_demo_delete").toString();
+      Schema schema =
+          new Schema(
+              Arrays.asList(
+                  Field.notNullable(
+                      "id", new org.apache.arrow.vector.types.pojo.ArrowType.Int(32, true)),
+                  Field.nullable(
+                      "val", org.apache.arrow.vector.types.pojo.ArrowType.Utf8.INSTANCE)));
+
+      // v1: create with three rows, keeping stable row ids so deletions are reportable.
+      byte[] batch1 =
+          writeBatch(allocator, schema, new int[] {1, 2, 3}, new String[] {"a", "b", "c"});
+      try (ArrowStreamReader reader1 =
+              new ArrowStreamReader(new ByteArrayReadableSeekableByteChannel(batch1), allocator);
+          ArrowArrayStream stream1 = ArrowArrayStream.allocateNew(allocator)) {
+        Data.exportArrayStream(allocator, reader1, stream1);
+        Dataset.write().stream(stream1)
+            .uri(uri)
+            .mode(WriteParams.WriteMode.CREATE)
+            .enableStableRowIds(true)
+            .execute()
+            .close();
+      }
+
+      // v2: delete one row.
+      try (Dataset ds = Dataset.open(uri, allocator)) {
+        ds.delete("id = 2");
+      }
+
+      try (Dataset ds2 = Dataset.open(uri, allocator)) {
+        DatasetDelta delta = ds2.delta(1L);
+        try (ArrowReader deleted = delta.getDeletedRowIds()) {
+          int total = 0;
+          while (deleted.loadNextBatch()) {
+            VectorSchemaRoot outRoot = deleted.getVectorSchemaRoot();
+            List<String> names =
+                outRoot.getSchema().getFields().stream()
+                    .map(Field::getName)
+                    .collect(Collectors.toList());
+            Assertions.assertEquals(Arrays.asList("_rowid"), names);
+            total += outRoot.getRowCount();
+          }
+          Assertions.assertEquals(1, total, "exactly one row was deleted");
+        }
+      }
+    }
+  }
+
   /** Helper: serialize a single Arrow batch with the given schema and (id, val) pairs. */
   private static byte[] writeBatch(RootAllocator allocator, Schema schema, int[] ids, String[] vals)
       throws IOException {

python/python/lance/dataset.py

@@ -5670,6 +5670,14 @@ def get_updated_rows(self) -> pa.RecordBatchReader:
         """
         return self._delta.get_updated_rows()
 
+    def get_deleted_row_ids(self) -> pa.RecordBatchReader:
+        """
+        Return a streaming RecordBatchReader of the row ids deleted in the range.
+
+        The batches carry a single ``_rowid`` column. Requires stable row ids.
+        """
+        return self._delta.get_deleted_row_ids()
+
 
 class _DatasetDeltaBuilder:
     """Internal builder for :class:`DatasetDelta`.

python/python/tests/test_delta.py

@@ -134,3 +134,28 @@ def test_delta_validation_errors():
         "and with_end_version",
     ):
         ds.delta(end_version=2)
+
+
+def test_delta_get_deleted_row_ids():
+    table = pa.table(
+        {
+            "id": pa.array([1, 2, 3, 4], type=pa.int32()),
+            "val": pa.array(["a", "b", "c", "d"], type=pa.string()),
+        }
+    )
+    ds = write_dataset(
+        table, "memory://delta_api_test_delete", enable_stable_row_ids=True
+    )
+    row_ids = ds.to_table(columns=[], with_row_id=True).column("_rowid").to_pylist()
+
+    ds.delete("id in (2, 3)")
+
+    delta = ds.delta(compared_against=1)
+    reader = delta.get_deleted_row_ids()
+
+    deleted = []
+    for batch in reader:
+        assert batch.schema.names == ["_rowid"]
+        deleted.extend(batch.column("_rowid").to_pylist())
+
+    assert sorted(deleted) == sorted([row_ids[1], row_ids[2]])

python/src/dataset.rs

@@ -4154,6 +4154,18 @@ impl DatasetDelta {
         let reader: Box<dyn RecordBatchReader + Send> = Box::new(LanceReader::from_stream(stream));
         reader.into_pyarrow(py)
     }
+    /// Get the row ids deleted between begin_version (exclusive) and end_version (inclusive) as a stream reader.
+    ///
+    /// Requires stable row ids on the dataset.
+    fn get_deleted_row_ids<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        use arrow::pyarrow::IntoPyArrow;
+        use arrow_array::RecordBatchReader;
+        let stream = rt()
+            .block_on(None, self.inner.get_deleted_row_ids())?
+            .infer_error()?;
+        let reader: Box<dyn RecordBatchReader + Send> = Box::new(LanceReader::from_stream(stream));
+        reader.into_pyarrow(py)
+    }
 }
 
 #[pyclass(

rust/lance-table/src/rowids.rs

@@ -357,6 +357,11 @@ impl RowIdSequence {
     /// Get the row id at the given index.
     ///
     /// If the index is out of bounds, this will return None.
+    /// The segments backing the sequence, in offset order.
+    pub fn segments(&self) -> &[U64Segment] {
+        &self.0
+    }
+
     pub fn get(&self, index: usize) -> Option<u64> {
         let mut offset = 0;
         for segment in &self.0 {