Repository navigation
feat(datafusion): support the paimon_incremental_query table function - #1042
jackylee-ch wants to merge 5 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.
2 participants
Closes #506.
Adds the
paimon_incremental_querytable function so DataFusion SQL can read asnapshot-id range
(start_exclusive, end_inclusive]as a batch:The optional mode is
auto/delta/changelog/diff(defaultauto); a$audit_logsuffix prependsrowkind. Names resolve as inFROM, andSET 'paimon.*'options apply as in an ordinary read, except those choosinganother range, which are rejected. Planning stays in core, matching Java's
incremental-between.Core's
to_incremental_arrownow decodes liketo_arrow, so data-evolutionBLOB/VECTOR files merge into rows and primary-key BLOBs honor
blob-as-descriptor. A data-evolution range reads a column it did not write asNULL, NOT NULL ones included; Java fails if two updates of the same rows leave
a NOT NULL column unwritten.