rerun-catalog-queries

查询 Rerun catalog

使用 Python 查询 Rerun catalog 的实用性能模式 （rerun.catalog.CatalogClient → dataset.reader(...) → DataFusion DataFrame）。DataFusion 部分的 技术栈由 datafusion-python 技能介绍——加载该技能可查阅 DataFrame / SessionContext / expression API 参考。本技能聚焦于 catalog 特有的行为，以及容易让团队措手不及的往返开销。

---

使用 rerun:// URL 连接

from rerun.catalog import CatalogClient

client = CatalogClient("rerun+https://api.example.cloud.rerun.io:443")
dataset = client.get_dataset(id=dataset_id)

CatalogClient 接受 rerun:// 或 rerun+https:// URL，本地服务则使用 rerun+http://。不要传入普通的 https:// URL。像 rerun://host:443/dataset/<dataset-id>?segment_id=<segment-id> 这样的 segment URL 会指定服务、dataset ID 和 segment ID。

客户端会从 REDAP_TOKEN 或 rerun auth login 存储的凭据中读取令牌，因此通常无需 传入 token=。

---

一句话说明查询成本模型

每次物化 dataset.reader(...) DataFrame 都会产生一次云端往返， 大部分成本来自网络和解码，而非计算。请优先规划往返次数，其次再规划负载字节数。

即使结果很小，catalog 的一次典型往返也需要几秒钟。因此：

• 30 个 segment × 每个查询 1 次 ≈ 90s。（朴素的逐 segment 循环。）
• 30 个 segment × 每个查询 4 次 ≈ 6 分钟。（一个拆分器分别对起点和终点运行 count + collect_column。）
• 1 次查询覆盖全部 30 个 segment ≈ 3s。

沿着实体维度也会发生相同的扇出：10 个实体 × 每个实体执行 1 次 filter_contents([one_entity]).reader() ≈ 30s；而一次 filter_contents([all_entities]).reader() ≈ 3s。尽可能将更多操作合并到一次往返中——跨 segment，也跨实体。

---

始终在 .reader(...) 前应用 filter_contents 和时间窗口筛选

这是影响最大的优化手段：

• filter_contents([entity_globs]) 会限制读取器生成的实体路径列。若不指定， 则会读取数据集中的每个实体。
• 时间窗口筛选会下推至存储层；当它将原始索引列与字面量比较时，可大幅减少扫描的 字节数：df.filter(col(index) >= lit(start)).filter(col(index) <= lit(end))。 任何单位的整数、时间戳和时长字面量都适用。对索引列执行 cast 或算术运算则不会下推。

组合使用：dataset.filter_segments(seg).filter_contents(entities).reader(...) .filter(in_window).select(...)。

---

重复探测时，df.cache() 是好帮手

当多个下游筛选、计数或收集调用会使用同一次物化结果时，使用 DataFrame.cache() 物化一次，并在缓存帧上执行操作：

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)

不使用 cache() 时，每次 count() / collect_column() 都会重新执行整个读取器链。

何时不要缓存。 cache() 会将数据物化为 Arrow 批次，从而打破惰性求值。如果下游代码 继续在其上组合更多 DataFusion 操作（联接、窗口、进一步筛选），并且只在最后物化一次， 那么在流水线中途缓存会将一次执行变成两次，还会阻止引擎跨越这一边界进行可能的物理计划 优化。应在使用者会执行终结操作（count()、collect_column()、to_arrow_table()）时 使用 cache()，而不是在它们会继续返回另一个惰性 DataFrame 时使用。

---

跨 segment 批处理：省略 filter_segments，按 rerun_segment_id 分组

对于需要在多个 segment 上执行相同查询的流水线，完全省略 filter_segments(...)，一次性拉取跨 segment 的表。每一行读取器数据都带有 rerun_segment_id 列——在本地分组：

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.
starts = cached.filter(col("v") == start_val).select("rerun_segment_id", index_col).to_arrow_table()

触发器 / 事件列足够小，一次性拉取所有 segment 的效率远高于逐 segment 循环。

---

在一个 segment 内按实体扇出

这与跨 segment 批处理相对应，作用于实体轴。如果 需要从单个 segment 的 N 个实体获取数据，不要循环：

# Anti-pattern: N reader setups, N round-trips.
for entity in entities:
    df = dataset.filter_segments(seg).filter_contents([entity]).reader(index=ix)
    ...

而是一次性读取全部数据，然后在本地按实体投影。根据下文的“Reader 行布局”，每行都包含一个实体的数据，其他实体对应的列为 NULL，因此 col("<entity>:<archetype>:<component>").is_not_null() 就是按实体筛选的条件：

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 构建物理计划时，可以在各实体投影之间共享底层扫描，因此即使有 N 个逻辑使用方，也只需与目录交互一次。这适用于构建每个源的 DataFusion 计划的生成器（重采样、括号查找、最近时间连接）——无需更改每个源的逻辑，即可减少网络扇出。

重构时要注意：如果下游查询使用 Reader 列的完整限定名（col(f"{entity}:{archetype}:{component}")），则无需在 shared 中为列设置别名。共享 Reader 的输出架构会保留原生列名，因此现有的按实体投影辅助函数无需更改即可继续工作。

---

count() 并非没有开销

这可能有悖直觉：df.count() 和 df.aggregate([], [F.count(col)]) 并不总是会下推。聚合计划可能会迫使引擎在服务端物化底层列数据，然后在客户端计数。对宽实体列使用 F.count(col)，可能会传输完整的结构体或 blob 载荷来统计空值数量。

对于“这里是否有数据”，以下替代方案按优先顺序排列：

需求	使用方式
“此筛选条件下有行吗？”	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()) 只需要空值缓冲区。事实并非如此——在大多数计划中，该运算符仍会访问载荷数据。

---

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

.reader(index=index_col, using_index_values=targets, fill_latest_at=True)

• 每个目标时间戳返回一行，每个实体列都包含目标时间戳及之前最新的非空值。
• 非常适合最近先前值重采样（无需 DataFusion）。
• 不要将其用于存在性检查，原因有二：
  1. 其语义是“在 T 之前曾发出过”，而不是“在 [start, T] 内发出过”。
  2. 服务端仍会传输每个实体的完整结构体或 blob 载荷，以计算最新已知值；下游的 is_not_null() 投影在传输后才运行，不会减少网络传输的字节数。

若要严格检查窗口内是否存在数据，优先对经过时间筛选的 Reader 按实体使用 limit(1)，并发运行。

---

使用 dataset.info() 探索陌生数据集

info = dataset.info()
print(info)  # compact text overview

它会报告 segment 的数量和大小、每一层包含该层的 segment 数量、每条时间线的类型、取值范围和 segment 覆盖情况，以及属性列名。它只读取 segment 表和架构，不读取记录数据，因此即使数据集很大，开销也很低。

使用它来选择 reader(index=…)，并在首次实际查询前确定时间窗口筛选条件的范围。TimelineInfo.min / .max 是可直接用作字面量的 numpy 值：

ts = next(t for t in info.timelines if t.name == "capture_time")
df = dataset.reader(index=ts.name).filter(col(ts.name) >= lit(ts.min))

此处找到的属性可以通过 dataset.segment_table() 筛选；层名称则会告诉你 register(…, layer_name=…) 已附加了哪些层。

---

Schema 检查成本低；先检查再探测

schema = dataset.filter_segments(seg).schema()
available = {(c.entity_path, c.component) for c in schema.component_columns()}

这只需一次往返，就能告诉你该 segment 注册了哪些 (entity, component) 对。如果某列不在 schema 中，就可以从清单中移除，无需再进行任何云端查询。这通常可以完全替代“此 entity 是否有事件”的探测。

注意：schema 存在 ≠ 有事件。即使未使用 topic，MCAP 也会记录其 topic schema。如果你的流水线需要区分“已注册但从未发出”与“已注册且有事件”，就必须进行探测——参见上文“这里是否有内容”模式。

---

Reader 行布局：entity 是列，而非行

来自 dataset.<filters>.reader(...) 的每一行都对应单个 entity 上的单个事件。其他 entity 的列在该行中为 null。这意味着：

• select("rerun_segment_id", "<entity>:<archetype>:<component>") 可以正常工作——使用 SQL 时，请为 entity 列加引号。
• 没有 rerun_entity_path 行属性。若要将行归属到 entity，可以一次只筛选一个 entity，或选择一个按 entity 区分的列（例如 :McapChannel:id），其非 null 模式可以标识来源。
• df.count() 返回的是所有 entity 的事件总数，而非各 entity 的事件数。

---

在 null struct 上访问字段会返回 0.0，而不是 null

DataFusion 中一个会影响读取 struct 消息的流水线的陷阱：

col("/some/entity:msg.MyType:message")[0]["sub"]["x"]
# When the parent struct is null on a row, this evaluates to 0.0
# (and "" for strings), not null.

对 struct 遍历投影添加 null 保护：

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)

仅对 struct 源应用此处理。 对标量 / blob 列（Scalars:scalars、EncodedImage:blob 等）而言，这种包装最多只是无效操作，最坏情况下则会以改变下游 join 行为的方式重写执行计划。只有在源确实遍历 struct 边界时，才启用此保护。

---

常见调试方法

当查询阶段比预期慢时：

1. 统计往返次数。 用 time.perf_counter() 包裹每个 to_arrow_table() / collect_column() / count()。每次调用都是一次往返。如果看到 N 次 segment 查询，至少需要 N × 数秒。
2. 分别统计构建和物化耗时。 分别测量惰性 DataFrame 的构建耗时和终端 to_arrow_table() 调用的耗时。如果“构建”耗时达到数秒，说明某处正在急切地物化数据（join 辅助函数中的 Arrow 往返、生成器中的 cache()，或链式 join 工具中隐藏的 .collect()）。正确的惰性计划无论结果大小如何，都应在约几毫秒内完成构建。
3. 测量字节数。 在调用 to_arrow_table() 后查看 tbl.nbytes，可以发现“我投影了 is_not_null()”实际却传输了数 MB 的情况。如果投影很小但字节数很大，说明该算子没有下推。
4. 先跨 segment，再跨 entity。 如果每个 segment 的查询本质相同（只是按 id 限定范围），就移除 filter_segments，并在本地按 rerun_segment_id 分组。如果 segment 内还有按 entity 循环的逻辑，也用同样方式折叠（使用一个 filter_contents([all]) reader，并在下游按 entity 进行 is_not_null() 筛选）。
5. 在终端调用重复使用前缓存。 如果两个 count() / collect_column() 调用共用同一个 reader，就在它们之间调用 df.cache()。如果使用方本身是还要继续组合的惰性 DataFrame，则不要缓存——缓存会破坏整个计划的优化。
6. 先限定时间窗口。 在任何涉及 payload 列的投影或聚合之前，始终先应用时间筛选。

---

另请参阅

• datafusion-python 技能——DataFrame API、SQL 对等性、表达式构建和常见陷阱（布尔运算符、不可变性等）。 如果未安装？请让用户全局安装： npx skills add apache/datafusion-python。