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.

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.