fix(sink): resolve architecture-review must-fix items for upsert sink (#76 follow-up)

[fightBoxing]

Summary

This PR applies the "must-fix" architecture-review outcomes from
#76 (feat: add primary-key upsert/delete and ALTER TABLE schema evolution)
as a self-contained, independently reviewable change.

It is branched from #76 (fightBoxing:feat/upsert-delete-alter-table @ 88b8c42)
so it can be merged either as a follow-up to #76 or squashed into #76 by
its author before that PR lands.

7 files changed, +334 / -128.
mvn -o clean test-compile on Flink 1.18 / 1.19 / 1.20: BUILD SUCCESS, 0 errors.

What changed and why

A1 + A3 + A5 — first-write path unified

The upsert sink previously took Fragment.write + Overwrite-commit on the
first write, gated by Files.exists(datasetPath). That had two defects:

1. Multi-subtask first-write clobber: Files.exists is not atomic with
   the subsequent Overwrite, so two subtasks that both observe "not
   exists" would both commit Overwrite, and one would silently lose its
   rows.
2. Remote storage broken: Files.exists(Paths.get("s3://…")) /
   Files.exists(Paths.get("tbdsfs://…")) returns false for every URI
   with a non-file scheme, so on remote storage every open would enter
   the first-write branch and re-Overwrite the entire dataset.

Now the sink uses an open-or-create strategy in open():

try { return Dataset.open(...); }
catch (…) { try { return Dataset.create(...); }
            catch (…) { return Dataset.open(...); /* peer won the race */ } }

This is atomic in the Lance native layer, works identically on local FS and
remote object stores, and makes the first-write path go through the
existing mergeInsert code (no separate Overwrite branch).

OVERWRITE mode is now explicitly rejected for remote paths in the sink —
truncation must be performed at DDL time via the catalog (which already
knows how to configure the storage environment).

A2 — checkpoint / consistency model tightened

• Delete before upsert within a flush. A half-failed flush must never
  leave behind a stale row that should have been superseded. With per-key
  buffer collapsing, this also keeps the operation idempotent under
  upstream replay.
• close() no longer implicitly flushes. Flink invokes close() on
  cancel / restart as well; writing on those paths violates the
  "checkpoint is the persistence boundary" contract. Rows still in the
  buffer are dropped and re-delivered by the (replayable) source on
  restart. This makes the at-least-once semantics honest.
• NULL and non-finite float primary-key values are rejected
  explicitly in the DELETE predicate builder. Previously col = NULL
  produced a silently no-op predicate (SQL UNKNOWN), so a -D with a
  NULL PK would silently fail to delete anything.
• OVERWRITE local-directory delete failures escalate to IOException
  (was LOG.warn, would previously leave half-old / half-new data).

B2 — buffer is now bounded

invoke() now triggers an early flush() once the collapsed key count
reaches write.batch-size. The check happens between events (never
mid-flush), so the delete-before-upsert invariant is preserved. Prevents
unbounded heap growth in the gap between checkpoints on wide-key workloads.

B5 — foreign-namespaced dataset config is now preserved

applyTableProperties UNSET path no longer deletes config keys that look
namespaced (engine.category.name). A Flink ALTER TABLE t RESET (...)
will no longer silently remove metadata written by sister engines
(Spark / Trino / Ray) on the same Lance dataset.

Engineering quality

• LanceDynamicTableSink assigns a stable operator UID
  (lance-upsert-sink) + name so state mapping survives job upgrades and
  savepoints.
• PrimaryKeySelector.project() extracted as a public static method;
  LanceUpsertSink.extractKey() delegates to it — the keyBy routing key
  and the buffer key can never drift.
• PrimaryKeyPersistence.persist() rejects column names containing a
  comma (would corrupt the comma-delimited encoding).
• ArrowArrayStreams narrowed from public to package-private (only
  called by LanceUpsertSink in the same package).
• LanceUpsertSinkITCase.twoSubtasksConcurrentFirstWrite() is now truly
  concurrent (CountDownLatch double-barrier + ExecutorService) —
  previously executed serially and would not have caught the A1 regression.
• New LanceUpsertSinkITCase.closeDoesNotImplicitlyFlush() regression
  test for the A2 close() semantics.

Deliberately deferred to follow-up issues

The following review items are not addressed here. Each has a
dedicated tracking issue that describes the constraint, options, and
acceptance criteria:

• A4 — replace SQL-string DELETE predicate with typed key encoding
• A6 — consume Flink TableChange in LanceCatalog.alterTable
  instead of the SchemaDiff heuristic
• A7 / B4 — single source of truth for connector option classification;
  cross-engine PK metadata key
• B1 / B3 — hot-key metrics and RootAllocator cap (the latter needs
  cross-component coordination across LanceSink / LanceUpsertSink /
  LanceCatalog / LanceNamespaceCatalog and should not land piecemeal)

Verification

JAVA_HOME=$(/usr/libexec/java_home -v 11) mvn -o clean test-compile
[INFO] Lance Flink Connector - Flink 1.18 ................. SUCCESS
[INFO] Lance Flink Connector - Flink 1.19 ................. SUCCESS
[INFO] Lance Flink Connector - Flink 1.20 ................. SUCCESS
[INFO] BUILD SUCCESS   (0 errors)

I did not run mvn test because it requires the Lance native library
and platform-specific artifacts I don't have set up locally. CI on this
repo should exercise all of LanceUpsertSinkITCase,
LanceCatalogTableITCase, SchemaDiffTest, and
LanceSchemaEvolutionITCase end-to-end.

Relationship to #76

This PR should not be merged before #76 — it depends on
88b8c42 as its base commit. Options for the reviewers:

• Squash-merge this into feat/upsert-delete-alter-table before merging
  feat: add primary-key upsert/delete and ALTER TABLE schema evolution #76 into main (recommended: keeps history minimal).
• Land feat: add primary-key upsert/delete and ALTER TABLE schema evolution #76 first, then merge this PR into main as a follow-up.

Either way, the resulting main state is identical.

[github-actions]

ACTION NEEDED
Lance follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

For details on the error please inspect the "PR Title Check" action.

[fightBoxing]

Follow-up issues (deliberately deferred, tracking):

• Replace SQL-string DELETE predicate with typed key encoding in LanceUpsertSink #78 — A4: Replace SQL-string DELETE predicate with typed key encoding
• Consume Flink TableChange in LanceCatalog#alterTable instead of inferring from SchemaDiff #79 — A6: Consume Flink TableChange in LanceCatalog#alterTable
• Single source of truth for connector option classification; retire dot-namespace UNSET heuristic #80 — A7/B4/B5: Single source of truth for connector option classification
• Upsert sink production hardening: hot-key metrics and Arrow allocator caps #81 — B1/B3: Hot-key metrics and Arrow allocator caps

Each has context, options considered, and acceptance criteria.

[fightBoxing]

Review: 请勿合并——本 PR 引入了 6 个 ITCase 回归，且 CI 看不到

重构方向我是认同的：用 openOrCreate 统一首写与稳态路径、消除 Overwrite 对同伴 subtask 的覆盖风险，这是对的。close() 不再隐式 flush 也是正确的判断。

但当前状态不能合并。本 PR 让 LanceUpsertSinkITCase 从全绿变成 6 个失败，而这个变化在 CI 里完全不可见。

实测对照

*ITCase 不被 surefire 的默认 include 匹配，而 pom.xml 至今没有配 failsafe（只有第 25 行一句注释提到它），所以这批用例在 CI 中从未执行。我在本地用 -Dtest='*ITCase' 强制跑，三方对照如下：

版本	LanceUpsertSinkITCase 结果
#76 head (88b8c42)	8 run, 0 failed
本 PR head (257a99c)	9 run, 6 failed
修复后（本地）	9 run, 0 failed

失败清单：

LanceUpsertSinkITCase.insertThenDeleteSameFlushLeavesKeyAbsent:146
LanceUpsertSinkITCase.deleteThenInsertSameFlushLeavesKeyPresent:170
LanceUpsertSinkITCase.updateAfterReplacesPreviousValue:195
LanceUpsertSinkITCase.replaySameUpsertBatchIsIdempotent:227
LanceUpsertSinkITCase.replayDeleteBatchIsIdempotent:246
LanceUpsertSinkITCase.twoSubtasksConcurrentFirstWrite:326

这些正是本 PR 声称要保证的语义：同 flush 内的折叠、replay 幂等、并发首写。

根因一：mergeInsert 返回新句柄，被丢弃了（前 5 个失败）

mergeInsert 不修改传入的 Dataset，它提交一个新版本并通过 MergeInsertResult#dataset() 返回新句柄。本 PR 在 open() 里一次性取得 this.dataset，之后每次 mergeInsertRows / deleteRows 都丢弃返回值，所以 this.dataset 永远钉在 open() 时的快照上。

后果是写不丢、读陈旧：Lance 提交时按最新表状态解析，所以数据落盘是对的；但任何用这个旧句柄做的读（包括测试里的 countRows）看到的都是变更前的状态。insertThenDeleteSameFlushLeavesKeyAbsent 报 expected: 0L but was: 1L 就是这个原因。

修法是在两处调用点接住返回值：

MergeInsertResult result = ArrowArrayStreams.mergeInsert(dataset, params, allocator, root);
this.dataset = result.dataset();

这一条同时解释了为什么 #76 版反而没问题：#76 在 flush() 里按 dataset == null 分支走 createDataset()，每轮重新拿句柄，掩盖了这个语义。

根因二：PrimaryKeyPersistence.persist 在并发首写时必冲突（第 6 个失败）

open() 第 190 行注释写的是 "Idempotent; safe to call on every open"，这个论断不成立。Dataset#updateConfig 是版本化事务，不是幂等 put。两个 subtask 同时 open() 时，第二个提交会被 Lance 的冲突解析器拒绝：

java.lang.RuntimeException: Incompatible transaction: This UpdateConfig transaction is
incompatible with concurrent transaction UpdateConfig at version 2.,
rust/lance/src/io/commit/conflict_resolver.rs:1132:34

把 persist 从 flush() 的首写分支提到 open()，恰好把这个调用从"只有一个 subtask 执行"变成"每个 subtask 都执行"，所以是这次重构放大了它。

建议两道防线：写前先读当前值，一致就不发起事务；冲突时先 checkoutLatest() 把句柄推进到最新版本再重读校验——这一步不能省，因为 getConfig() 读的是 open() 时的旧快照，不推进就永远看不到对端的提交。

前置建议

这两个根因都属于"SDK 方法存在但语义与直觉不符"，靠读代码评审很难发现，只能靠集成测试拦住。所以建议先把 failsafe 的 <executions> 绑定补上（integration-test + verify 两个 goal），让这批 ITCase 真正进入构建，再谈这个 PR 的合并。否则下一次同类回归依然会带着全绿的 CI 进来。

上述修复我已在本地完成并验证（三模块 surefire 308 + failsafe 83 全绿），可以在本 PR 更新后提交，或者由我另开 PR 接在后面。

[fightBoxing]

Closing in favour of #82, which contains this PR's commit plus fixes for the two regressions it carried.

To recap the reason, since the direction here was right and worth keeping: moving dataset creation and primary-key persistence out of flush() and into open() does remove the Overwrite-based first write that could clobber a peer subtask. But mergeInsert returns a new handle rather than mutating the receiver, and this PR discarded it, so this.dataset stayed pinned to the open() snapshot — writes landed, reads went stale. And hoisting PrimaryKeyPersistence.persist into open() turned a once-per-dataset updateConfig into once-per-subtask, which Lance's conflict resolver rejects; the "Idempotent; safe to call on every open" comment was the wrong assumption.

Neither was visible in CI because *ITCase files were never executed — surefire's default includes don't match that suffix and failsafe had no goal binding. #82 adds that binding first, so the 6 failures this PR introduced are now caught by the build rather than by hand.

Full detail in the review above.