[XiaoHongbo-Hope]

Purpose

Add a lazy binary as-of join for independently sampled multimodal streams.

The API follows pandas merge_asof, Polars join_asof, and DuckDB ASOF JOIN: preserve left rows, match at most one right row within by, and support backward, forward, or nearest matching with optional tolerance.

steps = join_asof(
    actions.scan(),
    images.scan().select("image"),
    on="event_time",
    by="episode_id",
    direction="nearest",
    tolerance=20,
).join_asof(
    states.scan().select("state"),
    direction="backward",
    tolerance=50,
)

Chained joins remain lazy. Paimon pins snapshots, plans compact metadata with _ROW_ID, fetches matched payloads by batch, and keeps BLOBs as descriptors.

Resampling, interpolation, windows, and streaming/watermark semantics are out of scope.

Tests

94 passed, 6 subtests passed

[JingsongLi]

Four reproducible edge cases found in f854204; details inline.

[JingsongLi]

[P2] Preserve the physical row ID before flattening projections

A valid nested projection such as left.scan().select(['_ROW.ID', 't']) flattens the payload field to _ROW_ID, so the actual internal row ID is renamed to _ROW_ID__0. After _project_fetch, this lookup therefore builds the position map from payload values. With left rows (t=10, _ROW.ID=1) and (t=20, _ROW.ID=0) in one group, and right values 10 and 20 at the corresponding timestamps, an exact join returns (t=20, value=10) and (t=10, value=20): the left rows are silently paired with the wrong right payloads. Please extract the physical row IDs before flattening, or track the internal field by its original identity rather than its projected name.

[JingsongLi]

[P2] Exclude join keys by their original projection paths

These names are already flattened, while by and on refer to original table columns. For a right table containing the grouping column payload_value and a separate payload: struct<value: int>, using by='payload_value' with right.scan().select('payload.value') drops the selected payload because it also flattens to payload_value. I reproduced the payload disappearing entirely; selecting ['payload.value', 'payload_value'] instead retains the renamed grouping value 1 rather than the requested payload value 7. Please decide which fields are join keys from their original paths or field identities, then retain the corresponding projected output fields.

[JingsongLi]

[P2] Preserve an input pinned before its first snapshot

The empty-input check happens after _metadata_builders() has already planned a scan. If an alignment is constructed with a nonempty left table and a newly created empty right table, then the right table receives its first append before the alignment is consumed, to_list() raises Temporal input changed from snapshot None to 1 during planning. The same alignment returns null matches before that append. This breaks the documented snapshot-pinned behavior for an initially empty optional stream. Please bypass physical snapshot planning for _temporal_empty while preserving authorization and the effective schema; _RowIdFetcher also plans unconditionally and needs the same treatment.

[JingsongLi]

[P2] Bound decoding memory before materializing a cached row group

list(...) retains the entire decoded row group before the cache checks its 64 MiB limit, so a small output batch no longer bounds payload memory. In an end-to-end join using a normally written right table with 4,096 BYTES values of 32 KiB each, read.batch-size=32 and output batch_size=1, peak Arrow allocation was about 128 MiB with this cache versus 3 MiB with the cache disabled. The oversized entry is immediately evicted, and each output batch repeats the full materialization. Larger payload groups can therefore exhaust memory despite the cache limit. Please use bounded incremental caching, or keep groups that cannot fit on a streaming read path.

[JingsongLi]

+1