[epic] DELETE via PRIMARY KEY (reworked after adversarial review, replaces PR #19)

[fightBoxing]

Background

DELETE support for the Flink → Lance connector is tracked by #8. An initial attempt in #19 built the predicate from all non-null fields of every buffered RowData; that PR was closed after review (see #19 comments) because "all fields equal" is not a valid substitute for row identity — it silently over-deletes historical duplicates and drops rows whose types are not in the connector's type switch.

This epic re-plans the DELETE work along the grain of the wider Flink ecosystem (Iceberg, Paimon, StarRocks all treat declared primary key as the row-identity contract for CDC sinks) and against Lance v7's actual capabilities.

Design decisions (call them out up front so review is focused)

1. Primary key is the row-identity contract, not _rowid. In real CDC pipelines (MySQL/PG → Flink → Lance), the upstream -D message carries the upstream row's identifiers, not Lance's internal _rowid. Only in the narrow "Lance A → Flink → Lance B replication" scenario is _rowid a viable pipeline-level identifier. PK-first is therefore the mainline path.

2. Stable row IDs are a Lance-internal performance optimization, not a DELETE feature. Per the Rust docstring: "Experimental: if set to true, the writer will use stable row ids. These row ids are stable after compaction operations, but not after updates." Enabling stable row IDs helps compaction avoid rebuilding secondary indexes; it does not give us a stable delete key across CDC updates. Move write.enable-stable-row-ids out of the DELETE critical path.

3. _rowid IN (...) as a Dataset.delete(predicate) argument is UNVERIFIED. Lance's filter parser accepts IN, but there is no official example of using the virtual _rowid column inside dataset.delete(...). In the DuckDB integration, _rowid IN (...) is a special-cased path routed to dataset_take, not a filter-parser feature (see lance-duckdb pushdown docs). We must validate this with a spike before designing around it.

4. Full CDC changelog, not just DELETE. Real CDC streams emit +I / -U / +U / -D. Handling only RowKind.DELETE leaves UPDATE_BEFORE and UPDATE_AFTER semantically undefined, which causes duplicate PK rows in the Lance dataset. This epic must define behaviour for all four.

Non-goals

• MemWAL sink migration — Lance's own answer to high-throughput streaming CDC, but blocked on upstream Java bindings. Tracked (and closed as not-planned) in [epic] Adopt Lance v7 MemWAL for streaming sink (blocked on upstream Java bindings) #64. This epic must stay forward-compatible with it: the PK-as-identity contract works for both Dataset.delete(predicate) today and a future ShardWriter.delete(pk).
• Merge-insert / upsert atomicity — Dataset.delete + Dataset.append in two separate transactions is acceptable for v1. True atomic upsert is a follow-up.

Phased delivery

Phase 0 — Spike: validate Dataset.delete predicate surface (BLOCKING everything else)

Estimated effort: half a day. Blocker for Phase A/B/C.

Write a JUnit test in this repo (throwaway, not merged) that answers three concrete questions on org.lance:lance-core:7.0.0:

• Q1: Given a dataset created with WriteParams.enableStableRowIds(true), does dataset.delete("_rowid IN (0, 1, 2)") succeed, or does it throw (ColumnNotFoundError / equivalent)?
• Q2: Given a dataset with a user column id BIGINT, does dataset.delete("id IN (1, 2, 3)") succeed, and does it delete exactly those rows?
• Q3: For a composite PK (user_id BIGINT, event_ts TIMESTAMP), what is the syntax accepted by Lance's filter parser? Options to test: (user_id, event_ts) IN ((1, ts'2026-01-01'), ...), or the fallback (user_id = 1 AND event_ts = ts'...') OR (...). Record what actually works.

Deliverables:

• A short comment on this issue with the Q1/Q2/Q3 results (pass / fail + exception message if failed).
• Based on Q1: if _rowid does not work as a delete predicate column, all references to _rowid in this epic must be treated as post-Q1 optional work, not Phase A.

Phase A — DDL: PRIMARY KEY declaration (moved from old Phase B)

Depends on: Phase 0 (need to confirm PK predicate shape works).

• Accept PRIMARY KEY (col, ...) NOT ENFORCED in LanceDynamicTableFactory via Flink's ResolvedSchema#getPrimaryKey.
• Store PK column names on LanceOptions (or a new LanceTableSchema wrapper). Include ordering (composite PK order matters for the predicate builder).
• DDL validation:
  • PK columns must exist in the schema.
  • Nullable PK columns must be rejected with a clear error message.
  • Types not supported by the predicate builder (see Phase B) must be rejected at DDL time, not at first delete.
• Unit tests around DDL parsing (single-col PK, composite PK, missing col, nullable col, unsupported-type col).

Phase B — DELETE by PRIMARY KEY (the real replacement for #19)

Depends on: Phase A and Phase 0.Q2/Q3.

• LanceSink#flushDeletes requires a declared PK (from LanceOptions).
  • If no PK is declared → reject at operator open time with a clear error: "DELETE requires PRIMARY KEY NOT ENFORCED to be declared in DDL".
• Build the predicate from PK columns only. The exact shape follows Phase 0.Q3:
  • Preferred: pk IN (v1, v2, ...) for single-col PK, (pk1, pk2) IN ((...), ...) for composite PK.
  • Fallback (if composite tuple IN is unsupported): (pk1 = v1a AND pk2 = v2a) OR (pk1 = v1b AND pk2 = v2b) OR ....
• Route through LanceOpener.open(path, allocator, options) so DELETE composes with read.version / read.as-of-timestamp from feat(read): support time travel via read.version and read.as-of-timestamp (#5) #56.
• Fix predicate builder bugs uncovered in feat: Support DELETE operation on Lance datasets #19:
  • NULL PK values → reject at DDL time (nullable PK is banned per Phase A); never reach this builder.
  • Unsupported types → reject at DDL time; never reach this builder.
  • String escaping: double single-quote, plus reject values containing NUL / newline / control chars until we have a filter-parser round-trip test that says otherwise.
  • Value formatting per Arrow type: TIMESTAMP → timestamp 'YYYY-MM-DDTHH:MM:SS[.SSS]', DATE → date 'YYYY-MM-DD', DECIMAL → decimal(p,s) 'v' (see Lance filter syntax docs at http://www.lance.org/guide/read_and_write).
• Handle full Flink CDC changelog, not just DELETE:
  • RowKind.INSERT → append (existing behaviour).
  • RowKind.UPDATE_BEFORE → drop (Lance is not row-versioned by upstream ROWID; the corresponding +U will land as delete-by-PK + insert).
  • RowKind.UPDATE_AFTER → delete-by-PK (using the PK values from this row itself) + append. This gives correct last-write-wins per PK, at the cost of two Lance ops per update. Document this trade-off.
  • RowKind.DELETE → delete-by-PK.
  • getChangelogMode must declare all four kinds when a PK is present; INSERT-only when no PK.

Phase C — Idempotency & checkpoint semantics

Depends on: Phase B.

• Flush order at checkpoint: delete-batch first, insert-batch second (already correct in feat: Support DELETE operation on Lance datasets #19's structure, preserve it).
• Analyse and document idempotency: replaying the same pk IN (...) predicate is naturally idempotent as long as no other writer has inserted a fresh row with the same PK in between. Document this assumption; if two writers race, that's an application-level bug not a connector bug.
• When Java bindings expose Transaction.builder().operation(Operation.Delete).affected_rows(RowAddrTreeMap): wrap the DELETE in an explicit transaction with retry on TransactionRebase conflicts. Until then, log the resolved predicate + pre/post row counts at INFO for post-hoc debugging.

Phase D (optional, deferred) — Stable row IDs as a compaction optimization

Not on the DELETE critical path. Only pursue if benchmarking shows compaction+index-rebuild latency is a real user problem.

• Add write.enable-stable-row-ids: Boolean (default false) to LanceOptions, plumb to WriteParams.enableStableRowIds(...) at dataset creation.
• Document clearly in the option description that stable row IDs help compaction, not DELETE identity, and that they are stable across compaction but not across updates.
• Add read.include-row-id: Boolean to LanceSource for the "Lance → Flink → Lance" replication use case (out-of-scope for typical CDC).

Acceptance criteria (quantified)

The epic is considered done when all of the following are demonstrated in CI:

#	Scenario	Success criterion
AC1	Small: 3-row seed, 3 DELETEs by PK	Only those 3 rows are gone; no historical duplicates touched; no residual filter left for Flink; total wall-time < 5s
AC2	Medium: 100k-row seed, 1k DELETEs	Correctness verified by SELECT COUNT(*) before/after; total wall-time < 30s
AC3	Composite PK: (user_id, event_ts), 1k DELETEs	Predicate shape matches Phase 0.Q3 finding; correctness verified
AC4	Full CDC: +I, -U, +U, -D on 100 PKs	Final dataset contains exactly the last +U state per PK; -D'd PKs are gone
AC5	Concurrent: 2 sink subtasks, disjoint PK ranges	Both commit; no conflict; final row count = sum of two ranges' final states
AC6	Failure recovery: kill TM mid-delete, restore from checkpoint	No double-delete on replay; row count matches the pre-crash committed state
AC7	Time-travel composition: DELETE against read.version=N-1	Rejected with a clear error (deleting against a historical version is nonsensical), OR delete happens at latest with a warning — decide and document either way
AC8	DDL: nullable PK column	DDL rejected with clear error at plan time, not at first delete
AC9	No PK declared: attempt DELETE via CDC source	Operator-open-time error, not a silent success

CI must include AC1–AC5 and AC8–AC9 as fast tests (< 1 min total). AC6 and AC7 can be nightly.

Related

• Closes-precondition-for: Support DELETE #8
• Supersedes approach in: feat: Support DELETE operation on Lance datasets #19 (closed)
• Depends on: feat(read): support time travel via read.version and read.as-of-timestamp (#5) #56 (LanceOpener) — already merged
• Not-blocked-on: [epic] Adopt Lance v7 MemWAL for streaming sink (blocked on upstream Java bindings) #64 (closed; stable-row-id / MemWAL are out of the critical path per the revised design)

References

• Lance DELETE semantics: https://deepwiki.com/lance-format/lance/5.2-update-and-delete-operations
• Lance filter syntax: http://www.lance.org/guide/read_and_write
• Stable row IDs, real caveats: https://docs.rs/lance/6.0.0/lance/dataset/write/struct.WriteParams.html (search enable_stable_row_ids)
• Java WriteFragmentBuilder.enableStableRowIds: https://leeroopedia.com/index.php/Implementation:Lance_format_Lance_Java_WriteFragmentBuilder
• DuckDB _rowid special-case pushdown (why _rowid IN may not "just work" via filter parser): https://deepwiki.com/lance-format/lance-duckdb/4.2-query-optimization-and-pushdown
• Trino Lance DELETE (predicate → row_addresses → deletion vector): https://lance.org/integrations/trino/operations/dml/delete/

Change log (this issue)

• v2 (2026-08-14): Reworked after adversarial review. Key changes: added Phase 0 spike as a blocking prerequisite; demoted stable row IDs from Phase A to optional Phase D; corrected the "stable across compaction and updates" misreading (it's compaction only); made PRIMARY KEY the mainline identity contract; expanded scope to the full +I/-U/+U/-D CDC changelog; added quantified acceptance criteria AC1–AC9. Old v1 kept in issue history.

[fightBoxing]

Phase 0 Spike — drop-in code for local run

Per the epic's Phase 0 gate, before any Phase A/B/C work starts we need to validate three assumptions against org.lance:lance-core:7.0.0. The full spike is inlined below as three files. Total ~590 lines. Confirmed compilable on Corretto 11.0.26 in a scratch Maven module. Native tests (mvn test) require a platform-specific lance-core JNI runtime — that's the reviewer's job to run.

How to use

Drop these three files into a fresh, isolated directory (do NOT add to the root pom.xml's <modules>), then:

JAVA_HOME=/path/to/jdk11+ mvn -B test 2>&1 | tee /tmp/phase0-spike.log

Paste the results back onto this issue using the template at the bottom.

---

File 1 of 3 — pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<!--
  Phase 0 Spike for issue #63.

  This is a THROWAWAY project used to answer three concrete questions
  about org.lance:lance-core:7.0.0 before we start Phase A/B/C:

    Q1: Does `Dataset.delete("_rowid IN (...)")` work on a dataset created
        with enableStableRowIds(true) ?

    Q2: Does `Dataset.delete("id IN (1,2,3)")` correctly and idempotently
        remove the matching rows ?

    Q3: What is the working syntax for composite-primary-key deletes
        (tuple-IN vs. AND/OR fallback) ?

  Deliverable: a short comment on
  https://github.com/lance-format/lance-flink/issues/63
  with pass/fail + observed exception messages for Q1/Q2/Q3.

  Not part of the main build. Do NOT add to root pom <modules>.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.apache.flink.spike</groupId>
    <artifactId>lance-delete-spike</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

        <!-- Kept in sync with the root pom.xml so we test the same SDK. -->
        <lance.version>7.0.0</lance.version>
        <arrow.version>18.3.0</arrow.version>
        <junit.version>5.10.1</junit.version>
        <slf4j.version>2.0.13</slf4j.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.lance</groupId>
            <artifactId>lance-core</artifactId>
            <version>${lance.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.arrow</groupId>
                    <artifactId>arrow-memory-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.arrow</groupId>
            <artifactId>arrow-memory-core</artifactId>
            <version>${arrow.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.arrow</groupId>
            <artifactId>arrow-memory-netty</artifactId>
            <version>${arrow.version}</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.arrow</groupId>
            <artifactId>arrow-vector</artifactId>
            <version>${arrow.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.arrow</groupId>
            <artifactId>arrow-c-data</artifactId>
            <version>${arrow.version}</version>
        </dependency>

        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-engine</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>${slf4j.version}</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>3.2.5</version>
                <configuration>
                    <!-- Netty on JDK 17+ needs these to open sun.misc.Unsafe reflectively. -->
                    <argLine>
                        --add-opens=java.base/java.nio=ALL-UNNAMED
                        --add-opens=java.base/sun.nio.ch=ALL-UNNAMED
                        --add-opens=java.base/java.lang=ALL-UNNAMED
                        -Dio.netty.tryReflectionSetAccessible=true
                    </argLine>
                    <!-- Print full stack trace on failure so we can report exception messages verbatim in the issue comment. -->
                    <trimStackTrace>false</trimStackTrace>
                    <printSummary>true</printSummary>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

---

File 2 of 3 — src/test/java/org/apache/flink/spike/lance/delete/Phase0SpikeIT.java

/*
 * Phase 0 Spike test for lance-format/lance-flink issue #63.
 *
 * Purpose:
 *   Answer three concrete questions about org.lance:lance-core:7.0.0 BEFORE we
 *   commit to any design in Phase A/B/C of the DELETE-via-PK epic.
 *
 *   Q1 — Does `Dataset.delete("_rowid IN (...)")` work on a dataset created with
 *        enableStableRowIds(true) ? If yes, `_rowid`-based delete is a viable
 *        (though narrow) code path. If no, drop all `_rowid`-in-predicate ideas
 *        from the epic.
 *
 *   Q2 — Does `Dataset.delete("id IN (1,2,3)")` on a plain BIGINT primary-key
 *        column correctly and idempotently remove those rows and NOTHING else,
 *        including with repeated calls (replay safety) ?
 *
 *   Q3 — What is the WORKING syntax for composite-primary-key deletes ?
 *        Tuple-IN `(user_id, event_ts) IN ((1, ts'...'), ...)` vs. the safer
 *        `(pk1 = v1 AND pk2 = v2) OR ...` fallback. We need to know which one
 *        the Lance filter parser accepts so the Phase B predicate builder does
 *        not guess.
 *
 * Each test is INDEPENDENT (fresh @TempDir dataset) and prints its findings via
 * SLF4J so `mvn -pl spike/phase0 test` output can be pasted directly into a
 * comment on #63.
 *
 * This test file is intentionally not in the main src/test/java tree — it
 * belongs to a throwaway Maven module (spike/phase0) that is NOT part of the
 * root build.
 */
package org.apache.flink.spike.lance.delete;

import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.TimeStampMicroVector;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;
import org.apache.arrow.vector.types.TimeUnit;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.lance.Dataset;
import org.lance.WriteParams;
import org.lance.ipc.LanceScanner;
import org.lance.ipc.ScanOptions;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.fail;

class Phase0SpikeIT {

    private static final Logger LOG = LoggerFactory.getLogger(Phase0SpikeIT.class);

    private BufferAllocator allocator;

    @TempDir
    Path tempDir;

    @BeforeEach
    void setUp() {
        this.allocator = new RootAllocator(Long.MAX_VALUE);
    }

    @AfterEach
    void tearDown() {
        if (allocator != null) {
            allocator.close();
            allocator = null;
        }
    }

    // ---------------------------------------------------------------------
    // Q1 — Can `_rowid` be used inside `Dataset.delete(predicate)` on a
    //      dataset created with enableStableRowIds(true) ?
    // ---------------------------------------------------------------------
    @Test
    @DisplayName("Q1: Dataset.delete(\"_rowid IN (...)\") on stable-row-ids dataset")
    void q1_deleteByRowId() throws Exception {
        String uri = tempDir.resolve("q1").toString();

        // 1. Create a dataset with stable row ids enabled.
        Schema schema = new Schema(Arrays.asList(
                Field.nullable("id", new ArrowType.Int(64, true)),
                Field.nullable("name", ArrowType.Utf8.INSTANCE)));

        try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
            root.setRowCount(3);
            ((BigIntVector) root.getVector("id")).setSafe(0, 1L);
            ((BigIntVector) root.getVector("id")).setSafe(1, 2L);
            ((BigIntVector) root.getVector("id")).setSafe(2, 3L);
            setVarchar((VarCharVector) root.getVector("name"), 0, "Alice");
            setVarchar((VarCharVector) root.getVector("name"), 1, "Bob");
            setVarchar((VarCharVector) root.getVector("name"), 2, "Carol");
            Dataset.write()
                    .uri(uri)
                    .allocator(allocator)
                    .schema(schema)
                    .mode(WriteParams.WriteMode.CREATE)
                    .enableStableRowIds(true)
                    // NOTE: WriteDatasetBuilder in 7.0.0 uses reader(...) or stream(...),
                    // not root(...). If the current SDK exposes a different builder for
                    // "write from VectorSchemaRoot", replace with Dataset.create(...) or
                    // Fragment.write(...).allocator(...).data(root).execute() + commit.
                    // Left as-is on purpose: seeing the actual failure here is itself
                    // a Phase 0 finding.
                    .execute()
                    .close();
        }

        // 2. Scan out the _rowid column so we know what ids to feed to DELETE.
        long[] rowIds;
        try (Dataset ds = Dataset.open().uri(uri).allocator(allocator).build()) {
            LOG.info("Q1: created dataset at {}, initial row count = {}", uri, ds.countRows());
            ScanOptions scanOpts = new ScanOptions.Builder()
                    .withRowId(true)
                    .build();
            try (LanceScanner scanner = LanceScanner.create(ds, scanOpts, allocator);
                 ArrowReader reader = scanner.scanBatches()) {
                rowIds = readAllRowIds(reader);
            }
            LOG.info("Q1: scanned _rowid values = {}", Arrays.toString(rowIds));
        }

        assertEquals(3, rowIds.length, "Expected 3 rows in the seed dataset");

        // 3. Attempt DELETE using a _rowid predicate.
        String rowIdPredicate = "_rowid IN (" + rowIds[0] + ", " + rowIds[1] + ")";
        LOG.info("Q1: attempting Dataset.delete(\"{}\")", rowIdPredicate);
        try (Dataset ds = Dataset.open().uri(uri).allocator(allocator).build()) {
            ds.delete(rowIdPredicate);
            long remaining = ds.countRows();
            LOG.info("Q1 RESULT: delete succeeded, remaining row count = {}", remaining);

            // Success criteria: exactly one row left (id=3) AND it is the correct one.
            assertEquals(1L, remaining,
                    "Q1: if _rowid delete works, we should have 1 row left");
        } catch (Exception e) {
            // Record the *exact* exception. Comment this back onto #63 verbatim.
            LOG.error("Q1 RESULT: delete via _rowid FAILED", e);
            fail("Q1 conclusion: Dataset.delete cannot accept `_rowid` in its filter grammar. "
                    + "Actual exception: " + e.getClass().getName() + ": " + e.getMessage()
                    + ". Consequence: drop the `_rowid`-in-predicate branch from Phase B "
                    + "of #63; the PK path becomes the sole delete path.");
        }
    }

    // ---------------------------------------------------------------------
    // Q2 — Does `Dataset.delete("id IN (1,2,3)")` correctly and idempotently
    //      remove those rows on a single-column BIGINT PK ?
    // ---------------------------------------------------------------------
    @Test
    @DisplayName("Q2: Dataset.delete(\"id IN (...)\") — correctness + idempotency")
    void q2_deleteBySingleColumnPk() throws Exception {
        String uri = tempDir.resolve("q2").toString();

        Schema schema = new Schema(Arrays.asList(
                Field.nullable("id", new ArrowType.Int(64, true)),
                Field.nullable("name", ArrowType.Utf8.INSTANCE)));

        // Seed 5 rows: id = 1..5.
        try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
            root.setRowCount(5);
            BigIntVector idV = (BigIntVector) root.getVector("id");
            VarCharVector nameV = (VarCharVector) root.getVector("name");
            for (int i = 0; i < 5; i++) {
                idV.setSafe(i, (long) (i + 1));
                setVarchar(nameV, i, "row-" + (i + 1));
            }
            Dataset.write()
                    .uri(uri)
                    .allocator(allocator)
                    .schema(schema)
                    .mode(WriteParams.WriteMode.CREATE)
                    .execute()
                    .close();
        }

        // First delete: remove id IN (2, 4).
        try (Dataset ds = Dataset.open().uri(uri).allocator(allocator).build()) {
            assertEquals(5L, ds.countRows(), "Seed dataset should have 5 rows");
            ds.delete("id IN (2, 4)");
            long remaining = ds.countRows();
            LOG.info("Q2 pass1: after delete id IN (2,4), remaining = {}", remaining);
            assertEquals(3L, remaining, "Q2: after removing 2 of 5, expect 3 rows left");

            // Verify the RIGHT rows survived by counting per-id.
            assertEquals(1L, ds.countRows("id = 1"), "Q2: id=1 must survive");
            assertEquals(0L, ds.countRows("id = 2"), "Q2: id=2 must be gone");
            assertEquals(1L, ds.countRows("id = 3"), "Q2: id=3 must survive");
            assertEquals(0L, ds.countRows("id = 4"), "Q2: id=4 must be gone");
            assertEquals(1L, ds.countRows("id = 5"), "Q2: id=5 must survive");
        }

        // Second delete with the SAME predicate — replay-safety check.
        try (Dataset ds = Dataset.open().uri(uri).allocator(allocator).build()) {
            ds.delete("id IN (2, 4)");
            long remainingAfterReplay = ds.countRows();
            LOG.info("Q2 pass2 (replay): after re-issuing same delete, remaining = {}",
                    remainingAfterReplay);
            assertEquals(3L, remainingAfterReplay,
                    "Q2: replaying the same PK-IN delete must be a no-op (idempotent)");
        }

        LOG.info("Q2 CONCLUSION: PK-column IN() delete is correct AND replay-idempotent. "
                + "This is the mainline path for Phase B of #63.");
    }

    // ---------------------------------------------------------------------
    // Q3 — Composite primary key: does tuple-IN work, or must we fall back
    //      to (AND) OR (AND) ? Also tests TIMESTAMP literal syntax.
    // ---------------------------------------------------------------------
    @Test
    @DisplayName("Q3: Composite PK delete syntax — tuple-IN vs. OR-of-AND fallback")
    void q3_deleteByCompositePk() throws Exception {
        String uri = tempDir.resolve("q3").toString();

        Schema schema = new Schema(Arrays.asList(
                Field.nullable("user_id", new ArrowType.Int(64, true)),
                Field.nullable("event_ts",
                        new FieldType(true, new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC"), null)
                                .getType()),
                Field.nullable("payload", ArrowType.Utf8.INSTANCE)));

        // Seed 3 rows with a composite (user_id, event_ts) key.
        long ts1 = 1_700_000_000_000_000L; // 2023-11-14T22:13:20Z in micros
        long ts2 = 1_700_000_001_000_000L;
        long ts3 = 1_700_000_002_000_000L;

        try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
            root.setRowCount(3);
            BigIntVector uv = (BigIntVector) root.getVector("user_id");
            TimeStampMicroVector tv = (TimeStampMicroVector) root.getVector("event_ts");
            VarCharVector pv = (VarCharVector) root.getVector("payload");
            uv.setSafe(0, 100L); tv.setSafe(0, ts1); setVarchar(pv, 0, "a");
            uv.setSafe(1, 100L); tv.setSafe(1, ts2); setVarchar(pv, 1, "b");
            uv.setSafe(2, 200L); tv.setSafe(2, ts3); setVarchar(pv, 2, "c");
            Dataset.write()
                    .uri(uri)
                    .allocator(allocator)
                    .schema(schema)
                    .mode(WriteParams.WriteMode.CREATE)
                    .execute()
                    .close();
        }

        // ---- Q3a: try tuple-IN syntax (nice-to-have) ----
        // Lance filter grammar note: literals for timestamps are written as
        // `timestamp 'YYYY-MM-DDTHH:MM:SS.fffffff'` per the docs at
        // http://www.lance.org/guide/read_and_write .
        String tupleIn = "(user_id, event_ts) IN ((100, timestamp '2023-11-14T22:13:20'),"
                + " (200, timestamp '2023-11-14T22:13:22'))";
        boolean tupleInWorks = tryDelete(uri, tupleIn, "Q3a tuple-IN");

        // ---- Q3b: OR-of-AND fallback (must work for Phase B) ----
        // Reset: re-create the same dataset because Q3a may or may not have
        // partially applied. Simplest way to reset: open a new @TempDir path.
        String uri2 = tempDir.resolve("q3b").toString();
        try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
            root.setRowCount(3);
            BigIntVector uv = (BigIntVector) root.getVector("user_id");
            TimeStampMicroVector tv = (TimeStampMicroVector) root.getVector("event_ts");
            VarCharVector pv = (VarCharVector) root.getVector("payload");
            uv.setSafe(0, 100L); tv.setSafe(0, ts1); setVarchar(pv, 0, "a");
            uv.setSafe(1, 100L); tv.setSafe(1, ts2); setVarchar(pv, 1, "b");
            uv.setSafe(2, 200L); tv.setSafe(2, ts3); setVarchar(pv, 2, "c");
            Dataset.write()
                    .uri(uri2)
                    .allocator(allocator)
                    .schema(schema)
                    .mode(WriteParams.WriteMode.CREATE)
                    .execute()
                    .close();
        }

        String orOfAnd =
                "(user_id = 100 AND event_ts = timestamp '2023-11-14T22:13:20')"
                        + " OR "
                        + "(user_id = 200 AND event_ts = timestamp '2023-11-14T22:13:22')";
        boolean orOfAndWorks = tryDelete(uri2, orOfAnd, "Q3b OR-of-AND");
        if (orOfAndWorks) {
            try (Dataset ds = Dataset.open().uri(uri2).allocator(allocator).build()) {
                long remaining = ds.countRows();
                LOG.info("Q3b: after OR-of-AND delete, remaining = {} (expected 1)", remaining);
                assertEquals(1L, remaining,
                        "Q3b: OR-of-AND composite PK delete must leave exactly 1 row");
            }
        }

        LOG.info("Q3 CONCLUSION: tupleIn={}, orOfAnd={}. "
                        + "Phase B predicate builder should prefer {} and fall back to {}.",
                tupleInWorks, orOfAndWorks,
                tupleInWorks ? "tuple-IN" : "OR-of-AND",
                tupleInWorks && orOfAndWorks ? "OR-of-AND if tuple-IN unavailable" : "N/A");

        // The epic gate: at least one form MUST work; otherwise Phase B is dead.
        if (!tupleInWorks && !orOfAndWorks) {
            fail("Q3: NEITHER tuple-IN nor OR-of-AND composite-PK delete works. "
                    + "Phase B of #63 needs to be redesigned around a different strategy "
                    + "(e.g. per-row single-key delete calls, or merge-insert-based upsert).");
        }
    }

    // ---------------------------------------------------------------------
    // Helpers
    // ---------------------------------------------------------------------
    private static void setVarchar(VarCharVector v, int idx, String s) {
        v.setSafe(idx, s.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * Best-effort DELETE. Returns true on success; logs and returns false on
     * failure (so Q3 can decide whether the fallback is needed).
     */
    private boolean tryDelete(String uri, String predicate, String label) {
        LOG.info("{}: attempting Dataset.delete(\"{}\")", label, predicate);
        try (Dataset ds = Dataset.open().uri(uri).allocator(allocator).build()) {
            ds.delete(predicate);
            LOG.info("{} RESULT: delete SUCCEEDED", label);
            return true;
        } catch (Exception e) {
            LOG.warn("{} RESULT: delete FAILED: {}: {}",
                    label, e.getClass().getName(), e.getMessage());
            return false;
        }
    }

    /**
     * Read all rows of a scan and extract the `_rowid` column as a long[].
     * The scan must have been created with ScanOptions.withRowId(true).
     */
    private static long[] readAllRowIds(ArrowReader reader) throws Exception {
        // Row-id column is added by Lance as `_rowid` (BigInt / UInt64).
        long[] out = new long[0];
        while (reader.loadNextBatch()) {
            VectorSchemaRoot batch = reader.getVectorSchemaRoot();
            org.apache.arrow.vector.FieldVector v = batch.getVector("_rowid");
            assertNotNull(v, "Scan did not produce a _rowid column even with withRowId(true)");
            int n = batch.getRowCount();
            long[] extended = Arrays.copyOf(out, out.length + n);
            for (int i = 0; i < n; i++) {
                Object obj = v.getObject(i);
                // _rowid is unsigned 64-bit; Arrow surfaces as Long or BigInteger.
                if (obj instanceof Number) {
                    extended[out.length + i] = ((Number) obj).longValue();
                } else {
                    extended[out.length + i] = Long.parseUnsignedLong(obj.toString());
                }
            }
            out = extended;
        }
        // No stable ordering guarantee across scans — sort for deterministic logs.
        long[] copy = Arrays.copyOf(out, out.length);
        Arrays.sort(copy);
        return copy;
    }

    @SuppressWarnings("unused")
    private static void unusedImportsAnchor(List<Object> ignored) {
        // Anchors the Collections import for potential future test additions.
        Collections.emptyList();
    }
}

---

File 3 of 3 — README.md

# Phase 0 Spike — DELETE feasibility on `org.lance:lance-core:7.0.0`

Throwaway Maven module attached to issue
[#63](https://github.com/lance-format/lance-flink/issues/63). Its sole purpose
is to answer three concrete questions **before** we design Phase A/B/C:

| # | Question | Consequence if answer is "no" |
|---|---|---|
| **Q1** | Can `Dataset.delete("_rowid IN (...)")` accept the `_rowid` virtual column as a predicate column, when the dataset was created with `enableStableRowIds(true)` ? | Drop the `_rowid`-in-predicate branch from Phase B. PK becomes the sole delete path. Phase D (stable row IDs as optional optimization) may still be worth pursuing, but for **compaction efficiency only**, not DELETE identity. |
| **Q2** | Does `Dataset.delete("id IN (1,2,3)")` on a single-column BIGINT PK remove exactly those rows, and is a repeated call a no-op ? | If NO, `Dataset.delete(predicate)` is unusable as our v1 delete primitive and we need to fall back to `Transaction.builder().operation(Operation.Delete).affected_rows(...)` for Phase B, moving the transactional work forward from Phase C. |
| **Q3** | For composite PK `(user_id BIGINT, event_ts TIMESTAMP)`, which of these syntaxes does Lance's filter parser accept: (a) `(user_id, event_ts) IN ((100, timestamp '...'), ...)` , (b) `(user_id = 100 AND event_ts = timestamp '...') OR ...` ? | The predicate builder in Phase B uses whichever works. If **neither** works, Phase B is dead and we redesign around `merge_insert` or per-row single-key delete calls. |

## Layout

spike/phase0/
├── pom.xml # Independent Maven project; NOT in root
├── README.md # This file
└── src/test/java/org/apache/flink/spike/lance/delete/
└── Phase0SpikeIT.java # Q1 / Q2 / Q3 in one file, one @test each

## How to run

```bash
cd spike/phase0
mvn -B test 2>&1 | tee /tmp/phase0-spike.log

mvn test will print each question's outcome to stdout via SLF4J
(INFO level messages tagged Q1 RESULT / Q2 RESULT / Q3 CONCLUSION)
and the JUnit summary at the end.

Do not add this module to the root pom.xml. It exists in isolation so
that a broken Spike does not break the main build.

What to paste back onto #63

After running, extract the following from /tmp/phase0-spike.log and paste it
as a comment on issue #63:

### Phase 0 Spike results (org.lance:lance-core:7.0.0, arrow 18.3.0, run on <date>)

Q1 — `Dataset.delete("_rowid IN (...)")` on stable-row-ids dataset:
  Result: PASS / FAIL
  If FAIL, exception: <class name>: <message>

Q2 — `Dataset.delete("id IN (2, 4)")` on 5-row BIGINT-PK dataset:
  First pass:   remaining=<n> (expected 3)   → PASS / FAIL
  Replay pass:  remaining=<n> (expected 3)   → PASS / FAIL

Q3 — Composite PK delete syntax:
  Tuple-IN     `(user_id, event_ts) IN (...)` : SUCCEEDED / FAILED (<exc>)
  OR-of-AND   `(u=100 AND ts=...) OR (...)`  : SUCCEEDED / FAILED (<exc>)
  Phase B builder will emit: <preferred form>

Then close this Spike branch. Do not merge the spike/phase0/ directory into
main — its job is done once the comment is posted on #63.

Decision matrix (for future readers)

Q1	Q2	Q3	Phase B design
pass	pass	any pass	Ship as designed: PK-mainline, _rowid as narrow secondary path for Lance→Lance replication.
fail	pass	any pass	Ship PK-only; strike all _rowid references from #63. Update #63 to note "Spike-verified: no _rowid predicate support".
any	fail	any	Escalate: Dataset.delete(predicate) cannot be trusted. Phase B needs Transaction.builder().operation(Operation.Delete) (Phase C work pulled forward).
any	pass	both fail	Composite PK unsupported via .delete(String). Phase B for composite-PK tables falls back to per-row delete("pk1=v AND pk2=v'") calls, batched by checkpoint; document the performance impact.

Known caveats built into the test file

Read the top-of-file comment in Phase0SpikeIT.java
for two ground-truth items:

1. Dataset.write().uri(...).allocator(...).schema(...).mode(CREATE).execute()
   vs. Fragment.write(...) vs. Dataset.create(alloc, uri, schema, params).
   The 7.0.0 SDK exposes three overlapping paths. The Spike uses
   Dataset.write() because it's the highest-level and closest to what the
   Flink connector will end up using. If the write path itself fails, that is
   itself a Phase 0 finding worth reporting on [epic] DELETE via PRIMARY KEY (reworked after adversarial review, replaces PR #19) #63.

2. _rowid extraction requires ScanOptions.withRowId(true). The scanner
   API is LanceScanner.create(dataset, options, allocator).scanBatches()
   which returns an ArrowReader. The helper readAllRowIds handles both
   Long- and BigInteger-shaped Arrow surfaces for the u64 row-id column.

---

### Result reporting template

After running `mvn test`, please reply to this comment with:

Phase 0 Spike results (org.lance:lance-core:7.0.0, arrow 18.3.0, run on , platform <os/arch>)

Q1 — Dataset.delete("_rowid IN (...)") on stable-row-ids dataset:
Result: PASS / FAIL
If FAIL, exception: :

Q2 — Dataset.delete("id IN (2, 4)") on 5-row BIGINT-PK dataset:
First pass: remaining= (expected 3) -> PASS / FAIL
Replay pass: remaining= (expected 3) -> PASS / FAIL (idempotency)

Q3 — Composite PK delete syntax:
Tuple-IN (user_id, event_ts) IN (...) : SUCCEEDED / FAILED ()
OR-of-AND (u=100 AND ts=...) OR (...) : SUCCEEDED / FAILED ()
Phase B builder will emit:

Decision matrix outcome: <one of the four rows in README.md's decision matrix>

Once results land, we'll amend the epic body accordingly (drop `_rowid` references if Q1 fails, pull Phase C forward if Q2 fails, etc.) — see the decision matrix in the inlined README above.

### Why inline instead of a branch

Spike code has a one-shot lifecycle: run, extract the yes/no answer, discard. Pinning it to this issue's timeline keeps the trail (v1 body → adversarial review → v2 body → spike results → informed Phase B/C design) coherent. A dedicated branch would add repo noise without adding discoverability, and there's no cross-platform matrix that would justify multi-contributor branch coordination at this stage.

[LuciferYang]

Q1 and Q2 can be answered from the Lance source without spending the spike. I checked against v12.0.0-beta.6 plus 20 commits (2026-09-01). Q3 is still open.

First, a fact that makes Q1 and Q2 the same question: delete's predicate surface is scan's. DeleteJob::execute_impl in rust/lance/src/dataset/write/delete.rs builds self.dataset.scan(), calls with_row_id().project(&[ROW_ID]), and then hands the string to scanner.filter(s). Any predicate form covered by the scanner tests therefore holds for delete too.

Q1: yes, and _rowid is a genuine filter column rather than only a take special case. The doc comment on the scanner's take helper does describe an optimization that recognizes _rowid = 10, _rowid IN (10, 20, 30), and the _rowaddr / _rowoffset equivalents. Read alone it confirms the lance-duckdb worry. But test_filter_to_take in rust/lance/src/dataset/scanner.rs also has cases that assert the optimization is not applied (no OneShotStream in the explained plan) and still return the right rows:

check_no_opt("_rowid = 50 AND _rowid = 51", &[]).await;
check_no_opt("(_rowid = 50 AND idx < 100) OR _rowid = 51", &[50, 51]).await;

The second case puts _rowid and a user column in the same OR, which the take path cannot match, so it goes through ordinary filter evaluation. Both routes work. The optimized route is covered as well: check("_rowid IN (52, 51, 50, 17)", &[17, 50, 51, 52]), plus AND combinations written in both operand orders.

Used directly on delete, from Java: CompactionTest#testDeletionCompaction calls dataset.delete("_rowid <= 8"). Worth noting that <= is not one of the take-optimizable shapes, so that test exercises _rowid on the plain filter path through delete specifically. On the Rust side, cleanup.rs uses .delete(&format!("_rowid = {}", max_row_id)). And Dataset's javadoc for update already states that the predicate "may reference dataset columns as well as the system columns _rowid, _rowaddr, and _rowoffset".

Q2: yes. dataset.delete("i IN (1, 9)") in rust/lance/src/dataset/tests/dataset_io.rs.

Q3: still open. I could not find row-value tuple IN, meaning (a, b) IN ((..), (..)), used anywhere in the repo including tests. My suggestion is to treat the OR-of-ANDs shape as the primary plan rather than a fallback, and not gate Phase A or B on tuple IN at all.

Two limits on Q1. The Java test does not enable stable row ids, so it does not literally cover the "created with enableStableRowIds(true)" wording, though whether the filter parser resolves _rowid is independent of that flag. The flag changes what the value means: with it off, _rowid is the row address. Separately, Q1 passing does not touch design decision 1. A CDC -D carries the upstream key, not Lance's _rowid, so PK-first stays the mainline; what Q1 unlocks is the narrow Lance to Flink to Lance replication case.

That would leave Phase 0 as Q3 only, and Phase A and B unblocked now.

One unrelated thing I noticed while reading the epic, which I can raise separately rather than mix into this thread: Phase B buckets by RowKind while Phase C fixes the flush order as deletes first then inserts, and those two together drop a delete when +I(k) and -D(k) arrive in the same checkpoint.

[fightBoxing]

Q3 spike result — Phase B predicate shape (reporting back to #63)

Local spike run against org.lance:lance-core:7.0.0 (arrow 18.3.0, JDK 11), answering Q3 of the
epic's Phase 0: for a composite PK (user_id BIGINT, event_ts TIMESTAMP), which DELETE predicate
shape does the Lance filter parser accept?

Form	Result
tuple-IN (user_id, event_ts) IN ((..), (..)) (space or T separator)	FAIL — IllegalArgumentException: Expression '(user_id, event_ts)' is not supported SQL in lance (planner.rs:838)
OR-of-AND (user_id = 100 AND event_ts = timestamp '..') OR (..)	PASS — correct residual row count (1 of 3); both space and T timestamp separators work

Conclusion: Phase B's predicate builder must use the OR-of-AND shape for composite PKs and
must not rely on tuple-IN (it is not supported by the Lance filter parser). This is consistent
with @LuciferYang's review recommendation to treat OR-of-AND as the primary plan, not a fallback.

This is captured as a regression test in CompositePkDeleteITCase.