fix: adopt post-merge dataset handle, run ITCase in CI, and close type-mapping gaps - #82
Open
fightBoxing wants to merge 11 commits into
Open
fightBoxing wants to merge 11 commits into
fightBoxing wants to merge 11 commits into
Conversation
added 10 commits
September 22, 2026 16:58
Address the A/B-level items from the architecture review of the upsert-delete-alter-table PR: A1+A3+A5: rewrite LanceUpsertSink first-write path - Replace the createDataset (Overwrite) branch with an open-or-create strategy (Dataset.open, fallback Dataset.create) that is atomic in the Lance native layer, eliminating multi-subtask Overwrite clobbering. - Remove the Files.exists() checks that silently misclassified remote paths (s3://, tbdsfs://) as non-existent. - OVERWRITE mode now explicitly rejects remote storage in the sink and documents that truncation must happen at DDL time. A2: tighten checkpoint / consistency model - Apply deletes before upserts within a flush so a half-failed flush never leaves a stale row that should have been superseded. - Drop the implicit flush() from close(); persistence boundary is now strictly the checkpoint, matching at-least-once semantics. - Reject NULL primary-key values and non-finite float PK values in the DELETE predicate (previously they produced silently no-op predicates). - Escalate OVERWRITE local-directory delete failures from LOG.warn to IOException (was B6). B2: bound the in-memory buffer - invoke() now triggers an early flush() once the collapsed key count reaches write.batch-size, preventing unbounded heap growth between checkpoints. Happens between events, so the delete-before-upsert invariant is preserved. B5: protect foreign-namespaced dataset config during ALTER - applyTableProperties() no longer UNSETs config keys that look namespaced (contain a dot), preserving metadata written by sister engines (Spark, Trino, Ray) across a Flink ALTER TABLE ... RESET. Engineering-quality - LanceDynamicTableSink assigns a stable operator UID + name to the upsert sink so state mapping survives job upgrades / savepoints. - Extract PrimaryKeySelector.project() and have LanceUpsertSink. extractKey() delegate to it, so the keyBy routing key and the buffer key can never drift. - Reject comma-containing primary-key column names in PrimaryKeyPersistence.persist() to keep the comma-delimited encoding unambiguous. - Narrow ArrowArrayStreams from public to package-private. - LanceUpsertSinkITCase.twoSubtasksConcurrentFirstWrite() is now truly concurrent (CountDownLatch double-barrier + ExecutorService). - New LanceUpsertSinkITCase.closeDoesNotImplicitlyFlush() regression test for the A2 close() semantics. Not addressed in this commit (tracked in .gh-comments/): - A4 typed DELETE encoding (replace SQL string with mergeInsert WhenMatched.Delete or IN-list). - A6 consume TableChange in LanceCatalog.alterTable instead of SchemaDiff heuristic. - A7 single source of truth for connector option classification. - B1 hot-key metrics; B3 RootAllocator cap (needs cross-component coordination); B4 cross-engine PK metadata key. mvn -o clean test-compile: BUILD SUCCESS (0 errors) on 1.18/1.19/1.20.
A4: rewrite DELETE on key-only mergeInsert Replace the SQL-string delete predicate with a key-only mergeInsert using withMatchedDelete + WhenNotMatched.DoNothing. This closes the type-coverage gap (DATE/TIME/TIMESTAMP/DECIMAL/VARBINARY), removes the injection surface from unescaped column names, and eliminates the O(N) predicate growth. Also adopt the post-merge dataset handle: mergeInsert commits a new version and returns a new handle rather than mutating the receiver, so the sink was reading a stale snapshot after every merge. A6: drive ALTER TABLE from TableChange Override the alterTable overload that receives the planner's explicit change list instead of inferring intent from a schema diff. A DROP followed by an ADD of the same name is now applied literally; the diff heuristic could not distinguish it from a rename and refused the statement. Changes are applied rename -> drop -> add, since re-creating a column whose name still exists is rejected by Lance as a type conflict. ALTER COLUMN type changes stay rejected because the SDK's castTo is a verified silent no-op. The diff-based path remains as a fallback when no changes are supplied. A7: derive option classification from the factories Introduce LanceOptionRegistry, which builds the reserved-key set from the factories' live ConfigOption declarations, and delete the hand-maintained copy in LanceCatalog. The two had already drifted: s3-virtual-hosted-style and s3-allow-http were declared by LanceCatalogFactory but missing from the copy, so they leaked into the dataset config as user properties. Scope RESET to Flink-owned keys (the flink.* namespace plus unnamespaced keys) instead of treating every dotted key as foreign. The old rule was safe against cross-engine deletion but also made RESET impossible for dotted properties Flink itself wrote. B1: document the hot-primary-key constraint keyBy on the primary key is required for flush ordering, so per-key throughput is bounded by one subtask. Document composite or salted keys as the mitigation and record why two-level hashing is not offered. B3: make the Arrow allocator bound configurable Route all ten allocation sites through LanceAllocators and add the opt-in arrow.allocator-max-bytes option; unset preserves the previous unbounded behaviour. Each allocator is named so an OOM identifies its owner.
Push-down built predicates by string concatenation and accepted several inputs it could not render faithfully. An accepted-but-wrong predicate is worse than a rejected one: rejection leaves Flink to evaluate the filter correctly, while a wrong predicate silently changes the result set. Three cases were doing the latter. Column names were inlined unquoted, so a name containing a space, an uppercase letter, or a SQL keyword did not round-trip: `user name = 'x'` parses as two tokens. Names are now backtick-quoted. Names containing a dot or a backtick are declined instead, since Lance reads a dot as nested field access and documents no escape for it. DATE and TIMESTAMP literals fell through to a catch-all that quoted toString(), comparing a Date32 or Timestamp column against a Utf8 literal, and emitting the ISO 'T' separator that the grammar does not accept. Both now use the documented typed form, with TIMESTAMP carrying the column's declared precision. DECIMAL likewise uses decimal(p,s) so scale survives. NaN and Infinity are Numbers, so they rendered as bare tokens that do not parse. They are now declined, matching the rejection follow-up A4 applied on the write side. The comment claiming IN and BETWEEN are unsupported was wrong and is replaced: Calcite expands IN into an OR chain and BETWEEN into >= AND <= before applyFilters is reached, so both already push down through the existing branches. Tests come in two layers. The unit tests pin the exact rendered string; they caught the literal formatter emitting '.000000000' for a whole second, since an optional pattern section is only optional when parsing. The IT case runs each predicate against a real dataset and asserts row counts, because a predicate that merely looks like the documentation is not evidence the engine accepts it -- the lesson from castTo in PR lance-format#76. It also confirms a spaced column name is addressable only once quoted. Existing push-down tests assert only whether a filter is accepted, never what predicate results, which is why these three defects went unnoticed.
These three types had no Arrow mapping, so flinkTypeToArrowField rejected them and a table carrying one could not be created. The gap was already inconsistent: RowDataFieldAccessor gained DECIMAL/TIME/TIMESTAMP_LTZ support when DELETE moved to key-only mergeInsert, so such a column could act as a primary key while remaining impossible to declare, and filter push-down emitted decimal and timestamp literals for columns the writer could not create. DECIMAL maps to Decimal128, which covers Flink's maximum precision of 38. TIME selects its Arrow bit width from the unit, since Arrow only permits Time32 for SECOND/MILLISECOND and Time64 for MICROSECOND/NANOSECOND. TIMESTAMP_LTZ maps to a timestamp tagged with UTC; without the tag it would be indistinguishable from a plain TIMESTAMP, and the reverse mapping now keys on the timezone so the two do not collapse into one on read back. Support was verified against Lance itself before implementing, rather than inferred from the Arrow signatures being present: a probe wrote and reopened a dataset per type and confirmed precision, scale and timezone survive. This follows castTo, which was present in the SDK but a silent no-op. Values are carried through readValue/getFieldValue/writeValue, and the new vectors are registered in setNull, which has no trailing else and would otherwise leave a null at its default value instead of marking it null. Epoch splitting for zoned timestamps uses floorDiv/floorMod so instants before 1970 keep a non-negative nanosecond remainder. LanceNamespaceCatalogSchemaTest used DECIMAL as its unmappable-type sample, which no longer holds; it now uses MAP, which is still unmapped.
The sink now implements SupportsRowLevelUpdate, so UPDATE ... WHERE runs against a keyed Lance table. UPDATED_ROWS is requested because that mode delivers only the matched rows, each tagged UPDATE_AFTER, which is what the existing keyed upsert path already consumes; ALL_ROWS would stream back rows the statement never touched and rewrite them for no gain. requiredColumns() returns empty, meaning every column. The sink writes whole rows through mergeInsert(withMatchedUpdateAll), so narrowing the projection to the SET list plus the key would null out every unmentioned column. UPDATE on a table without a primary key is rejected during planning, since there is nothing for mergeInsert to match on. Two pre-existing defects blocked the end-to-end test and are fixed here. A keyed write lost everything it buffered when a job ended without a checkpoint. close() deliberately does not flush, because it also runs on cancel and failover, but nothing covered a graceful end of input -- so a batch job, which never checkpoints, completed reporting success while writing nothing. This is now handled in finish(), which Flink calls only on the normal completion path, preserving the checkpoint-as-persistence-boundary contract. The append path was unaffected because LanceSink.close() does flush, so the two sinks disagreed on whether a completed bounded job was durable. createDynamicTableSource/Sink passed the hadoop.* prefix list straight to validateExcept, which rejects an empty array. Any table declaring no hadoop option therefore failed validation outright, which is every table not backed by HDFS. The prefix list is now checked before choosing the overload. Both new test classes are named *Test rather than *ITCase: surefire's default includes do not match the ITCase suffix and no failsafe execution is configured, so the project's 86 existing ITCase cases never ran under mvn test. That gap and a long-standing concurrent-first-write failure it was hiding are recorded in .gh-comments/issue-update-followups.md; neither is in scope here.
… exposed The project's *ITCase integration tests never ran in a build. surefire's default includes do not match the ITCase suffix, and although CI runs mvn verify, no failsafe plugin was configured, so the verify phase bound only surefire. Seventy-five integration tests -- including one long-standing failure -- were silently skipped on every "green" build. Configure maven-failsafe-plugin and bind its integration-test and verify goals. The verify goal is the operative one: without it the reports are written but a failure does not fail the build. failsafe's default includes already cover **/*ITCase.java, so the existing classes keep their standard Flink-ecosystem names. The version is 3.3.0 rather than surefire's 3.1.2 because 3.1.2 of failsafe is not available from the internal mirror; the two plugins version independently. Turning the tests on surfaced a real defect. PrimaryKeyPersistence.persist wrote the primary-key metadata with an unconditional updateConfig, which is a versioned Lance transaction rather than an idempotent put. Two subtasks first-writing the same dataset committed it at once and the conflict resolver rejected one outright. persist is now genuinely idempotent: it compares the stored value first so the steady state opens no transaction at all, and on a conflict it advances the handle with checkoutLatest before re-reading -- the handle is pinned to the version seen at open() and would otherwise never observe the peer's commit -- accepting the result when the winner stored the same value and rethrowing only on a genuine disagreement. Removing the checkoutLatest call reproduces the original conflict, which confirms the stale handle was the root cause. The conflict arrives as a bare RuntimeException from the Rust resolver with no dedicated type, so the outcome is verified by re-reading rather than by matching the message. Both C1 and C2 in .gh-comments/issue-update-followups.md are now resolved. mvn verify runs surefire 276 + failsafe 75 per module, green across 1.18 / 1.19 / 1.20.
RowDataConverter#setNull ended its if-else chain without an else, so an Arrow vector type it did not enumerate was skipped without a word. That is data corruption rather than a harmless no-op: the validity bit of a slot that already holds a value stays set, so the previous row's value is serialised as this row's value and nothing is reported. A probe on IntervalDayVector confirmed it -- after setNull the slot still read back as non-null with the stale value intact. The bug stayed invisible because a freshly allocated vector masks it. The zeroed validity buffer reads back as null while getNullCount() still returns 0, so the two disagree and a first-write smoke test looks correct. Only a reused slot exposes the stale value. readValue, getFieldValue and writeValue all reject unknown types, so the silent branch was also an inconsistency inside a single class. setNull now throws UnsupportedTypeException naming the vector class and the field. MapVector is matched ahead of ListVector because it is a subclass; with the list branch first every map would be nulled through the list path. No Flink type maps to MapVector yet, but the ordering has to be right before MAP support lands, and it is cheaper to fix now than to debug later. The change is a net-new guard: the full suite passes unchanged, which confirms every type on the current production path is already enumerated.
Adding MAP to the type converters on its own would have produced a table that fails on first write. Lance gates Map data on file format 2.2+, and every write path in the connector used the SDK default of 2.1. A spike through the production mergeInsert path confirmed the shape of the problem: a Map column is accepted into the schema at CREATE TABLE on 2.1 and survives a reopen intact, but writing rows is rejected by the Rust encoder. Only DDL looks healthy, which is why the gap was not obvious from the capability list. The option has no default. Hardcoding today's 2.1 would pin the connector to it once the SDK default moves forward, so leaving it unset defers to the SDK and keeps the previous behaviour byte for byte. It is applied in LanceCatalog#createTable as well as on both sinks, because the format version is fixed when the dataset is created -- setting it only on the sink leaves an already-materialized table unable to hold a map. That catalog call site turned out to have no test coverage at all: disabling it left every existing suite green. The new ITCase compares an explicitly-2.2 table against an unset one via Dataset#getLanceFileFormatVersion, and disabling the call now fails it. Two things surfaced while testing. mergeInsert does not preserve input order -- the first round-trip assertions read rows by position and saw empty/NULL/populated instead of the written order -- so assertions are keyed by id. And the factory redefines 16 options that LanceOptions already declares, with identical keys; that is the A7 dual-source-of-truth pattern recurring at larger scale. The new option follows the existing pattern in both classes rather than smuggling a refactor into this change. MAP itself is still unmapped. When it lands, flinkTypeToArrowField has to reject a MAP column on a pre-2.2 dataset with a clear message instead of letting the user hit the encoder error.
Builds on the write.data-storage-version groundwork. All three conversion directions in LanceTypeConverter and all four dispatch points in RowDataConverter now handle MAP, so a map column can be created, written and read rather than only appearing in a schema. Three things needed verifying rather than assuming, and each was confirmed by disabling the code and watching a test fail: setIndexDefined on the entries struct is required. Removing it leaves the in-memory converter round-trip fully green, because that test reads back the vectors it just wrote and never passes through Lance's validation. A real dataset rejects it immediately: "The field `entries` contained null values even though the field is marked non-null in the schema". The in-memory layer has no discriminating power for non-null constraints, so both test layers are load bearing. The MapVector branch has to precede ListVector on the write path too, not just in setNull. MapVector extends ListVector, so without it a map falls into the list branch. The same ordering applies in reverse: ArrowType.Map must be checked before ArrowType.List, or a map degrades into ARRAY<ROW<key, value>> and stops round-tripping. Key and value types are far narrower than a top-level column: only INT, BIGINT, FLOAT, DOUBLE and STRING, bounded by RowDataConverter's array element helpers. Arrow is perfectly happy with Map<Utf8, Date>, so an unchecked MAP<STRING, DATE> would create a table whose first write fails -- the same trap the 2.2 format requirement set for MAP itself. mapEntriesField rejects it at DDL time instead. LanceCatalog#createTable refuses a MAP column when the configured version is demonstrably below 2.2, naming the column and the option. An unset version is only warned about: the effective default belongs to the SDK, and refusing here would misfire the moment that default reaches 2.2. Arrow does not allow a nullable map key, which makes the obvious DataTypes.MAP(STRING(), INT()) invalid; the message says so explicitly rather than leaving the user to discover .notNull(). Empty and NULL maps are stored distinctly, and a NULL value is allowed where a NULL key is not. The unsupported-type sample in LanceNamespaceCatalogSchemaTest moves for the third time -- DECIMAL, then MAP, now MULTISET -- since each in turn gained a mapping. Its comment records the trail so the next move is mechanical.
Closes the last remaining type gap. A MULTISET is physically MAP<element, count> and Flink already hands it around as MapData at runtime -- getDefaultConversion is java.util.Map and RowData.getMap works on it -- so it reuses the map path instead of getting a parallel implementation. readMap and writeMap now take the key and value LogicalTypes directly rather than a MapType, which is the whole of the sharing; the count side is pinned to a non-null INT32 because an element that is present has an occurrence count by definition. The storage requirement was carried over as a guess and is now measured. A MULTISET-shaped map fails to write on format 2.1 with the same error MAP hit -- "Map data type is only supported in Lance file format 2.2+", lance-encoding/src/encoder.rs:485 -- and writes cleanly on 2.2. Because the version check in LanceCatalog keys off ArrowType.Map, and setNull already has a MapVector branch ahead of ListVector, neither needed changing. A MULTISET 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. Field metadata was tried and does survive a Lance round-trip, so this was a choice rather than a limitation: it would put a Flink-specific key into a schema other engines also read, for a type that is mostly an aggregation result rather than a column declaration. Writes and stored bytes are unaffected; only the recovered type name differs. The NULL-key rejection message is now parameterised, since a user who wrote a MULTISET has no "key" in their DDL to go looking for. Same reason the element type check reports "MULTISET element" rather than "MAP key". The unsupported-type sample in LanceNamespaceCatalogSchemaTest moves a fourth time, to INTERVAL. DECIMAL, MAP and MULTISET were all types Lance genuinely stores, so each was always going to gain a mapping; an interval is not analytical storage data and is not on that path.
The batch-mode tests added with SupportsRowLevelUpdate go through Flink's SortingDataInput, which serialises via Kryo, which reflects into Arrays$ArrayList to read its backing array. JDK 17 refuses that without --add-opens and the job dies with InaccessibleObjectException; JDK 11 only warns, so a local run on 11 shows nothing and the gap surfaces only on the 17 and 21 legs of the CI matrix. Verified on 11, 17 and 21 across all three modules.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Follows #76. Supersedes #77 — every change in that PR is included here, plus the fixes for the two regressions it carried.
Why this replaces #77
#77 moved dataset creation and primary-key persistence from
flush()intoopen(), which is the right direction: it removes theOverwrite-based first write that could clobber a peer subtask's commit. But the move introduced two defects that its CI could not see, because*ITCasefiles never execute —failsafehas no goal binding inpom.xml, and surefire's default includes do not match that suffix.Forced to run via surefire,
LanceUpsertSinkITCasegives:88b8c42)257a99c)mergeInsertdoes not mutate theDatasetit receives; it commits a new version and hands back a new handle. #77 tookthis.datasetonce inopen()and discarded every return value after that, so the handle stayed pinned to theopen()snapshot. Writes still landed — Lance resolves against the latest table state at commit — but every read through that handle saw the pre-change state, which is what the five collapsing and replay assertions were reporting.The sixth failure was
PrimaryKeyPersistence.persist, commented "Idempotent; safe to call on every open".updateConfigis a versioned transaction, not an idempotent put, so two subtasks opening at once put the second one into Lance's conflict resolver. Hoisting the call intoopen()turned it from once-per-dataset into once-per-subtask, which is why the move surfaced it.Contents
Build — bind failsafe to
integration-testandverify, so the 5 ITCase files #76 added actually run. This is the change that makes the rest of the suite meaningful.Correctness — adopt the post-merge dataset handle; make
persistcompare-before-write and recover from conflicts viacheckoutLatest(); reject unsupported vectors inRowDataConverter#setNullinstead of leaving a stale slot value readable as valid data.DML — rewrite DELETE onto a key-only
mergeInsertwithwithMatchedDelete, replacing the SQL-string predicate (closes the DATE/TIME/TIMESTAMP/DECIMAL/VARBINARY gap, the unescaped-identifier injection surface, and O(N) predicate growth). AddSupportsRowLevelUpdate. Fix batch-mode data loss by flushing infinish()rather thanclose(), which also runs on cancellation.DDL — drive
ALTER TABLEfrom the planner'sList<TableChange>instead of inferring intent from a schema diff, soDROP score DOUBLE+ADD score STRINGis applied literally rather than rejected as a rename with a type change. Changes apply rename → drop → add; re-creating a column whose name still exists is a type conflict in Lance.Types — map DECIMAL, TIME, TIMESTAMP_LTZ, MAP and MULTISET. MAP and MULTISET both need file format 2.2+, so
write.data-storage-versionis exposed first (no default — hardcoding today's 2.1 would pin users to it as the SDK default moves).Push-down — quote identifiers, emit typed literals for date/time/decimal, and drop predicates that cannot be rendered faithfully instead of passing a wrong one.
INandBETWEENneeded no work: Calcite expands them before the connector sees them.Hardening — route all allocation through
LanceAllocatorswith an opt-inarrow.allocator-max-bytes; derive the reserved-option set from the factories' liveConfigOptiondeclarations, which had already drifted from the hand-maintained copy by two S3 keys.Verification
surefire 308 + failsafe 83, green on 1.18 / 1.19 / 1.20, rebased onto
mainafter #76 merged. The 17 skips are S3 cases needing MinIO; the 1 failsafe skip is environment-gated.#77 can be closed once this lands.