feat(datafusion): support the paimon_incremental_query table function

[jackylee-ch]

Closes #506.

Adds the paimon_incremental_query table function so DataFusion SQL can read a
snapshot-id range (start_exclusive, end_inclusive] as a batch:

SELECT * FROM paimon_incremental_query('db.t', 0, 5);
SELECT * FROM paimon_incremental_query('db.t', 0, 5, 'delta');
SELECT * FROM paimon_incremental_query('db.t$audit_log', 0, 5, 'changelog');

The optional mode arg is auto/delta/changelog/diff (default auto); a
$audit_log suffix prepends the rowkind column. Planning stays in paimon
core: this picks the core planner matching Java's incremental-between for
the table and streams its output, reserving reader memory from the session
pool.

It also makes core's to_incremental_arrow decode files as to_arrow does:
data-evolution BLOB/VECTOR column files merge into rows by row id, and
primary-key BLOB columns return payloads, not stored descriptors.

[JingsongLi]

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.

[JingsongLi]

[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.

[jackylee-ch]

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.

[JingsongLi]

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.

[JingsongLi]

[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.

[JingsongLi]

[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.

[JingsongLi]

[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.