feat(java): expose MemWAL shard delete (#7688)

## What

Mirrors the Python `ShardWriter.delete` binding on the Java side,
bringing the Java MemWAL bindings to parity with Python. The Rust core
`ShardWriter::delete` and the Python binding already exist; this wires
up the Java surface.

- **`java/lance-jni/src/mem_wal.rs`** — `nativeDelete` JNI entry point
(`inner_delete`) that streams Arrow key batches into
`ShardWriter::delete`, mirroring `nativePut`/`inner_put`.
- **`ShardWriter.java`** — public `delete(ArrowReader)` wrapper
mirroring `put`, with Javadoc copied from the Rust core (tombstone
semantics, nullable-column and primary-key constraints).
- **`MemWalTest.java`** — integration test
`testShardWriterDeleteMasksBaseRow` asserting a tombstone masks the
deleted base row, mirroring the Python
`test_shard_writer_delete_binding_masks_base_row`.

All validation and tombstone construction stay centralized in the Rust
core — the binding is a thin wrapper, per the cross-language binding
guidelines.

## Testing

- `cargo clippy --tests --manifest-path ./lance-jni/Cargo.toml` — clean
- `./mvnw spotless:check` + `cargo fmt` — clean
- `./mvnw test -Dtest=MemWalTest#testShardWriterDeleteMasksBaseRow` —
`Tests run: 1, Failures: 0`

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **New Features**
  * Added support for deleting records through the shard writer API.
* Deletions now work using primary-key-only input and are reflected as
hidden tombstone rows.
* **Bug Fixes**
* Deleted rows are now masked from query results while preserving other
existing records and newly added data.
* Added validation and safer handling for empty delete inputs and
invalid usage.
* **Tests**
* Added an integration test covering delete behavior and tombstone
masking.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: hamersaw

java/lance-jni/src/mem_wal.rs

@@ -181,6 +181,29 @@ fn inner_put(env: &mut JNIEnv, this: JObject, stream_addr: jlong) -> Result<()>
     Ok(())
 }
 
+#[unsafe(no_mangle)]
+pub extern "system" fn Java_org_lance_memwal_ShardWriter_nativeDelete(
+    mut env: JNIEnv,
+    this: JObject,
+    stream_addr: jlong,
+) {
+    ok_or_throw_without_return!(env, inner_delete(&mut env, this, stream_addr));
+}
+
+fn inner_delete(env: &mut JNIEnv, this: JObject, stream_addr: jlong) -> Result<()> {
+    let stream_ptr = stream_addr as *mut FFI_ArrowArrayStream;
+    let reader = unsafe { ArrowArrayStreamReader::from_raw(stream_ptr) }?;
+    let batches: Vec<RecordBatch> = reader.collect::<std::result::Result<_, _>>()?;
+    if batches.is_empty() {
+        return Ok(());
+    }
+
+    let guard =
+        unsafe { env.get_rust_field::<_, _, BlockingShardWriter>(&this, NATIVE_SHARD_WRITER) }?;
+    RT.block_on(guard.writer.delete(batches))?;
+    Ok(())
+}
+
 /// Test-support: write a primary-key dedup sidecar (`_pk_index/`) for a
 /// flushed-generation dataset already staged at `gen_path`, mirroring what
 /// production flush emits. Lets Java tests stage a *faithful* flushed

java/src/main/java/org/lance/memwal/ShardWriter.java

@@ -36,6 +36,7 @@
  * <pre>{@code
  * try (ShardWriter writer = dataset.memWalWriter(shardId)) {
  *   writer.put(reader);
+ *   writer.delete(keys);
  * }
  * }</pre>
  *
@@ -99,6 +100,34 @@ public void put(ArrowReader reader) {
 
   private native void nativePut(long streamAddress);
 
+  /**
+   * Delete rows from the MemWAL by primary key.
+   *
+   * <p>Each batch in {@code reader} must carry this shard's primary key column(s); other columns
+   * are ignored. Lance builds a tombstone row per key — the primary key plus {@code _tombstone =
+   * true} and null in every other column — and appends it like an ordinary write. The tombstone is
+   * the newest value for its key: it wins newest-per-PK resolution (suppressing the older real row)
+   * and is then dropped from query results.
+   *
+   * <p>Only supported in memtable mode. Because a tombstone nulls every non-PK column, those
+   * columns must be nullable in the base schema; deleting against a schema with a non-nullable
+   * non-PK column errors. Deleting on a shard with no primary key columns also errors.
+   *
+   * @param reader the keys to delete; consumed fully by this call
+   */
+  public void delete(ArrowReader reader) {
+    Preconditions.checkNotNull(reader, "reader must not be null");
+    try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {
+      Preconditions.checkArgument(nativeShardWriterHandle != 0, "ShardWriter is closed");
+      try (ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) {
+        Data.exportArrayStream(allocator, reader, stream);
+        nativeDelete(stream.memoryAddress());
+      }
+    }
+  }
+
+  private native void nativeDelete(long streamAddress);
+
   /** Return a snapshot of cumulative write statistics. */
   public WriteStats stats() {
     try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) {

java/src/test/java/org/lance/memwal/MemWalTest.java

@@ -97,6 +97,22 @@ private static VectorSchemaRoot lookupRoot(BufferAllocator allocator, long[] ids
     return root;
   }
 
+  /** Build a single-batch root carrying only the {@code id} primary key, for deletes. */
+  private static VectorSchemaRoot keysRoot(BufferAllocator allocator, long[] ids) {
+    VectorSchemaRoot root =
+        VectorSchemaRoot.create(
+            new Schema(
+                Collections.singletonList(Field.nullable("id", new ArrowType.Int(64, true)))),
+            allocator);
+    BigIntVector idVector = (BigIntVector) root.getVector("id");
+    idVector.allocateNew(ids.length);
+    for (int i = 0; i < ids.length; i++) {
+      idVector.set(i, ids[i]);
+    }
+    root.setRowCount(ids.length);
+    return root;
+  }
+
   /** Build a single-batch append-only root without primary-key metadata. */
   private static VectorSchemaRoot appendOnlyRoot(
       BufferAllocator allocator, long[] ids, String prefix) {
@@ -382,6 +398,52 @@ void testShardWriterPutAndLsmScanner(@TempDir Path tempDir) throws Exception {
     }
   }
 
+  @Test
+  void testShardWriterDeleteMasksBaseRow(@TempDir Path tempDir) throws Exception {
+    String path = tempDir.resolve("base").toString();
+    String shardId = UUID.randomUUID().toString();
+    try (BufferAllocator allocator = new RootAllocator();
+        Dataset dataset = writeLookupDataset(allocator, path, new long[] {1, 2, 3}, "base")) {
+      dataset.initializeMemWal(new InitializeMemWalParams());
+
+      ShardWriterConfig config =
+          new ShardWriterConfig()
+              .withDurableWrite(true)
+              .withSyncIndexedWrite(true)
+              .withMaxWalBufferSize(1)
+              .withMaxWalFlushIntervalMs(10);
+
+      try (ShardWriter writer = dataset.memWalWriter(shardId, config)) {
+        try (VectorSchemaRoot root = lookupRoot(allocator, new long[] {4}, "writer");
+            ArrowReader reader = toReader(allocator, root)) {
+          writer.put(reader);
+        }
+        try (VectorSchemaRoot keys = keysRoot(allocator, new long[] {2});
+            ArrowReader reader = toReader(allocator, keys)) {
+          writer.delete(reader);
+        }
+
+        Map<Long, String> byId = Collections.emptyMap();
+        long deadline = System.currentTimeMillis() + 10_000;
+        while (System.currentTimeMillis() < deadline) {
+          try (LsmScanner scanner = writer.lsmScanner();
+              ArrowReader reader = scanner.scanBatches()) {
+            byId = readByName(reader);
+          }
+          if (!byId.containsKey(2L) && "writer_4".equals(byId.get(4L))) {
+            break;
+          }
+          Thread.sleep(50);
+        }
+
+        assertEquals("base_1", byId.get(1L));
+        assertFalse(byId.containsKey(2L), "deleted base row should be masked by the tombstone");
+        assertEquals("base_3", byId.get(3L));
+        assertEquals("writer_4", byId.get(4L));
+      }
+    }
+  }
+
   @Test
   void testLsmScannerFromSnapshots(@TempDir Path tempDir) throws Exception {
     String basePath = tempDir.resolve("base").toString();