Repository navigation
feat(datafusion): support the paimon_incremental_query table function - #1042
jackylee-ch wants to merge 23 commits into
Conversation
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS at 4fad44b. Exposing the core incremental planner through SQL addresses #506 with useful end-to-end scope. The new execution adapter, however, materializes the complete range outside the configured DataFusion memory budget.
Validation: all 7 original real SQL write/read tests passed. A separate PK input-changelog probe verified delete/insert/update row kinds, persisted sequence numbers, filtering/projection, and Diff across two real snapshots. The automatic merge with main b262b5f is clean. Additional real SQL probes reproduce the resource issue below: a zero-byte execution memory pool rejects an ordinary scan but permits the incremental query, and selecting only id with LIMIT 1 over 12 snapshots retains the 24 MiB unselected payload range (about 27.4 MB peak allocations during incremental execution). No additional correctness/schema/auth finding was established.
| .map_err(to_datafusion_error)?; | ||
| let mut batches = Vec::new(); | ||
| while let Some(batch) = stream.try_next().await.map_err(to_datafusion_error)? { | ||
| batches.push(batch); |
There was a problem hiding this comment.
[P1] Keep incremental reads streaming and honor execution resources
This drains the core stream into a Vec before the first output batch; projection happens only afterwards and execute ignores its TaskContext. Delta/Changelog core already yields bounded batches, but the SQL adapter retains all decoded columns for the entire snapshot range, even for SELECT id ... LIMIT 1. I reproduced roughly 27.4 MB peak incremental allocations from a 24 MiB payload range that the query never selects. There is no cap on the range, so normal CDC/backfill ranges spanning multiple 256 MiB target files can exhaust the process before producing any row. The configured DataFusion budget does not protect it: with GreedyMemoryPool(0), ordinary SELECT id FROM table returns ResourcesExhausted, while this function reads all three rows successfully. Please forward the core stream lazily, project batches as they are emitted (and push a supported read projection where possible), and attach the executing TaskContext's reader resources as the ordinary Paimon scan does. Retained output, if any is necessary, must be accounted by its consumer.
There was a problem hiding this comment.
Fixed in 9568181: IncrementalQueryExec now forwards core's stream batch by batch, attaches reader_resources(&context, partition) like the ordinary scan, and pushes the projection via with_read_type (Diff still decodes full rows). Core's to_incremental_arrow read files raw, splitting data-evolution BLOB/VECTOR columns into separate rows and returning primary-key BLOB descriptors; it now decodes them as to_arrow does. A data-evolution range is planned as one batch, as in Java's incremental-between. Tests cover the zero-byte pool, SELECT id over 1 MiB payloads and these reads.
`IncrementalQueryExec` collected the whole snapshot range into a `Vec` before emitting a batch, decoded every column even for a narrow projection, and ignored the executing `TaskContext`, so the DataFusion memory pool never saw the reader's working memory. Forward core's batch stream lazily, push the projection into the read (`$audit_log` columns are mapped by position, one read field per projected column), attach the task's reader resources, and rebuild each batch against the declared (projected) schema, as the ordinary scan does. Core's `to_incremental_arrow` decoded each planned file raw, so a data-evolution BLOB or VECTOR column file came back as rows of its own and a primary-key BLOB column as its stored descriptor. Decode those files as `to_arrow` does, read limit included, and plan a data-evolution Delta range as one batch, as Java's `incremental-between` scan does, so a column update merges into the rows it updates when both are in the range. Document `paimon_incremental_query` in the SQL guide.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS at 9568181. The incremental SQL entry point addresses #506 and the previous eager-collection finding is fixed: batches are streamed, projection is pushed into reads, and reader resources come from the executing TaskContext.
Validation: all 14 incremental SQL tests, 42 core incremental tests, and 8 partition-filter tests passed, including the zero-byte pool and stream-before-last-file checks. Three independent tests using real SQL-written Parquet/BLOB data reproduce the remaining issues below: explicit catalog selection reads a different warehouse; a range containing two data-evolution partial updates rejects a missing NOT NULL base column; and the session BLOB-descriptor option is ignored. The merge with main 7dd674b is clean. These are local production-path probes, not live cloud-store deployment tests.
| }; | ||
| let identifier = parse_table_identifier(FUNCTION_NAME, &base_name, &self.default_database)?; | ||
|
|
||
| let catalog = Arc::clone(&self.catalog); |
There was a problem hiding this comment.
[P1] Honor the explicit catalog instead of silently reading another warehouse
The UDTF captures one catalog, and each register_catalog replaces the same globally registered function. Meanwhile parse_table_identifier discards the first component of a three-part name. I reproduced with two warehouses registered as cat_a and cat_b, both containing default.t: ordinary fully qualified SELECT returns [1] from cat_a and [2] from cat_b, but paimon_incremental_query('cat_a.default.t', 0, 1) and the cat_b call both return [2]. A supported, explicitly qualified table name thus silently reads the wrong data source. Please resolve the catalog component against the session's catalog registry, or at least reject a component that does not match the bound catalog rather than discarding it. Add the two-warehouse regression; successful single-catalog reads do not cover this.
There was a problem hiding this comment.
Fixed in f8016e1: the function now resolves the name in the calling session as a FROM clause does (t and db.t take the current catalog and database, cat.db.t names the catalog) and looks the catalog up in the session's catalog list; an unknown or non-Paimon catalog is an error. Added the two-warehouse regression, checked against ordinary SELECT for 1-, 2- and 3-part names, after USE, and with a database named like the other catalog. vector_search, full_text_search, hybrid_search and blob_view still bind the catalog registered last; this PR leaves them unchanged.
| .resolving_primary_key_blobs(&core_options, |read| read.read_raw(&data_splits)); | ||
| } | ||
| if core_options.data_evolution_enabled() { | ||
| return self.read_with_evolution(&data_splits, &core_options); |
There was a problem hiding this comment.
[P2] Allow absent base columns when merging an update-only incremental range
The new ordinary evolution-reader path also applies its full-table missing-provider check to partial incremental data. With a valid DE table (id INT NOT NULL, name STRING, value INT), snapshot 1 inserts two rows, snapshot 2 MERGE-updates name, and snapshot 3 MERGE-updates value. I reproduced paimon_incremental_query('default.de', 1, 3) failing with Cannot read non-nullable field 'id' without a provider. The range legitimately contains the two partial column files but no original id file. The single-update range (1,2] succeeds and NULL-fills id, as this PR's documentation promises for columns outside the range; adding the second partial file switches into the merge path and fails. Please allow the incremental evolution merge to NULL-fill fields absent from the selected range, including logically NOT NULL fields, while retaining the ordinary full-table integrity check. Cover the update-only range in auto/delta and plain/audit reads; changing the DataFusion Arrow schema alone will not fix the reader's validation.
There was a problem hiding this comment.
Fixed in f8016e1: for a split of change events (DataSplit::is_streaming), the merge reader reads a field no file of the group wrote as NULL, NOT NULL and nested ones included; snapshot reads keep the check. This covers auto/delta and plain/audit here and core's other change-event reads (plan_combined_delta + to_arrow, incremental-between-timestamp, stream follow-ups). Java rejects this case; the SQL guide says so. A merge group of only .blob/.vector. files (e.g. a range that only rewrites a VECTOR column) still fails with "requires at least one normal data file", as on main.
| /// projects its output. The stream is `'static`, so it outlives the | ||
| /// transient read builder it is built from. | ||
| async fn open(&self, resources: ResourceContext) -> DFResult<ArrowRecordBatchStream> { | ||
| let mut read_builder = self.table.new_read_builder(); |
There was a problem hiding this comment.
[P2] Apply session BLOB read options before opening the range
The new function retains only the catalog-loaded Table and never receives the SQLContext's dynamic options. This affects actual output semantics, not just snapshot-selection settings: after SET 'paimon.blob-as-descriptor' = 'true', I wrote a primary-key BLOB table and compared the same snapshot through ordinary SELECT and this UDTF. The ordinary scan returns a 177-byte descriptor, while the incremental call returns the three payload bytes AAA. A caller explicitly requesting descriptors still fetches/receives the full payload through the new entry point. Please propagate relevant session read options into the table/read builder, with an explicit policy for selectors that conflict with the function's range, and add the ordinary-vs-incremental descriptor regression.
There was a problem hiding this comment.
Fixed in f8016e1: session options go through the copy_with_time_travel an ordinary read uses, so blob-as-descriptor and read.batch-size apply and descriptors resolve (regression added). Options choosing another range (time-travel selectors, incremental-between*, creation-time filters, non-default scan.mode) are rejected, whether the session or the table read sets them. Immutable overrides apply as in ordinary SELECT: after SET 'paimon.data-evolution.enabled'='false' both return NULL-id fragments. Java's checkImmutability rejects them. Reject in core, or keep the stored value?
`paimon_incremental_query` was bound to the catalog registered last and dropped the catalog part of a three-part name, so a qualified name could read another warehouse. It now resolves the name in the calling session as a FROM clause does and looks the catalog up in the session's catalog list. It also ignored the session's dynamic options. They now apply as they do to an ordinary read of the table, and the table is registered for BLOB descriptor reads. Options that choose another range to read (time-travel selectors, `incremental-between*`, creation-time filters, a non-default `scan.mode`) are rejected, whether the session or the table read sets them, and `$audit_log` keeps the ordinary audit-log rule. A data-evolution range of column updates without the rows they update failed with "Cannot read non-nullable field 'id' without a provider". A split of change events now reads a field no file of its group wrote as NULL, NOT NULL and nested fields included; a snapshot read keeps the check.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS at f8016e1. The prior explicit-catalog, update-only NOT NULL and session BLOB-option issues are fixed on their covered paths. One additional regression remains in the existing core incremental API: a valid VECTOR-only update range now fails after being routed through the ordinary evolution merge reader. The public write/update/commit/scan/read probe fails on this head and passes unchanged on exact base 8d08e48; details inline.
Validation: 358 DataFusion library tests, all 20 incremental SQL tests, 43 core incremental batch tests, 8 partition-filter tests and two independent PK SQL probes passed. The extra PK probes cover changelog row kinds, persisted sequence values, Diff, projection/filter/count and exclusion of later snapshots. Related core evolution, table-reader and option suites, formatting and diff checks passed. Current CI's only failed check is inherited TLS dependency advisories.
| .resolving_primary_key_blobs(&core_options, |read| read.read_raw(&data_splits)); | ||
| } | ||
| if core_options.data_evolution_enabled() { | ||
| return self.read_with_evolution(&data_splits, &core_options); |
There was a problem hiding this comment.
[P2] Keep dedicated-only VECTOR updates readable incrementally
This unconditional DE routing regresses a valid existing to_incremental_arrow call. With row tracking, data evolution and vector.file.format=parquet enabled, the public data-evolution writer updating only embedding intentionally commits a single *.vector.parquet file. Delta (1,2] therefore contains only that streaming file. On this head the new evolution-reader path errors with "Field merge split with .blob/.vector. files requires at least one normal data file"; the identical public-API probe passes on base 8d08e48 and returns the rewritten vectors with the unwritten id column NULL. The ordinary evolution reader's pre-existing dedicated-only limitation does not apply to the baseline raw incremental reader. Preserve dedicated-only change-event decoding, or support its row-range anchor in the merge reader, without borrowing historical files outside the requested range.
There was a problem hiding this comment.
Fixed in 8f26c02: change events merge a row-id group only when its files are column slices of one row range; others, like this VECTOR-only update, are decoded file by file. PTAL.
`to_incremental_arrow` merged every data-evolution row-id group of a range, so a range that rewrote only VECTOR columns, such as recomputed embeddings, failed with "requires at least one normal data file", as did one batch of two BLOB rewrites. Base decoded them file by file. Change events now merge a group only when its files are column slices of one row range: its normal files cover the same rows, its BLOB and VECTOR files lie within them, and each VECTOR field is written in one layout. Other groups are decoded file by file again, and incremental plans keep groups whole, so every projection reads the same rows. A BLOB file read alone honors `blob-as-descriptor`, and change events skip the managed-BLOB preselection, whose row ids can name rows of several files. Merged change events read the rows no BLOB or VECTOR file of the range wrote as NULL, NOT NULL fields included. The BLOB and video readers build the nullable read schema, so a NOT NULL field reads a placeholder, a row an update left unwritten, as NULL.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS at 8f26c02. The prior dedicated-only VECTOR update case is fixed, but a supported transition from dedicated to inline VECTOR storage still loses a newly written embedding in core incremental reads.
Validation so far: 75 core incremental/partition integration tests, all 358 DataFusion library tests and 21 incremental SQL tests passed. An additional public-API probe creates an id+VECTOR table with dedicated storage, writes two rows, persists removal of vector.file.format, and rewrites row 0 to [9,9]. The combined Delta (0,2] read on this head returns only the old [1,0] and [0,1]; the exact same probe passes on merge-base ecbf580 and retains [9,9]. This was independently reviewed. Details inline. Current CI has one inherited dependency-advisory failure (RUSTSEC-2026-0385); that dependency is updated on current main.
| let Some(columns) = info | ||
| .normalized_write_cols | ||
| .as_deref() | ||
| .filter(|_| is_vector_store_file_name(&file.file_name)) |
There was a problem hiding this comment.
[P1] Include inline VECTOR providers in the layout fallback
This filter ignores normal Parquet files that actually carry an inline VECTOR column. Removing vector.file.format is accepted: after an initial dedicated-vector write, a subsequent embedding rewrite produces a newer normal partial-column file covering the same row IDs. With just the old .vector.parquet counted here, the combined group is approved for merging, but build_source_plan prefers the dedicated provider and silently drops the new inline embedding. I reproduced this through public write, persisted RemoveOption, rewrite, combined Delta scan and projected read APIs: the new [9,9] update disappears on this head, while the unchanged test retains it on exact base ecbf580. The ordinary snapshot-reader limitation pre-exists; this PR newly exposes it to to_incremental_arrow by routing that API through the evolution reader. Include actual inline VECTOR providers when deciding whether layouts can merge, and use the existing file-by-file fallback for mixed inline/dedicated groups.
There was a problem hiding this comment.
Fixed in 60fcd30: change events merge a row-id group only if the merge reads each BLOB or VECTOR column where it is stored; otherwise they decode the group file by file. PTAL.
A table can move a VECTOR or BLOB column between dedicated files and inline storage, for example by removing `vector.file.format` or setting `blob-descriptor-field`, and a later rewrite of the same rows writes the column the other way. Change events merged such a row-id group, but the source plan reads each field from one layout: a combined range lost the rewrite's new embedding, and BLOB files written before the change read as NULL. Base returned both. Change events now merge a group only when each BLOB or VECTOR field of the table is stored in one layout, a BLOB field in the layout the table now reads, and no older normal file holds such a field that the newest normal file naming it leaves out: with the write-cols optimization, a file without write columns holds only the fields its stored schema keeps inline. Other groups are decoded file by file, as base decoded them, and a BLOB file read alone is no longer parsed as descriptors or views.
Planning prunes a data-evolution row-id group by the merged stats of its newest column providers. Change events may decode a group file by file, returning each file's own rows, so a filter on a value one file wrote and a newer file of the same rows overwrote dropped the group: a filtered combined range returned none of the rows the unfiltered range returned for that value. A streaming group now also stays when one of its files may match the filter.
Without `$audit_log`, the function now returns the columns of a query on the table, `_ROW_ID` included on a data-evolution table, so a column update returned without the row's other columns can be matched to its row. The `$audit_log` suffix is matched case-insensitively, as a system-table name is in a FROM clause, and the docs state that the table name is split at dots and used as written, as in the other Paimon table functions.
An exact row count of a data-evolution split assumes merged rows. The scan reported it as exact statistics, from which `COUNT(*)` is answered, and the partition-count rewrite answered `COUNT(*)` from the manifests of the snapshot read. Change events may decode a row-id group file by file, so an incremental window over a field stored two ways, or over a compacted range a later update rewrote, counted fewer rows than it returned. The rewrite also counted a whole snapshot for the window of an append table. A streaming split that is not raw convertible now gives no merged row count, and the rewrite skips incremental windows. The docs add that the compaction a write runs after its commit can keep rows from merging.
The previous commit withdrew the merged row count from change events that may decode a row-id group file by file, and with it the early stop of a limit: `LIMIT 1` over a range of a table whose inserts write BLOB or VECTOR files planned every split. Such a group returns at least its merged rows, as each of its files returns its own, so the limit now plans on that lower bound, while the merged row count stays exact. The docs add that a lookup or full-compaction changelog lands in a later COMPACT snapshot, that a row read file by file is matched to the table's rows by `_ROW_ID`, and that row-position selections and chunk shuffle of a data-evolution window count each row once while a group read file by file returns the selected rows of each of its files.
Chunk shuffle looked up a normal anchor file in every data-evolution row-id group, although only deletion vectors, which belong to that file, need one. Change events load no deletion vectors and read a range of BLOB or VECTOR rewrites alone file by file, but `with_chunk_shuffle` on such a range failed at planning, as it did on base. A group now needs an anchor only when it has deletion files; otherwise it is sliced over its row range, as a group with a normal file already was. A limit test now also stops at a group read file by file, and the docs say only the splitting at dots is shared with the other Paimon table functions.
A row of a data-evolution window read file by file carries only the columns its file wrote. The docs now say to join such rows by `_ROW_ID` with the table `VERSION AS OF` the range's end snapshot and to filter after the join, that a Python read has to project `_ROW_ID` for that, and that selected row ids, like positions and chunk sizes, count each row once. The chunk shuffle docstrings say chunks count row ids on data-evolution tables.
UPDATE, DELETE and MERGE INTO find the rows they change by a query on the table through the session, and `vector_search` reads its hits that way. With `SET 'paimon.incremental-between-timestamp'`, the query returns the window's changes instead of the table's rows, so an UPDATE wrote values computed from older versions over newer ones, as it did on base. Since change events of a window holding a compaction are now read file by file, such a window also returns a row once per file that wrote it: a DELETE removed rows whose older version its filter matched, and `vector_search` returned the column update of each hit, its other columns NULL. The writes that refuse a time-travel selector, these three among them, now also refuse the window, and so does `vector_search`.
A delta range returns the rows its APPEND commits wrote, rows a later DELETE removed included. Rows of a data-evolution update read on their own carry the partition columns as well as the updated ones. A row read file by file comes once per file, so the join back to the table's rows uses distinct `_ROW_ID`s. The function reads the range in one partition and applies `WHERE` filters after the read.
…ked up c141ccb refused writes and `vector_search` only while the session set `incremental-between-timestamp`. A table can store the option too (`ALTER TABLE ... SET TBLPROPERTIES`), and then every read of it returns the window's changes, the target lookup of UPDATE, DELETE and MERGE INTO and the reads of `vector_search`, `full_text_search` and `hybrid_search` hits among them. A window holding a compaction is read file by file, so a DELETE removed a row its filter matched only in an older version, UPDATE and MERGE wrote values computed from older versions, `vector_search` returned the column-update row of each hit with its other columns NULL, and the other two returned each hit twice, where base failed or deleted nothing. These now check the table they read, as `paimon_incremental_query` checks its range options, and name the table or branch that stores the window. The writes and `vector_search` read with the session's options applied; the other two searches take no session options, so only a stored window stops them. TRUNCATE TABLE, DROP PARTITION and INSERT OVERWRITE ... PARTITION read no rows, so they no longer refuse a session window.
A row of a data-evolution range read file by file comes once per file that wrote it, so a Python read matches those rows to the table's rows by their distinct `_ROW_ID`s, as sql.md already says for SQL.
A REST read resolves blob views by reading the referenced upstream rows by `_ROW_ID` with the upstream table's own options. If the upstream stores `incremental-between-timestamp`, that read returns the window's changes, and since change events of a window holding a compaction are read file by file, a row comes once per file that wrote it with its BLOB NULL in the files that did not: every view resolved to NULL, where base failed. Resolution now fails with an error naming the upstream table, for every engine and binding that reads through core.
A refusal named a branch read through a `t$branch_b` name as "branch 'b' of table 'db.t$branch_b'". It now names the table without the suffix, as for a branch the session selects. Tests now also pin that writes and `vector_search` check the branch the session or the name selects, and that TRUNCATE TABLE runs while a window is set.
510cff4 refuses to resolve blob views whose upstream table stores `incremental-between-timestamp`. The Blob View section now says so and names both ways out: remove the option from the upstream table, or read the raw references with `blob-view.resolve.enabled=false`.
Closes #506.
Adds the
paimon_incremental_querytable function reading snapshots(start, end]:paimon_incremental_query('t', 0, 5, 'delta').Modes
auto/delta/changelog/diff. Columns matcht(_ROW_IDondata-evolution tables) or
t$audit_log.SET 'paimon.*'options apply; rangeones are refused.
to_incremental_arrownow resolves PK BLOBs liketo_arrow. It andincremental-between-timestampwindow reads now decode data-evolution fileslike
delta(sql.md): a split's column slices merge; a row can come once perfile.
On main, window
COUNT(*)pushdown counted the table,to_arrowfailed oncompacted-then-updated or VECTOR-only ranges and chunk shuffle on BLOB-only ones
(fixed); UPDATE/DELETE/MERGE used old rows and
vector_searchgave NULL ids.They now refuse a window, other searches a stored one, blob views one stored
upstream, even views main resolved.