feat: support Spark Structured Streaming writes

[LuciferYang]

Closes #246.

Summary

Adds a Structured Streaming sink for Lance. Each non-empty micro-batch issues one Lance transaction stamped with lance.streaming.queryId and lance.streaming.epochId in its transaction properties; replay dedupe scans recent transaction history via DatasetDelta.listTransactions for a matching (queryId, epochId) pair and skips the commit on a hit.

SparkWriteBuilder implements SupportsStreamingUpdateAsAppend and LanceDataset advertises TableCapability.STREAMING_WRITE (on both the plain and blob-v2 capability sets). Append / Complete / Update output modes flow through the existing V2 plumbing. CTAS and staged commits are rejected at toStreaming() with a clear error.

streamingQueryId is a required option — the idempotency key for the dedupe scan; it must be globally unique across concurrent queries on the same table. lance.streaming.dedupe.lookback.versions (default 100, max 10 000) caps the per-commit scan depth.

Output modes & empty epochs

Spark mode	Lance behavior
append (default)	one Append per epoch
complete	one Overwrite per epoch (truncate + write the full result)
update	append-only via SupportsStreamingUpdateAsAppend (delta rows appended, not merged)

The streaming overwrite signal is driven solely by Spark's output mode (truncate()), never by a write_mode option — so a stray write_mode=overwrite cannot turn an Append stream into a per-epoch truncate.

Empty epochs: an empty append epoch issues no transaction; an empty complete epoch still commits one empty Overwrite to truncate the table to the new (empty) result (skipping it would leave the prior epoch's rows on disk).

Why one transaction per epoch

PR #399 used a two-transaction Append + UpdateConfig design, which doubled manifest growth and per-epoch latency at scale. Stamping the identity inside the Append moves the dedupe signal into transaction history and keeps writes to one manifest update per epoch.

Why the sink doesn't pin the dataset version

LanceBatchWrite pins dataset.version() in its constructor because it commits exactly once. Streaming reuses one sink across many epochs, so a pinned version would go stale immediately — the dedupe scan would point at the wrong range and the transaction's readVersion would lag. Every commit opens at the current latest instead, and toStreaming() strips any pinned version to enforce this. testMultipleEpochsOnSameSinkAdvanceVersionMonotonically regresses this path.

Robustness

• Vended credentials. The streaming commit merges namespace-vended initialStorageOptions (from describeTable) into the CommitBuilder write params, mirroring LanceBatchWrite, so writes to credential-vending namespace / object-store tables authenticate.
• Bounded at-least-once. The dedupe scan covers the last lookback versions; a duplicate can slip through once lookback-or-more unrelated commits intervene before a replay. If a version inside the window is removed by VACUUM/cleanup (making listTransactions throw), the scan degrades to that bounded at-least-once fallback (logs a WARN and proceeds) rather than permanently wedging the query.
• streamingQueryId is trimmed once and used verbatim as the stamped key (no whitespace asymmetry).

Compatibility

Supported on Spark 3.4 / 3.5 / 4.0 / 4.1. On 3.4, StreamingWrite.useCommitCoordinator is absent from the interface so the override is inert there; Lance commits each epoch atomically via CommitBuilder, making the commit coordinator immaterial on every version. A concrete TestStreamingWrite runs the shared suite on each version.

Tests

• BaseStreamingWriteTest (runs on 3.4/3.5/4.0/4.1) — 10 cases: append happy-path, missing & whitespace streamingQueryId rejection, empty-epoch no-op, real non-empty replay dedupe, lookback-window expiry, multi-epoch monotonic version, overwrite replace + empty truncate, and write_mode=overwrite-does-not-truncate-append.
• LanceSparkWriteOptionsTest — lookback default / round-trip / bounds and non-numeric parse-error cases.
• SparkWriteTest — toStreaming returns LanceStreamingWrite when configured, throws IllegalArgumentException without streamingQueryId, rejects staged commits.

Local verification: make lint clean; full base + Spark 3.5 suite green (464 + 1064, 0 failures); streaming tests 10/10 on Spark 3.4 and 3.5.

Out of scope

• Streaming reads (MicroBatchStream) — follow-up.
• Row-level UPDATE / DELETE on the streaming path.
• Blob v2 source-context copy-through (a batch-only optimization) does not fire on the streaming path; direct blob-data streaming works.
• In-memory dedupe cache to skip the per-commit DatasetDelta scan.
• High-frequency (< 30s trigger) and Spark 4.1 Real-Time Mode writes — MemWAL-backed sink follow-up (design + Java API gap analysis pending). Docs section "Trigger cadence, fragment count & compaction" documents the fragment-count cost and Spark's own rejection behavior for these modes.

Docs at docs/src/streaming.md (wired into the site nav).

[LuciferYang]

cc @hamersaw @jackye1995 @fangbo

[fangbo]

@LuciferYang I'm not familiar with Spark Structured Streaming. I'm very sorry that I cannot review this PR.

[hamersaw]

@LuciferYang I'm curious about your use-case here? By appending a new fragment per epoch the "many small fragments" problem is going to cause issues. Even at a 10s epoch interval this means 8640 fragments per day. Is this not a concern for your use case? I'm wondering if it would make sense to build on top of the WAL implementation in OSS for this instead?

[LuciferYang]

Good to pin down the scope of this PR and how alternatives fit.

1. Target workload: coarse-grained micro-batch, not sub-second.
This sink targets Trigger.ProcessingTime in the tens-of-seconds-to-minutes range, Trigger.AvailableNow backfills, and Kafka / CDC pipelines that already batch upstream. Using a 1-minute trigger as a reference (Spark schedules the next batch at max(triggerInterval, batchCompletionTime), so busy pipelines produce fewer than the nominal count), that's ~1440 fragments/day — the same order of magnitude as a daily batch job, well within normal OPTIMIZE cadence. Under-30-second triggers do hit real manifest growth; I'll document that as not recommended, but I don't plan to enforce it with a hard error in this PR — legitimate edge cases exist (short experiments, mixing with AvailableNow), and a code-level threshold is easy to get wrong.

Relative to hand-rolled foreachBatch, this sink adds epoch idempotency ((queryId, epochId) dedupe scan), automatic Append/Complete/Update routing per Spark output mode, and credential-vending namespace pass-through. A user willing to buffer records across epochs and hand-roll idempotency in foreachBatch could avoid the per-epoch fragment growth, but the logic is error-prone.

2. Spark's engine already gates most sub-second paths.

• Trigger.Continuous requires the source to declare TableCapability.CONTINUOUS_READ; the Lance source doesn't, so Lance→Lance continuous queries fail at planning. A Kafka→Lance continuous query passes source-side validation and would reach Write.toStreaming() — but that combination produces catastrophic fragment counts for this sink, and I'll flag it as discouraged in the docs.
• Spark 4.1's Real-Time Mode uses RealTimeModeAllowlist — only Kafka / ForeachWriter / Console / Memory sinks are allowlisted; other DSv2 sinks (including Lance) fail at query start. I don't think we should work around that.
• Hand-rolled foreachBatch isn't an escape hatch — it invokes a batch write per epoch and produces the same manifest growth.

3. MemWAL is the natural foundation for a follow-up sink, not a replacement for this PR.
The OSS WAL implementation you're pointing at is, I believe, MemWAL (MemTable + WAL LSM) in lance-format/lance — its LSM staging (WAL append + MemTable buffering + background flush) is the right substrate for high-frequency writes. MemWAL has both Rust core and a Java module (lance/java/src/main/java/org/lance/memwal/, 17 classes covering ShardWriter.put, LsmScanner, initializeMemWal, sharding, stats), which gives us a foundation to build on. But whether Java exposes everything a Spark exactly-once sink needs — a way to stamp / query epoch identity on a WAL entry, WAL replay from an offset, explicit memtable seal / rotate, crash-consistent shard-state queries — I haven't verified end to end yet. That gap analysis is exactly what the follow-up issue is for, alongside the Spark-side design questions: where does the epoch idempotency key live (WAL entry, shard manifest, or dataset config)? What's the shard-to-Spark-task mapping? What's the consistency protocol between WAL replay and memtable-flush checkpoints on recovery? Non-trivial protocol choices — worth their own PR and review cycle.

4. Why not make MemWAL the only streaming path?
Because Lance tables don't have a MemWAL index by default. Existing tables and users who don't need high-frequency writes shouldn't be forced to provision MemWAL just to use the streaming API. This sink works with any Lance table; a MemWAL sink would target tables that explicitly opt into high-frequency semantics. Complementary, not overlapping.

Concretely for this PR I'll:

1. Add a "Trigger cadence, fragment count & compaction" section to docs/src/streaming.md with the trigger→fragments/day table, Spark's rejection behavior for Trigger.Continuous (source-side) and Trigger.RealTime (sink allowlist), a call-out that Kafka→continuous→Lance-sink is discouraged, and an explicit "< 30s trigger not recommended — use the MemWAL sink when it lands" note.
2. Add "High-frequency (<30s) and Real-Time Mode writes — MemWAL-backed sink follow-up" to the PR description's "Out of scope" list.
3. Open a follow-up issue tracking the MemWAL-backed sink. Scope of that issue: (a) audit which primitives the current Java API covers vs. what a Spark exactly-once sink needs, (b) the exactly-once design questions above, (c) whether missing pieces belong to lance-java, lance Rust core, or lance-spark.

The Spark Structured Streaming users I've worked with are all in the coarse-grained micro-batch regime, which is what this sink targets. If you'd prefer code-level enforcement beyond docs + follow-up (e.g., a WARN at a trigger-interval threshold), happy to add it — but I'd lean on Spark's own allowlist for the hard boundary and use docs + PR scope in lance-spark.

Does this address the scope concern for this PR? @hamersaw