fix: adopt post-merge dataset handle, run ITCase in CI, and close type-mapping gaps

.gh-comments/issue-map-type-blocked-by-storage-version.md

@@ -0,0 +1,141 @@
+# MAP 类型支持的前置阻塞：Lance 文件格式版本
+
+## 结论
+
+`MAP` 映射不能只补类型转换层。Lance 的 Map **数据写入**要求文件格式 2.2+，而连接器
+当前所有写入路径都使用 SDK 默认的 2.1，因此照清单补完 `LanceTypeConverter` 与
+`RowDataConverter` 只会得到一个「建表成功、写入必失败」的功能——比不做更糟。
+
+## 实测证据
+
+spike 走生产 `mergeInsert` 路径（复用 `ArrowArrayStreamsTestAccess`，而非自建 harness）。
+
+**默认 2.1**：schema 可建、可往返，写数据时在 Rust 编码层被拒：
+
+```
+UnsupportedOperationException: Not supported: Map data type is only supported in
+Lance file format 2.2+, current version: 2.1
+  at lance-encoding/src/encoder.rs:485
+```
+
+注意失败点在**写入**而不是建表。`Dataset.create(schema)` 带 Map 列完全成功，重开后
+schema 也无损：`Map(false)<entries: Struct<key: Utf8 not null, value: Int(32, true)>>`。
+这正是这个缺口危险的地方——任何只验证 DDL 的测试都看不出问题。
+
+**显式 `withDataStorageVersion("2.2")`**：三种场景全部无损往返：
+
+| 行 | 写入 | 读回 | isNull |
+|---|---|---|---|
+| 0 | `{"a":1,"b":2}` | `[{"key":"a","value":1},{"key":"b","value":2}]` | false |
+| 1 | 空 map | `[]` | false |
+| 2 | NULL | `null` | true |
+
+第 2 行实证了 NULL map 走 `setNull` 路径正确，即上一轮 `MapVector` 分支顺序修正
+（`MapVector extends ListVector`，必须先匹配）所保护的场景。
+
+## 当前代码状态
+
+三处 `WriteParams` 构造点均未设置存储版本，全部走默认 2.1：
+
+- `LanceUpsertSink.java:231`（open-or-create）
+- `LanceCatalog.java:595`（CREATE TABLE 物化）
+- `LanceSink.java:169`（Fragment 写入，只设了 `maxRowsPerFile`）
+
+`WriteParams.Builder#withDataStorageVersion(String)` 在 lance-core 7.0.0 存在且有效
+（已实测）。
+
+## 因此的工作拆分
+
+**前置项：暴露存储版本配置** —— 已完成
+
+新增 `write.data-storage-version`，接入全部三处 `WriteParams` 构造点。默认值为
+`noDefaultValue()`，不硬编码 `2.1`——否则 SDK 升级默认版本时我们反而把用户钉死在旧版本。
+未设置时行为与改动前完全一致。
+
+`LanceCatalog#createTable` 的接入是必需的而非可选：版本在建表时固化，只在 sink 设置对
+已建表无效。该接入点此前**没有任何测试覆盖**——禁用它全部既有套件依然全绿，因此补了
+`DataStorageVersionMapITCase#createTableAppliesConfiguredVersion`，用
+`Dataset#getLanceFileFormatVersion()` 对比「显式 2.2」与「未设置」两张表，并实测禁用后
+该用例失败。
+
+**主体项：MAP 类型映射** —— 已完成
+
+三个转换方向（`flinkTypeToArrowField`、`arrowTypeToFlinkType`、`toDataType`）与
+`RowDataConverter` 的四个分派点全部接入。
+
+`LanceCatalog#createTable` 在检测到 MAP 列而 `write.data-storage-version` 明确低于 2.2 时
+直接拒绝，错误信息点名列与选项。版本**未设置**时只告警不拦截——有效默认归 SDK 所有，硬拦
+截会在 SDK 默认前移到 2.2 后误伤合法用法。
+
+## 实测确认的三件事
+
+**`setIndexDefined` 是必需的，不是保险。** 移除后纯内存的 `RowDataConverter` 往返测试依然
+全绿，因为它读回自己写的向量、从不经过 Lance 校验；但真实数据集立刻报
+`The field \`entries\` contained null values even though the field is marked non-null in the
+schema`（`lance-file/src/writer.rs:397`）。这说明这两层测试缺一不可，内存层对非空约束没有
+判别力。
+
+**写路径的 MapVector 前置分支同样必需。** 禁用 MAP 写分派后落进 `ListVector` 分支，报
+`Unsupported write type: MapType`。读路径同理——`ArrowType.Map` 必须在 `ArrowType.List`
+之前判断，否则 map 会退化成 `ARRAY<ROW<key, value>>` 且不再往返。
+
+**key/value 的类型范围远窄于顶层列。** 只支持 INT/BIGINT/FLOAT/DOUBLE/STRING，受
+`RowDataConverter` 的 `readArrayData`/`writeArrayData` 限制（ARRAY 元素同此约束）。
+`MAP<STRING, DATE>` 在 Arrow 层完全合法、建表会放行，但首次写入必失败——与 MAP 本身踩的
+是同一个坑，因此在 `mapEntriesField` 里提前拒绝。要放宽得先扩那两个 helper。
+
+## 用户侧易踩点
+
+`DataTypes.MAP(STRING(), INT())` 的 key 默认可空，而 Arrow 不允许可空 key，所以最自然的
+写法会被拒绝。正确写法是 `DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.INT())`。
+错误信息显式说明了这一点。
+
+## 过程中另外发现的问题
+
+**`mergeInsert` 不保证行序。** 首版往返断言按位置取行，实际读回顺序是
+`空 / NULL / {a,b}`，与写入顺序不同。断言已改为按 `id` 关联。这不是缺陷，但任何按位置
+断言的测试都会随机失败。
+
+**工厂与 `LanceOptions` 之间有 16 个选项重复定义**，键名完全重合（`path`、`write.mode`、
+`read.*`、`index.*`、`vector.*` 等）。这是 A7「双份真相」在另一处的同类复发，规模更大。
+本次按既有模式在两处都加了新选项以保持一致，未夹带重构——独立处理更安全。
+
+**「不支持类型」的测试样本第三次搬家。** `LanceNamespaceCatalogSchemaTest` 里这个用例先用
+DECIMAL、后用 MAP，两者各自获得映射后都得换；现已改用 `MULTISET`。这类测试天生会随能力
+扩张而失效，注释里记了迁移史以便下次直接换。
+
+## 未验证项
+
+- 2.1 与 2.2 数据集能否在同一路径混用（时间旅行读旧版本）。
+- 2.2 是否影响其他类型的编码或既有索引。
+
+## MULTISET —— 已完成
+
+推测得到证实：MULTISET 形状的 map（`Map<Utf8, Int32>`）在 2.1 上建表成功、写入失败，报的
+是与 MAP 完全相同的 `Map data type is only supported in Lance file format 2.2+
+(lance-encoding/src/encoder.rs:485)`。2.2 上写入正常。
+
+实现完全复用 MAP 的通路：探针确认 `MultisetType.getDefaultConversion()` 是
+`java.util.Map`、`RowData.getMap()` 可用，因此 MULTISET 在运行时就是 `MapData`。
+`readMap`/`writeMap` 的签名从收 `MapType` 改为直接收 key/value 两个 `LogicalType`，MAP 与
+MULTISET 共用同一实现，不存在平行分支。count 侧固定为非空 INT32。
+
+`LanceCatalog` 的版本校验无需改动——`containsMap` 按 `ArrowType.Map` 判断，MULTISET 映射成
+同一 Arrow 类型，自动覆盖。`setNull` 同理，上一轮加的 MapVector 前置分支直接生效。
+
+### 一个有意接受的不对称
+
+MULTISET 列**读回后呈现为 `MAP<element, INT>`**。Arrow map 不携带任何可区分
+`MAP<T NOT NULL, INT>` 与 `MULTISET<T NOT NULL>` 的信息，而 MAP 是远更常见的声明，因此无
+标记的 map 一律解析为 MAP。
+
+我实测过打元数据的可行性：`FieldType` 支持自定义 metadata，且 Lance **确实**原样往返了
+`{flink.type=MULTISET}`。技术上可行，但我选择不用——这会为一个极少作为表列声明的类型
+（MULTISET 主要来自 `COLLECT()` 聚合结果）在跨引擎共读的 schema 里塞进一个 `flink.` 专有
+键，正是 A7 遗留项在推动消除的那类耦合。写入与存储数据不受影响，仅恢复出的类型名不同。
+
+### 「不支持类型」测试样本第四次搬家，这次应该是最后一次
+
+`LanceNamespaceCatalogSchemaTest` 的样本历经 DECIMAL → MAP → MULTISET，现改为
+`INTERVAL`。前三者都是 Lance 真正会存储的数据类型，所以迟早都会获得映射；区间类型不属于
+分析存储的数据，不在这条路径上。

.gh-comments/issue-update-followups.md

@@ -0,0 +1,67 @@
+# UPDATE 实现过程中暴露的既有缺陷（已解决）
+
+这两项都不是 `UPDATE` 引入的。它们是实现 `UPDATE` 时因首次有端到端 SQL 测试覆盖而暴露
+出来的既有问题，均已在本轮修复。保留本文件作为背景记录。
+
+---
+
+## C1：`*ITCase` 测试在构建中从未被执行 —— 已修复
+
+**曾经的现象**
+
+surefire 的默认 includes 不匹配 `*ITCase` 后缀，而项目未配置 failsafe 插件。CI 跑的是
+`mvn verify`，但 `verify` 阶段的插件链里只有 surefire，没有任何 failsafe，因此 75 个
+集成测试从未在构建中运行——CI 每次"通过"都不包含它们，其中还藏着一个真实失败（见 C2）。
+
+**修复**
+
+在 `pom.xml` 的 pluginManagement 声明 `maven-failsafe-plugin`（3.3.0，因内部镜像无 3.1.2），
+并在 build/plugins 绑定 `integration-test` + `verify` 两个 goal。verify goal 是关键：
+没有它，报告会生成但构建仍然通过。
+
+failsafe 默认 includes 含 `**/*ITCase.java`，正好匹配项目约定，无需自定义。既有 ITCase
+未改名，保留 Flink 生态的标准命名。
+
+**验证**
+
+`mvn verify` 现在每模块执行 surefire 276 + failsafe 75，三个 Flink 版本模块全绿。
+移除 verify goal 曾确认失败的集成测试不会中断构建，加回后恢复拦截。
+
+---
+
+## C2：两个 subtask 并发首写时主键元数据提交冲突 —— 已修复
+
+**曾经的现象**
+
+`LanceUpsertSinkITCase#twoSubtasksConcurrentFirstWrite` 失败：
+
+```
+RuntimeException: Incompatible transaction: This UpdateConfig transaction is
+incompatible with concurrent transaction UpdateConfig at version 2.
+  at PrimaryKeyPersistence.persist(...)
+  at LanceUpsertSink.open(...)
+```
+
+**成因**
+
+`PrimaryKeyPersistence.persist` 无条件调用 `dataset.updateConfig` 写主键元数据。该调用
+是一个版本化的 Lance 事务，不是幂等 put；多个 subtask 并发首写时同时提交，Lance 的冲突
+解析器直接拒绝而非合并。原注释"Idempotent; safe to call on every open"是错误假设。
+
+**修复**
+
+让 `persist` 真正幂等，两道防线：
+
+1. 写入前先比较——值已一致则完全不发起事务，稳态零写入；
+2. 冲突时先 `dataset.checkoutLatest()` 把句柄推进到最新版本（`getConfig` 读的是 open 时
+   的旧快照，不推进就永远看不到对端提交），再重读校验；若目标值已达成则视为成功，
+   否则照常抛出。
+
+第 2 步的句柄推进是关键：移除它测试立即复现原冲突，证明句柄陈旧才是根因。冲突异常是
+Rust 层的裸 `RuntimeException` 无专用类型，因此用重读校验结果而非匹配消息来判断。
+
+**验证**
+
+`twoSubtasksConcurrentFirstWrite` 连续 3 次稳定通过；移除 `checkoutLatest` 立即复现原
+冲突（证明测试与诊断有效）。另加 `PrimaryKeyPersistenceConcurrencyTest`（5 个单元用例），
+其中一条专门钉住"容错不得吞掉真实分歧"——不同的 key list 仍必须写入。

docs/src/config.md

@@ -30,6 +30,70 @@ catalog types are `'lance'` (directory/S3) and `'lance-namespace'` (dir/rest).
 | `write.batch-size` | ❌ | 1024 | Write batch size |
 | `write.mode` | ❌ | append | `append` or `overwrite` |
 | `write.max-rows-per-file` | ❌ | 1000000 | Maximum rows per data file |
+| `write.data-storage-version` | ❌ | *(SDK default)* | Lance file format version for written data files, e.g. `2.2`. A `MAP` column requires 2.2+. Fixed when the dataset is created — changing it later does not upgrade an existing dataset. |
+
+Leaving `write.data-storage-version` unset lets the Lance SDK choose, which is
+why it has no default here: pinning today's default would hold the connector
+back once the SDK moves forward.
+
+Some Arrow types are gated on this version. A `MAP` column is accepted into the
+schema at `CREATE TABLE` on any version, but writing rows fails inside the Lance
+encoder below 2.2 — so set the option at table creation, not after the first
+write attempt.
+
+### MAP columns
+
+A `MAP` column requires `write.data-storage-version` to be `2.2` or newer.
+`CREATE TABLE` fails fast if the option is set to something older, naming the
+column. If the option is unset the connector only logs a warning, since the
+effective default belongs to the Lance SDK.
+
+Two constraints apply to the key and value types:
+
+```sql
+-- Rejected: DataTypes.MAP(STRING(), INT()) yields a nullable key, and Arrow
+-- does not allow one.
+CREATE TABLE t (attrs MAP<STRING, INT>) WITH (...);
+
+-- Correct: the key is declared NOT NULL.
+CREATE TABLE t (attrs MAP<STRING NOT NULL, INT>) WITH (
+  'write.data-storage-version' = '2.2', ...
+);
+```
+
+Keys and values support `INT`, `BIGINT`, `FLOAT`, `DOUBLE` and `STRING`. This is
+narrower than what a top-level column accepts — the same limit applies to
+`ARRAY` elements — and a type outside it is rejected at `CREATE TABLE` rather
+than on the first write. An empty map and a `NULL` map are stored distinctly. A
+`NULL` value is allowed; a `NULL` key is not.
+
+### MULTISET columns
+
+A `MULTISET` is stored as `MAP<element, count>`, so it carries the same 2.2
+requirement, the same element type restrictions, and the same `NOT NULL` rule —
+the element becomes the map key:
+
+```sql
+-- Rejected: DataTypes.MULTISET(DataTypes.STRING()) yields a nullable element.
+CREATE TABLE t (tags MULTISET<STRING>) WITH (...);
+
+-- Correct.
+CREATE TABLE t (tags MULTISET<STRING NOT NULL>) WITH (
+  'write.data-storage-version' = '2.2', ...
+);
+```
+
+The count side is always a non-null `INT`, since an element that is present has
+an occurrence count by definition.
+
+One asymmetry is worth knowing about: a `MULTISET` column **reads back as
+`MAP<element, INT>`**. An Arrow map carries nothing that separates
+`MAP<T NOT NULL, INT>` from `MULTISET<T NOT NULL>`, and `MAP` is the far more
+common declaration, so an untagged map resolves to `MAP`. Writes are unaffected,
+and the stored data is identical either way — only the recovered type name
+differs. Tagging the field with metadata would make the distinction survive, but
+it would put a Flink-specific key into a schema other engines also read, so it
+is deliberately not done.
 
 ### Vector index
 

docs/src/operations/dml/insert-into.md

@@ -1,7 +1,7 @@
 # INSERT INTO
 
-The Lance Flink sink appends rows to a Lance dataset. Write mode is controlled by
-the `write.mode` option.
+The Lance Flink sink writes rows to a Lance dataset. Behaviour depends on whether
+the table declares a primary key.
 
 ## Write modes
 
@@ -17,23 +17,97 @@ INSERT INTO vectors VALUES
     (1, 'Hello World', ARRAY[0.1, 0.2, 0.3, 0.4]);
 ```
 
+## Upsert mode
+
+When a table declares `PRIMARY KEY (...) NOT ENFORCED`, the sink accepts a CDC
+changelog stream and maps it onto Lance native operations:
+
+| Change | Applied as |
+|---|---|
+| `+I` / `+U` | `mergeInsert` with update-all + insert-all |
+| `-D` | key-only `mergeInsert` with matched-delete + not-matched-do-nothing |
+| `-U` | dropped (the new value carries all required state) |
+
+```sql
+CREATE TABLE users (
+    id BIGINT,
+    name STRING,
+    PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+    'connector' = 'lance',
+    'path' = '/data/users.lance'
+);
+```
+
+Deletes are matched by primary-key value, so any column type the connector can
+write can also serve as a primary key, including `DATE`, `TIMESTAMP`, `DECIMAL`
+and `VARBINARY`.
+
 ## Sink options
 
 | Option | Default | Description |
 |---|---|---|
 | `write.batch-size` | 1024 | Rows buffered before a flush |
 | `write.mode` | `append` | `append` or `overwrite` |
 | `write.max-rows-per-file` | 1000000 | Rows per data file |
+| `arrow.allocator-max-bytes` | unlimited | Upper bound in bytes for the Arrow allocator |
+
+### Bounding Arrow memory
+
+Each sink, source, catalog and index builder creates its own Arrow allocator.
+By default these are unbounded, which lets a single oversized batch exhaust
+off-heap memory and affect other slots in the same TaskManager.
+
+`arrow.allocator-max-bytes` caps each allocator individually:
+
+```sql
+CREATE TABLE users (
+    id BIGINT,
+    name STRING,
+    PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+    'connector' = 'lance',
+    'path' = '/data/users.lance',
+    'arrow.allocator-max-bytes' = '536870912'
+);
+```
+
+The bound is per allocator instance, not a total for the job. Size it against
+one component's working set — roughly `write.batch-size` times the row width,
+with headroom — rather than against the TaskManager's whole off-heap budget.
+
+## Hot primary keys
+
+In upsert mode events are routed with `keyBy` on the primary key so that all
+events for a key reach one subtask in order. This ordering is required for
+correctness, but it also means a single key's throughput is capped by one
+subtask.
+
+If the key distribution is skewed, that subtask gates the whole pipeline while
+its peers idle. The sink collapses repeated writes to the same key within a
+checkpoint, which absorbs update-heavy skew, but not sheer volume concentrated
+on one key.
+
+Where the natural key is known to be skewed, prefer a composite primary key
+including a higher-cardinality column, or salt the key:
+
+```sql
+PRIMARY KEY (tenant_id, event_id) NOT ENFORCED
+```
+
+Two-level hashing is not offered: it would break in-key ordering, which the
+sink's flush sequencing depends on.
 
 ## Current limitations
 
 | Statement | Status |
 |---|---|
 | `INSERT INTO` (append) | ✅ |
 | `INSERT OVERWRITE` | ✅ |
-| `UPDATE` | ❌ — not implemented |
-| `DELETE` | ❌ — in progress (see issue #63 / #74) |
-| Primary key / upsert | ❌ — PK declaration and CDC changelog not yet supported |
+| Primary key / upsert | ✅ |
+| `DELETE` (via CDC changelog) | ✅ |
+| `UPDATE` (standalone statement) | ❌ — not implemented |
 
-> The sink currently declares insert-only changelog mode. CDC `UPDATE` / `DELETE`
-> support is tracked in the connector roadmap.
+> Standalone `UPDATE` and `DELETE` SQL statements are not supported; row-level
+> changes are applied through a CDC changelog stream into a table declaring a
+> primary key.