Back to Details

rerun-catalog-queries

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Rerun catalog queries

Rerun catalog 查询

Practical performance patterns for querying a Rerun catalog from Python (
rerun.catalog.CatalogClient
→ 
dataset.reader(...)
→ DataFusion 
DataFrame
). The DataFusion side of the stack is covered by the 
datafusion-python
 skill — load that for 
DataFrame
/ 
SessionContext
/ expression-API references. This skill focuses on catalog-specific behaviors and the round-trip costs that catch teams off guard.
从Python查询Rerun catalog的实用性能模式（
rerun.catalog.CatalogClient
→ 
dataset.reader(...)
→ DataFusion 
DataFrame
）。DataFusion相关内容由**
datafusion-python
**技能覆盖——如需
DataFrame
/
SessionContext
/表达式API相关参考，请加载该技能。本技能聚焦于catalog特有的行为，以及容易让团队踩坑的往返成本。

The query cost model in one sentence

一句话总结查询成本模型

Every materialization of a 
dataset.reader(...)
DataFrame is one cloud round-trip, with most of the cost being the network/decode pair, not the compute. Plan for round-trip count and payload bytes, in that order.
A typical catalog round-trip is a few seconds even for a tiny result. So:
  • 30 segments × 1 query each ≈ 90s. (Naive per-segment loop.)
  • 30 segments × 4 queries each ≈ 6 minutes. (A splitter that runs 
    count
    + 
    collect_column
    for both starts and stops.)
  • 1 query covering all 30 segments ≈ 3s.
The same fan-out happens along the entity axis: 10 entities × 1 
filter_contents([one_entity]).reader()
each ≈ 30s, vs one 
filter_contents([all_entities]).reader()
≈ 3s. Push as much as you can into one round-trip — across segments and across entities.
每次实例化
dataset.reader(...)
生成的
DataFrame
都会产生一次云服务往返请求，其中大部分成本来自网络传输和解码，而非计算。请优先规划往返请求的次数，其次关注 payload 字节大小。
即使是极小的查询结果，一次典型的catalog往返请求通常也需要几秒时间。因此：
  • 30个段 × 每个段1次查询 ≈ 90秒。（朴素的按段循环方式。）
  • 30个段 × 每个段4次查询 ≈ 6分钟。（同时执行
    count
    +
    collect_column
    来获取起始和终止数据的拆分器。）
  • 1次查询覆盖所有30个段 ≈ 3秒。
同样的请求扩散也会沿实体维度发生：10个实体 × 每个实体1次
filter_contents([one_entity]).reader()
≈ 30秒，而1次
filter_contents([all_entities]).reader()
≈ 3秒。尽可能将查询合并到一次往返请求中——跨段、跨实体合并。

Always apply 
filter_contents
and time-window filters before 
.reader(...)

务必在调用
.reader(...)
前应用
filter_contents
和时间窗口过滤器

The single biggest lever:
  • filter_contents([entity_globs])
    restricts which entity-path columns the reader produces. Without it, every entity in the dataset is read.
  • For Scalars-typed columns this also reduces array nesting depth from 
    list<list<double>>
    to 
    list<double>
    .
  • Time-window filters (
    df.filter(col(index).cast(int64) >= start) .filter(col(index).cast(int64) <= end)
    ) push down to storage and dramatically reduce bytes scanned. The order matters: filter then reader-bound projection, never the other way around.
Combined: 
dataset.filter_segments(seg).filter_contents(entities).reader(...) .filter(in_window).select(...)
.
这是提升性能最关键的手段：
  • filter_contents([entity_globs])
    会限制读取器生成的实体路径列范围。如果不使用它，数据集内的所有实体都会被读取。
  • 对于Scalars类型的列，这还能将数组嵌套深度从
    list<list<double>>
    降低到
    list<double>
    。
  • 时间窗口过滤器（
    df.filter(col(index).cast(int64) >= start) .filter(col(index).cast(int64) <= end)
    ）会下推到存储层，大幅减少扫描的字节数。顺序很重要：先过滤再执行读取器绑定的投影操作，绝不能反过来。
组合使用示例：
dataset.filter_segments(seg).filter_contents(entities).reader(...) .filter(in_window).select(...)
。

df.cache()
is your friend for repeated probes

df.cache()
是重复查询的好帮手

When the same materialization gets used by multiple downstream filter/count/collect calls, materialize once with 
DataFrame.cache()
and operate on the cached frame:
python
cached = (
    dataset
    .filter_segments(seg)
    .filter_contents([entity])
    .reader(index=index_col)
    .select(col(index_col).cast(pa.int64()).alias(index_col), value.alias("v"))
    .cache()  # one network round-trip, materializes into in-memory batches
)
starts = cached.filter(col("v") == start_val).collect_column(index_col)
stops = cached.filter(stop_pred(col("v"))).collect_column(index_col)
Without 
cache()
, each 
count()
/ 
collect_column()
re-executes the whole reader chain.
When NOT to cache. 
cache()
forces materialization into Arrow batches, breaking laziness. If downstream code keeps composing more DataFusion ops on top (joins, windows, further filters) and only materializes once at the end, caching mid-pipeline turns one execution into two and pre-empts whatever physical-plan optimizations the engine could have done across the boundary. Reach for 
cache()
when the consumers are terminal (
count()
, 
collect_column()
, 
to_arrow_table()
), not when they're another lazy 
DataFrame
.
当同一个实例化结果被多个下游的filter/count/collect调用使用时，用
DataFrame.cache()
实例化一次，然后基于缓存的DataFrame进行操作：
python
cached = (
    dataset
    .filter_segments(seg)
    .filter_contents([entity])
    .reader(index=index_col)
    .select(col(index_col).cast(pa.int64()).alias(index_col), value.alias("v"))
    .cache()  # 一次网络往返请求，将数据实例化为内存批处理
)
starts = cached.filter(col("v") == start_val).collect_column(index_col)
stops = cached.filter(stop_pred(col("v"))).collect_column(index_col)
如果不使用
cache()
，每次
count()
/
collect_column()
都会重新执行整个读取器链。
不适合使用cache的场景：
cache()
会强制将数据实例化为Arrow批处理，打破惰性执行。如果下游代码还会在其之上组合更多DataFusion操作（连接、窗口、进一步过滤），并且仅在最后才实例化结果，那么在流水线中途缓存会将一次执行拆分为两次，还会阻止引擎在边界处进行物理计划优化。仅当消费者是终端操作（
count()
、
collect_column()
、
to_arrow_table()
）时才使用
cache()
，而不是当消费者是另一个惰性
DataFrame
时。

Cross-segment batching: drop 
filter_segments
, group by 
rerun_segment_id

跨段批处理：去掉
filter_segments
，按
rerun_segment_id
分组

For pipelines that need the same query on many segments, omit 
filter_segments(...)
entirely and pull a single cross-segment table. Every reader row carries a 
rerun_segment_id
column — group locally:
python
df = dataset.filter_contents(entities).reader(index=index_col)
cached = df.select(
    "rerun_segment_id",
    col(index_col).cast(pa.int64()).alias(index_col),
    value.alias("v"),
).cache()
对于需要在多个段上执行相同查询的流水线，完全省略
filter_segments(...)
，直接拉取一张跨段表。读取器返回的每一行都带有
rerun_segment_id
列——在本地进行分组：
python
df = dataset.filter_contents(entities).reader(index=index_col)
cached = df.select(
    "rerun_segment_id",
    col(index_col).cast(pa.int64()).alias(index_col),
    value.alias("v"),
).cache()

Now N filter/aggregate calls are local, not network.

现在N次过滤/聚合操作都是本地操作，无需网络请求。

starts = cached.filter(col("v") == start_val).select("rerun_segment_id", index_col).to_arrow_table()

Trigger / event columns are tiny enough that pulling all segments at
once dominates per-segment looping by an order of magnitude.

---
starts = cached.filter(col("v") == start_val).select("rerun_segment_id", index_col).to_arrow_table()

触发/事件列的数据量极小，因此一次性拉取所有段的数据比按段循环快一个数量级。

---

Per-entity fan-out within a segment

单段内的按实体请求扩散

Symmetric to cross-segment batching, along the entity axis. If you need data from N entities of a single segment, don't loop:
python
undefined
与跨段批处理对称，沿实体维度优化。如果需要从单个段的N个实体获取数据，不要循环处理：
python
undefined

Anti-pattern: N reader setups, N round-trips.

反模式：N次读取器初始化，N次往返请求。

for entity in entities: df = dataset.filter_segments(seg).filter_contents([entity]).reader(index=ix) ...

Instead pull all N at once and project per entity locally. Per "Reader
row layout" below, every row carries data for one entity and NULLs
for the others, so `col("<entity>:<archetype>:<component>").is_not_null()`
is the per-entity filter:

```python
shared = (
    dataset
    .filter_segments(seg)
    .filter_contents(sorted(set(entities)))
    .reader(index=ix)
    .filter(col(ix).cast(pa.int64()).between(start_ns, end_ns))
)
for entity in entities: df = dataset.filter_segments(seg).filter_contents([entity]).reader(index=ix) ...

取而代之的是一次性拉取所有N个实体的数据，然后在本地按实体投影。根据下方的“读取器行布局”，每一行只包含一个实体的数据，其他实体的列均为NULL，因此可以用`col("<entity>:<archetype>:<component>").is_not_null()`作为按实体过滤的条件：

```python
shared = (
    dataset
    .filter_segments(seg)
    .filter_contents(sorted(set(entities)))
    .reader(index=ix)
    .filter(col(ix).cast(pa.int64()).between(start_ns, end_ns))
)

Each downstream consumer narrows to its entity's rows lazily.

每个下游消费者会惰性地缩小到对应实体的行。

src_a = shared.filter(col(f"{ent_a}:{comp_a}").is_not_null()).select(ix, f"{ent_a}:{comp_a}") src_b = shared.filter(col(f"{ent_b}:{comp_b}").is_not_null()).select(ix, f"{ent_b}:{comp_b}")

DataFusion can share the underlying scan across the per-entity
projections when it builds the physical plan, so this stays a single
catalog round-trip even though there are N logical consumers. Works
inside generators that build per-source DataFusion plans (resampling,
bracket lookup, nearest-in-time joins) — collapse the network fan-out
without changing the per-source logic.

A trap when refactoring: if a downstream query uses a reader column's
fully-qualified name (`col(f"{entity}:{archetype}:{component}")`),
you don't need to alias the column in `shared`. The shared reader's
output schema preserves native column names, so existing per-entity
projection helpers keep working unchanged.

---
src_a = shared.filter(col(f"{ent_a}:{comp_a}").is_not_null()).select(ix, f"{ent_a}:{comp_a}") src_b = shared.filter(col(f"{ent_b}:{comp_b}").is_not_null()).select(ix, f"{ent_b}:{comp_b}")

当DataFusion构建物理计划时，可以在按实体投影之间共享底层扫描操作，因此即使有N个逻辑消费者，这仍然只是一次catalog往返请求。这种方式适用于构建按源DataFusion计划的生成器（重采样、括号查找、最近时间连接）——无需改变按源逻辑即可消除网络请求扩散。

重构时的陷阱：如果下游查询使用读取器列的全限定名称（`col(f"{entity}:{archetype}:{component}")`），则无需在`shared`中对列进行别名。共享读取器的输出模式会保留原生列名，因此现有的按实体投影辅助工具无需修改即可正常工作。

---

count()
is not free

count()
并非无成本

Counter-intuitive: 
df.count()
and 
df.aggregate([], [F.count(col)])
do not always push down. Aggregate plans can force the engine to materialize the underlying column data server-side, then count on the client. 
F.count(col)
over wide entity-columns can ship full struct or blob payloads to count nullity.
Alternatives, in order of preference for "is anything here":
NeedUse
"any row in this filter?"
df.select(col(index)).limit(1).to_arrow_table().num_rows > 0
— server short-circuits on first match
"count rows in a tiny window"
df.filter(window).select(col(index)).count()
after the time filter
"count each entity in a wide query"per-entity 
limit(1)
probes, threaded — not one big 
count(col)
aggregate
A trap that fooled us: assuming 
bool_or(col.is_not_null())
would only need nullity buffers. It does not — the operator still touches payload data on most plans.
违反直觉的是：
df.count()
和
df.aggregate([], [F.count(col)])
并不总是能下推到存储层。聚合计划可能会强制引擎在服务器端实例化底层列数据，然后在客户端进行计数。对宽实体列执行
F.count(col)
可能会传输完整的结构体或blob payload来统计空值情况。
针对“是否存在数据”需求的替代方案，按优先级排序：
需求解决方案
“该过滤器下是否存在数据行？”
df.select(col(index)).limit(1).to_arrow_table().num_rows > 0
— 服务器会在找到首个匹配项后立即终止查询
“统计小窗口内的数据行数”在时间过滤后执行
df.filter(window).select(col(index)).count()
“统计宽查询中每个实体的数据行数”按实体执行
limit(1)
探测，并行处理 — 不要用一次大型
count(col)
聚合
我们曾踩过的陷阱：假设
bool_or(col.is_not_null())
只需要空值缓冲区。事实并非如此——在大多数计划中，该操作仍然会触及payload数据。

using_index_values
+ 
fill_latest_at
is great for resampling, not for presence

using_index_values
+ 
fill_latest_at
适合重采样，不适合存在性检查

python
.reader(index=index_col, using_index_values=targets, fill_latest_at=True)
  • Returns one row per target timestamp, each entity column carrying its latest non-null value at-or-before the target.
  • Excellent for nearest-prior resampling (no DataFusion required).
  • Don't use it as a presence check. Two reasons:
    1. Semantics are "ever emitted before T", not "emitted in [start, T]".
    2. The server still ships the full struct/blob payload for every entity to compute the latest-known value; a downstream 
      is_not_null()
      projection runs post-transfer and doesn't reduce wire bytes.
For a strict in-window presence check, prefer per-entity 
limit(1)
over the time-filtered reader, run concurrently.
python
.reader(index=index_col, using_index_values=targets, fill_latest_at=True)
  • 每个目标时间戳返回一行，每个实体列携带该时间戳或之前最新的非空值。
  • 非常适合最近前驱重采样（无需DataFusion）。
  • 不要将其用作存在性检查。原因有二：
    1. 语义是“在T之前是否曾发送过”，而非“在[start, T]区间内是否发送过”。
    2. 服务器仍会传输每个实体的完整结构体/blob payload来计算最新已知值；下游的
      is_not_null()
      投影是在传输后执行的，无法减少网络字节数。
如需严格的窗口内存在性检查，建议对时间过滤后的读取器并行执行按实体
limit(1)
探测。

Schema introspection is cheap; use it before probing

模式 introspection 成本低；在探测前使用它

python
schema = dataset.filter_segments(seg).schema()
available = {(c.entity_path, c.component) for c in schema.component_columns()}
This is a single round-trip and tells you which 
(entity, component)
pairs the segment registered. If a column isn't in the schema, you can drop it from your manifest without any further cloud queries. This is often a complete substitute for "does this entity have events" probing.
Caveat: schema presence ≠ events. The MCAP records the topic schema even for unused topics. If your pipeline cares about distinguishing "registered but never emitted" vs "registered with events", you have to probe — see "is anything here" patterns above.
python
schema = dataset.filter_segments(seg).schema()
available = {(c.entity_path, c.component) for c in schema.component_columns()}
这是一次往返请求，能告诉你该段注册了哪些
(entity, component)
对。如果某个列不在模式中，你可以在不进行进一步云查询的情况下将其从清单中移除。这通常可以完全替代“该实体是否有事件”的探测操作。
注意：模式存在 ≠ 存在事件。MCAP会记录主题模式，即使该主题未被使用。如果你的流水线需要区分“已注册但从未发送”和“已注册且有事件”，则必须进行探测——参考上述“是否存在数据”的模式。

Reader row layout: entities are columns, not rows

读取器行布局：实体是列，不是行

Every row from 
dataset.<filters>.reader(...)
corresponds to a single event on a single entity. Other entities' columns are null on that row. Implications:
  • select("rerun_segment_id", "<entity>:<archetype>:<component>")
    works — quote the entity column when using SQL.
  • There is no 
    rerun_entity_path
    row attribute. To attribute rows to entities you either filter to one entity at a time, or pick a per-entity column (e.g. 
    :McapChannel:id
    ) whose non-null pattern identifies the source.
  • df.count()
    returns total events across all entities, not per-entity counts.
dataset.<filters>.reader(...)
返回的每一行对应单个实体的单个事件。该行中其他实体的列均为NULL。这意味着：
  • select("rerun_segment_id", "<entity>:<archetype>:<component>")
    是可行的——在使用SQL时，请为实体列添加引号。
  • 不存在**
    rerun_entity_path
    **行属性。要将行关联到实体，你要么每次过滤到一个实体，要么选择一个按实体划分的列（例如
    :McapChannel:id
    ），其非空模式可以标识数据源。
  • df.count()
    返回所有实体的总事件数，而非按实体统计的数量。

Field access on a null struct returns 
0.0
, not null

对空结构体进行字段访问返回
0.0
，而非NULL

A DataFusion gotcha that bites pipelines reading struct messages:
python
col("/some/entity:msg.MyType:message")[0]["sub"]["x"]
这是DataFusion的一个陷阱，会影响读取结构体消息的流水线：
python
col("/some/entity:msg.MyType:message")[0]["sub"]["x"]

When the parent struct is null on a row, this evaluates to 0.0

当父结构体在某一行中为NULL时，该表达式的计算结果为0.0

(and "" for strings), not null.

（字符串类型则为""），而非NULL。


Wrap struct-walk projections with a null guard:

```python
parent = col("/some/entity:msg.MyType:message")
leaf = parent[0]["sub"]["x"]
guard = parent.is_null() | parent[0].is_null() | parent[0]["sub"].is_null()
expr = F.when(guard, lit(None)).otherwise(leaf)
Only apply this to struct sources. For scalar / blob columns (
Scalars:scalars
, 
EncodedImage:blob
, etc.) the wrap is a no-op at best and at worst rewrites the plan in ways that change downstream join behavior. Gate the guard on whether the source actually walks through a struct boundary.

用空值保护包装结构体遍历投影：

```python
parent = col("/some/entity:msg.MyType:message")
leaf = parent[0]["sub"]["x"]
guard = parent.is_null() | parent[0].is_null() | parent[0]["sub"].is_null()
expr = F.when(guard, lit(None)).otherwise(leaf)
仅对结构体源应用此操作。对于标量/blob列（
Scalars:scalars
、
EncodedImage:blob
等），包装操作最好的情况是无作用，最坏的情况会重写计划，改变下游连接行为。仅当源确实需要遍历结构体边界时才添加保护。

Common debug recipe

常见调试流程

When a query stage is slower than expected:
  1. Count round-trips. Wrap each 
    to_arrow_table()
    / 
    collect_column()
    / 
    count()
    with 
    time.perf_counter()
    . Each is a round-trip. If you see N segment queries, that's N × few-seconds minimum.
  2. Split build vs materialize timing. Time the lazy DataFrame construction separately from the terminal 
    to_arrow_table()
    call. If "build" takes seconds, something inside is materializing eagerly (an Arrow round-trip in a join helper, a 
    cache()
    in a generator, a 
    .collect()
    hidden in a chained-join utility). A correctly lazy plan should build in ~milliseconds regardless of result size.
  3. Measure bytes. 
    tbl.nbytes
    after 
    to_arrow_table()
    reveals when "I projected 
    is_not_null()
    " actually shipped megabytes. If bytes are large despite a small projection, the operator didn't push down.
  4. Cross-segment first, then cross-entity. If the per-segment query is fundamentally the same (just scoped by id), drop 
    filter_segments
    and group by 
    rerun_segment_id
    locally. If you also have a per-entity loop within a segment, collapse it the same way (one 
    filter_contents([all])
    reader, per-entity 
    is_not_null()
    filters downstream).
  5. Cache before terminal re-use. If two 
    count()
    / 
    collect_column()
    calls share the same reader, 
    df.cache()
    between them. Don't cache if the consumers are themselves lazy DataFrames being composed further — caching breaks plan-wide optimization.
  6. Window first. Always push the time filter before any projection or aggregate that touches payload columns.
当查询阶段比预期慢时：
  1. 统计往返请求次数。用
    time.perf_counter()
    包装每个
    to_arrow_table()
    /
    collect_column()
    /
    count()
    调用。每个调用都是一次往返请求。如果看到N次段查询，那至少需要N×几秒的时间。
  2. 拆分构建与实例化时间。分别统计惰性DataFrame构建的时间和终端
    to_arrow_table()
    调用的时间。如果“构建”耗时几秒，说明内部有某些操作在提前实例化数据（连接辅助工具中的Arrow往返请求、生成器中的
    cache()
    、链式连接工具中隐藏的
    .collect()
    ）。正确的惰性计划无论结果大小如何，构建时间都应该在毫秒级。
  3. 测量字节数。
    to_arrow_table()
    之后的
    tbl.nbytes
    可以揭示“我只投影了
    is_not_null()
    ”却实际传输了兆字节数据的情况。如果投影范围很小但字节数很大，说明操作没有下推到存储层。
  4. 先跨段，再跨实体。如果按段查询本质上是相同的（只是按id限定范围），则去掉
    filter_segments
    ，在本地按
    rerun_segment_id
    分组。如果在单个段内还有按实体循环，也用同样的方式合并（一次
    filter_contents([all])
    读取器，下游按
    is_not_null()
    过滤实体）。
  5. 终端复用前先缓存。如果两个
    count()
    /
    collect_column()
    调用共享同一个读取器，在它们之间使用
    df.cache()
    。如果消费者本身是会被进一步组合的惰性DataFrame，则不要缓存——缓存会破坏全局计划优化。
  6. 先应用窗口过滤。在任何触及payload列的投影或聚合操作之前，务必先应用时间过滤。

See also

另请参阅

  • datafusion-python
    skill — DataFrame API, SQL parity, expression building, common pitfalls (boolean operators, immutability, etc.). Not installed? Ask the user to install it globally: 
    npx skills add apache/datafusion-python
    .
  • datafusion-python
    技能——DataFrame API、SQL兼容性、表达式构建、常见陷阱（布尔运算符、不可变性等）。未安装？请用户全局安装：
    npx skills add apache/datafusion-python
    。