Multimodal Reads and Row IDs

Filter and project multimodal rows, fetch BLOB payloads, and carry row IDs between processing stages. The examples use the docs table from Multimodal Tables. Examples with frames require a frame table with the named grouping, ordering, and payload columns.

Scan

Use scan() for ordinary table reads. where() accepts SQL-like predicate strings; it does not accept lower-level Predicate objects.

result = (
    docs.scan()
    .where("category = 'lake'")
    .select(["id", "content"])
    .limit(10)
    .to_pandas()
)

Use to_arrow_batch_reader() to consume an ordinary scan without collecting all rows into one Arrow table:

with docs.scan().where("category = 'lake'").to_arrow_batch_reader() as reader:
    for batch in reader:
        consume(batch)

As-of joins

join_asof preserves each left row and matches at most one right row in the same by group. Chain calls to align multiple streams lazily.

from datetime import timedelta
from pypaimon.multimodal import join_asof

aligned = join_asof(
    actions.scan().select(["episode_id", "event_time", "action"]),
    images.scan().where("camera = 'left'").select("image"),
    on="event_time",
    by="episode_id",
    direction="nearest",
    tolerance=timedelta(milliseconds=20),
).join_asof(
    topics.scan().where("topic = '/robot/state'").select("value"),
    direction="backward",
    tolerance=timedelta(milliseconds=50),
)

for batch in aligned.to_arrow_batch_reader(batch_size=128):
    train(batch)

direction is backward, forward, or nearest; tolerance is inclusive and zero permits only exact matches. Nearest ties use the earlier time. For duplicate timestamps, backward uses the last row and forward uses the first. Nearest uses the last row for an exact match; otherwise it uses the backward or forward candidate's rule. Misses return null.

Both top-level and chained join_asof calls accept allow_exact_matches, which defaults to True. Set it to False to exclude all right rows with the same timestamp as the left row. Backward then matches strictly earlier timestamps, forward matches strictly later timestamps, and nearest selects the closest non-equal timestamp, still preferring the earlier time on ties. For example, with a left time of 10 and right times [9, 10], strict backward matches 9. This is useful when aligning an event with a state from strictly before it. Combining allow_exact_matches=False with tolerance=0 (or timedelta(0) for timestamp keys) produces no matches: left rows are preserved with null right values.

Keys must be non-null with matching types. Use right_on for a different right timestamp and suffix for conflicts. Select the right timestamp to compute the match delta.

Inputs are snapshot-pinned (resolved_snapshots). Left rows stream, right join keys stay in memory, and BLOBs remain descriptors.

Use interpolate(left, right, ...) directly or chain aligned.interpolate(...). It supports integer or floating-point scalars and fixed-size lists. Exact timestamps use the exact right row; otherwise it requires surrounding rows in the same by group and never extrapolates. When set, tolerance must include both surrounding rows.

states_at_steps = aligned.interpolate(
    states.scan().select(["joint_position", "velocity"]),
    tolerance=timedelta(milliseconds=50),
)

Use join_window(left, right, ...) directly or chain aligned.join_window(...). Unlike single-table rolling().agg(), it joins each left row to right rows in the same by group and [left - preceding, left + following] time window, then aggregates them. following defaults to zero. mean, min, and max require integer or floating-point scalars; count, first, and last also accept non-numeric values. closed is both, left, right, or neither and refers to the interval endpoints. Nulls are skipped; empty windows return null, except count returns zero. Use (source, operation) pairs to aggregate one source column more than once; {"value": "mean"} remains valid shorthand.

steps_with_imu = aligned.join_window(
    imu.scan().select("acceleration"),
    preceding=timedelta(milliseconds=50),
    following=timedelta(milliseconds=50),
    aggregations={
        "acceleration_mean": ("acceleration", "mean"),
        "acceleration_max": ("acceleration", "max"),
    },
)

Reading BLOB columns

scan().read_blobs(column) bulk-fetches a BLOB column's bytes for the filtered rows using concurrent, same-file coalesced ranged reads. This is much faster than a per-row loop and avoids the slow row-by-row blob resolution on data-evolution tables. It returns (scalar_table, {column: rows}), row-aligned. Scalar BLOB rows are bytes | None; ARRAY<BLOB> rows are None or ordered lists of bytes | None; MAP<K, BLOB> rows are None or ordered lists of (key, bytes | None) pairs. Empty arrays and maps remain empty lists. The scalar table drops the readable BLOB columns, including unrequested ARRAY and MAP BLOBs.

scalar, blobs = (
    docs.scan()
    .where("category = 'lake'")
    .read_blobs("image", parallelism=32)
)
images = blobs["image"]          # list[bytes | None], aligned with scalar rows

where() takes SQL-like strings only, so filter by a list of keys with IN (...):

ids = [1, 2, 3]
in_clause = ", ".join(str(int(value)) for value in ids)
scalar, blobs = docs.scan().where(f"id IN ({in_clause})").read_blobs("image")

where() is a SQL string, so build the IN (...) clause only from trusted, already-escaped ids -- do not interpolate untrusted external input.

Read several BLOB columns at once, including ARRAY<BLOB> and MAP<K, BLOB>:

scalar, blobs = docs.scan().where("category = 'lake'").read_blobs(
    ["image", "audio", "pages", "renditions"]
)
pages = blobs["pages"]  # row-aligned lists of page payloads, or None
renditions = [None if row is None else dict(row) for row in blobs["renditions"]]

For a memory-bounded read (e.g. streaming into a trainer), stream_blobs yields one batch at a time, so peak memory is a single batch rather than the whole result:

for scalar, blobs in docs.scan().where("category = 'lake'").stream_blobs("image"):
    for img in blobs["image"]:
        ...  # feed the trainer, then drop the batch

Notes:

• parallelism ~16–32 is the sweet spot over the public network; go higher on an internal/VPC endpoint.
• Fastest when the matching rows are contiguous (bulk/range reads coalesce into a few large reads); scattered point reads coalesce less.
• Blob reads are available only on scan(), not on the search() queries.

Contiguous windows for PyTorch

Use to_contiguous_window_dataset to build map-style training samples from a single frame table. Each row represents one time step; task text, labels, and other context must already be materialized in that table. The Dataset does not read companion tables. It builds a compact index of group/order values and Paimon row IDs, then fetches requested columns and BLOB payloads on demand from the snapshot recorded in dataset.snapshot_id. Row tracking must be enabled.

pip install 'pypaimon[torch]'

from functools import partial

import torch
from pypaimon.multimodal.window_transforms import images_to_tensor, to_tensor

float32_window = partial(to_tensor, dtype=torch.float32)

windows = (
    frames.scan(tag_name="train-v1")
    .where("split = 'train'")
    .to_contiguous_window_dataset(
        columns=["state", "image", "action"],
        group_key="episode_index",
        order_key="frame_index",
        frame_offsets={"state": [-2, -1, 0], "action": list(range(16))},
        boundary="pad",
        column_transforms={
            "state": float32_window,
            "image": images_to_tensor,
            "action": float32_window,
        },
    )
)

sample = windows[0]
assert sample["action"].shape == (16, action_size)
assert sample["action_is_pad"].shape == (16,)
assert sample["state"].shape == (3, state_size)
assert sample["image"].shape == (1, channels, height, width)

The group and order keys identify the anchor. frame_offsets defines each column's relative frame positions, including history, future, and sparse sampling. Offset order and duplicates are preserved. A selected column omitted from the mapping uses [0], so the image above is read only at the anchor. Without any window definition, every selected column uses [0]. Each column has its own Boolean <column>_is_pad tensor, where True marks an out-of-group position. These generated names must not collide with other output fields.

Anchors start at each group's first row and advance by stride (default one). The stride controls anchor spacing, not the spacing inside a window:

boundary	Behavior at either end of a group
drop (default)	Keep only anchors for which all requested offsets are valid.
pad	Keep scheduled anchors and repeat the nearest endpoint for missing positions.
error	Fail construction if any scheduled anchor has an incomplete window.

pad_values={"action": [0.0, 0.0]} overrides endpoint repetition for that column. Supply values in the same raw representation as a table cell. Padding happens before column transforms, so normalization also affects constant padding. Use the mask when excluding padded actions from a loss.

As an alternative, specify seconds with delta_timestamps, an explicit positive fps, and optional tolerance_s (default 1e-4). For example, delta_timestamps={"state": [-0.1, 0.0]}, fps=10 selects offsets [-1, 0]. Both fps and tolerance_s are only accepted with delta_timestamps. Offsets must align to the regular frame grid within the tolerance; this API does not search actual timestamps or interpolate missing frames. Do not combine delta_timestamps, frame_offsets, and window_size.

column_transforms receive one padded Python list per projected column. This is where applications define tensor dtype and shape or decode BLOB bytes. Even singleton windows retain the time dimension. images_to_tensor returns TCHW images with EXIF orientation applied. Eight-bit pixels become float32 in [0, 1]; use partial(images_to_tensor, return_uint8=True) to keep uint8. Higher-bit-depth images become float32 in their original units. All images in a column window must have the same shape.

The optional adapter receives the transformed sample and can normalize values, rename/combine fields, remove singleton time dimensions, or coordinate random augmentation across cameras and time. Statistics must be supplied explicitly; the Dataset never computes or discovers normalization statistics. For example:

class NormalizeAction:
    def __init__(self, mean, std):
        self.mean = torch.as_tensor(mean, dtype=torch.float32)
        self.std = torch.as_tensor(std, dtype=torch.float32)
        if not torch.isfinite(self.std).all() or not (self.std > 0).all():
            raise ValueError("Action standard deviations must be finite and positive.")

    def __call__(self, sample):
        sample["action"] = (sample["action"] - self.mean) / self.std
        return sample

Pass adapter=NormalizeAction(train_action_mean, train_action_std) when constructing the Dataset. Use training-set statistics, with the same values for validation. Keep callbacks picklable (top-level functions, partial, or callable classes) for multi-worker DataLoader use. Physical reads are coalesced in __getitems__, while samples and transforms retain their requested order and duplicates. Negative indices and slices follow Python sequence semantics.

Rows are sorted by group and order. Order values must be non-null integers increasing by exactly one within each group. Duplicate steps and internal gaps, including gaps introduced by a scan filter, fail validation. Windows never cross groups. A snapshot pin fixes the data being read; use retained tags for long-running training. Sampler progress and augmentation RNG state remain the training application's responsibility.

Existing window_size=16 calls remain supported, with anchor_columns selecting singleton fields and tail="drop"/"pad"/"error" controlling the old forward window. These calls preserve their single is_pad mask and previous shapes. anchor_columns and tail require window_size; do not combine tail and boundary. New applications should use field offsets and per-column masks.

Columns configured by video-frame-field are rejected: a window read would drop the frame_index and other metadata carried by their VideoFrameDescriptor values. Read those columns with to_torch() instead.

Distributed BLOB processing with Ray

For larger jobs, read descriptors with to_ray(), then fetch and process BLOB bytes on Ray workers with map_with_blobs.

Scalar, ARRAY, and MAP BLOB columns are supported. In the callback's blobs dictionary, each column contains row-aligned values: bytes for scalar BLOBs, lists for ARRAY BLOBs, and lists of key-value pairs for MAP BLOBs. Null cells, null elements, empty containers, and element order are preserved. All source BLOB columns are excluded from scalar_batch, including unrequested ones. After Ray transformations, use table.map_with_blobs(...) to supply the source table's BLOB column information even if a transform changes Arrow nested types.

import ray
import pyarrow as pa
import pypaimon.multimodal as pm

ray.init(num_cpus=8)

conn = pm.connect(database="default", options={"warehouse": "file:///tmp/warehouse"})
docs = conn.get_table("docs")

ids = ["a", "b", "c"]
scalar_cols = ["id", "category"]
blob_cols = ["image"]
in_clause = ", ".join(f"'{i}'" for i in ids)


def process_batch(scalar_batch, blobs):
    images = blobs["image"]
    # Decode, infer, write samples, or train here.
    return pa.table({"rows": [scalar_batch.num_rows]})


ds = (
    docs.scan()
    .where(f"id IN ({in_clause})")
    .select(scalar_cols + blob_cols)
    .to_ray()
)

# Optional Ray transforms are fine if BLOB descriptor columns remain.
# ds = ds.filter(lambda row: row["category"] == "lake")

result_ds = docs.map_with_blobs(ds, blob_cols, process_batch)

Notes:

• process_batch must return a small Ray-compatible batch; return an empty pyarrow.Table for side-effect-only jobs.
• Avoid returning raw BLOB bytes, which would materialize payloads in Ray's object store.
• Tune to_ray(...) and map_with_blobs(...) parameters only when needed.

Row IDs

Multimodal tables enable row-tracking.enabled by default, so each row has a Paimon system column named _ROW_ID. Use _ROW_ID as an internal coordination key for retrieval, reranking, inference, and training jobs. Keep user-visible document IDs, object keys, or primary keys in normal columns.

Use with_row_id() to include _ROW_ID in scan or search results. The method appends _ROW_ID to the current projection; if no projection is set, it returns all table columns plus _ROW_ID.

candidates = (
    docs.search([0.1, 0.2, 0.3], column="embedding")
    .where("category = 'lake'")
    .select(["id", "content"])
    .limit(100)
    .with_row_id()
    .to_pandas()
)

row_ids = candidates["_ROW_ID"].tolist()

When a row-id manifest is persisted beyond a short-lived latest-snapshot workflow, record the source snapshot or tag and pass it back to the read API. scan, search, search_vectors, search_hybrid, and take_row_ids accept snapshot_id or tag_name. These two options are mutually exclusive.

Use take_row_ids to fetch rows selected by an earlier scan, vector search, full-text search, hybrid search, sampler, or split manifest. Results are not guaranteed to follow the input row-id order. Include _ROW_ID and reorder on the client if order matters.

payload = (
    docs.take_row_ids(row_ids, snapshot_id=source_snapshot_id)
    .select(["id", "content", "image"])
    .with_row_id()
    .to_arrow()
    .to_pylist()
)

payload_by_row_id = {row["_ROW_ID"]: row for row in payload}
ordered_payload = [payload_by_row_id[row_id] for row_id in row_ids]

This pattern keeps broad candidate generation cheap: first search or filter to produce a compact row-id manifest, then fetch only the columns needed by the next stage.

# Stage 1: broad retrieval with a narrow projection.
manifest = (
    docs.search(
        query_vector,
        column="embedding",
        snapshot_id=source_snapshot_id,
    )
    .select(["id"])
    .limit(500)
    .with_row_id()
    .to_pandas()
)

# Stage 2: expensive reranking payload, fetched only for candidates.
rerank_payload = (
    docs.take_row_ids(
        manifest["_ROW_ID"].tolist(),
        snapshot_id=source_snapshot_id,
    )
    .select(["id", "content", "image"])
    .with_row_id()
    .to_list()
)

payload_by_row_id = {row["_ROW_ID"]: row for row in rerank_payload}
rerank_inputs = [
    {
        "row_id": row_id,
        "candidate_rank": rank,
        "doc": payload_by_row_id[row_id],
    }
    for rank, row_id in enumerate(manifest["_ROW_ID"])
]

For offline inference, feature backfills, or training splits, store row IDs in a work queue or manifest table together with the source table snapshot or tag used to produce them. Workers can then read row-id batches and fetch only the columns they need:

def run_inference_worker(docs, row_id_batch, source_snapshot_id):
    rows = (
        docs.take_row_ids(row_id_batch, snapshot_id=source_snapshot_id)
        .select(["id", "content"])
        .with_row_id()
        .to_list()
    )

    outputs = []
    for row in rows:
        outputs.append(
            {
                "id": row["id"],
                "source_row_id": row["_ROW_ID"],
                "prediction": model_predict(row["content"]),
            }
        )
    return outputs

Best practices:

• Treat _ROW_ID as an internal row handle, not a business identifier.
• Store the source snapshot or tag with persisted row-id manifests.
• Pass the stored snapshot_id or tag_name when consuming persisted row-id manifests.
• Use business keys or application columns when writing inference or training outputs back to a table.
• Reorder take_row_ids results client-side when input order matters.

BLOB URI affinity

For small inference batches, map_with_blobs(..., batch_size=32, blob_uri_affinity=True) sorts scalar BLOB descriptors by URI and offset to coalesce reads across inference batches. This distributed sort may reorder rows; use it when fewer storage requests outweigh the shuffle cost. prefetch_bytes defaults to 64 MiB per payload window, except for an oversized inference batch. Descriptor batches contain up to max(1024, batch_size) rows; the UDF still receives at most batch_size rows. MAP and ARRAY BLOB columns use the default mapping path without URI affinity.