rerun-catalog-queries

Rerun カタログクエリ

Python から Rerun カタログをクエリする実践的なパフォーマンスパターン （rerun.catalog.CatalogClient → dataset.reader(...) → DataFusion DataFrame）。スタックの DataFusion 側については datafusion-python スキルで扱っています。DataFrame／SessionContext／式 API のリファレンスが必要な場合は、そちらを読み込んでください。このスキルでは、カタログ固有の動作と、チームが不意を突かれやすいラウンドトリップのコストに焦点を当てます。

---

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> のようなセグメント URL には、サーバー、データセット ID、セグメント ID が含まれます。

クライアントは REDAP_TOKEN からトークンを読み取るか、rerun auth login で保存された認証情報を使うため、通常は token= を指定する必要はありません。

---

クエリのコストモデルを一言で説明すると

dataset.reader(...) の DataFrame を実体化するたびに、クラウドへのラウンドトリップが 1 回発生します。コストの大部分は計算ではなく、ネットワークとデコードの組み合わせです。まずラウンドトリップの回数を、次にペイロードのバイト数を考慮してください。

典型的なカタログのラウンドトリップは、結果がわずかでも数秒かかります。そのため、以下のようになります。

• 30 セグメント × 各 1 クエリ ≈ 90 秒。（セグメントごとにループする単純な方法。）
• 30 セグメント × 各 4 クエリ ≈ 6 分。（開始値と終了値の両方に対して count と collect_column を実行する分割処理。）
• 30 セグメントすべてを対象とする 1 クエリ ≈ 3 秒。

同じようなファンアウトはエンティティ方向にも発生します。10 エンティティ × 各エンティティに対する filter_contents([one_entity]).reader() 1 回 ≈ 30 秒であるのに対し、filter_contents([all_entities]).reader() 1 回なら ≈ 3 秒です。セグメント間でもエンティティ間でも、できるだけ多くの処理を 1 回のラウンドトリップにまとめてください。

---

.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 の演算（結合、ウィンドウ、追加のフィルターなど）をさらに組み合わせ、最後に一度だけ実体化する場合、パイプラインの途中でキャッシュすると、1 回の実行が 2 回になり、境界を越えてエンジンが実行計画に適用できた最適化が先回りして妨げられます。cache() を使うのは、利用側が終端処理（count()、collect_column()、to_arrow_table()）の場合にしてください。別の遅延 DataFrame に続ける場合には使わないでください。

---

セグメント間のバッチ処理：filter_segments を省略し、rerun_segment_id でグループ化する

多くのセグメントで同じクエリを実行するパイプラインでは、filter_segments(...) を完全に省略し、セグメントをまたぐテーブルを 1 つ取得します。リーダーの各行には 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()

トリガー／イベント列は非常に小さいため、セグメントごとにループするよりも、すべてのセグメントを一度に取得する方が桁違いに効率的です。

---

セグメント内でエンティティごとにファンアウトする

エンティティ軸に沿った処理で、セグメント間のバッチ処理と対称的です。単一セグメントの N 個のエンティティからデータが必要な場合、次のようにループしないでください。

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

代わりに N 個すべてを一度に取得し、エンティティごとにローカルで射影します。以下の「Reader の行レイアウト」のとおり、各行には 1 つのエンティティのデータが格納され、その他のエンティティは 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 個あっても、カタログへの往復は 1 回だけです。リサンプリング、ブラケット検索、最近傍時刻結合など、ソースごとの DataFusion プランを構築するジェネレーター内で使用できます。ソースごとのロジックを変更せずに、ネットワークのファンアウトを抑えられます。

リファクタリング時の落とし穴：下流のクエリでリーダー列の完全修飾名（col(f"{entity}:{archetype}:{component}")）を使う場合、shared 内で列にエイリアスを付ける必要はありません。共有リーダーの出力スキーマはネイティブな列名を保持するため、既存のエンティティごとの射影ヘルパーは変更せずに動作します。

---

count() は無料ではありません

直感に反しますが、df.count() と df.aggregate([], [F.count(col)]) は、常にプッシュダウンされるわけではありません。集約プランによっては、エンジンがサーバー側で基盤となる列データを具体化してから、クライアント側でカウントすることがあります。幅広いエンティティ列に対する F.count(col) は、NULL かどうかを数えるために、構造体や Blob のペイロード全体を転送する可能性があります。

「データが存在するかどうか」を調べる方法を、推奨順に示します。

必要なこと	使用するもの
「このフィルターに該当する行があるか？」	df.select(col(index)).limit(1).to_arrow_table().num_rows > 0 — サーバー側で最初の一致が見つかった時点で処理を終了します
「小さなウィンドウ内の行数を数える」	時間フィルターの後に df.filter(window).select(col(index)).count() を使用します
「幅広いクエリでエンティティごとに数える」	エンティティごとに limit(1) のプローブをスレッドで並行実行します。大きな count(col) 集約を 1 つ使う方法は避けてください

私たちも誤解していた落とし穴があります。bool_or(col.is_not_null()) なら NULL かどうかのバッファーだけが必要だと思っていましたが、そうではありません。ほとんどのプランでは、この演算子もペイロードデータにアクセスします。

---

using_index_values と fill_latest_at はリサンプリングには最適ですが、存在確認には向きません

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

• 対象タイムスタンプごとに 1 行を返し、各エンティティ列には対象時刻以前で最も新しい非 NULL 値が格納されます。
• 最近傍の過去値へのリサンプリングに最適です（DataFusion は不要です）。
• 存在確認には使用しないでください。理由は 2 つあります。
  1. セマンティクスは「T より前に一度でも出力された」であり、「[start, T] の範囲内で出力された」ではありません。
  2. 最新値を計算するため、サーバーはすべてのエンティティについて構造体や Blob のペイロード全体を転送します。下流の is_not_null() 射影は転送後に実行されるため、ネットワーク上のデータ量は減りません。

厳密なウィンドウ内の存在確認には、時間フィルターを適用したリーダーに対して、エンティティごとに limit(1) を使用し、並行して実行する方法を推奨します。

---

dataset.info() で不慣れなデータセットを調べる

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

セグメント数とサイズ、各レイヤーを持つセグメント数、各タイムラインの種類・値の範囲・セグメントの適用範囲、プロパティ列名が報告されます。セグメントテーブルとスキーマのみを読み取り、記録データは読み取らないため、大規模なデータセットでも低コストです。

最初の実際のクエリを実行する前に、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 = dataset.filter_segments(seg).schema()
available = {(c.entity_path, c.component) for c in schema.component_columns()}

これは単一のラウンドトリップで、セグメントに登録された (entity, component) のペアを確認できます。スキーマにないカラムは、追加のクラウドクエリを実行せずにマニフェストから削除できます。多くの場合、「このエンティティにイベントがあるか」を調べるプローブの完全な代替になります。

注意点：スキーマに存在すること ≠ イベントがあることです。未使用のトピックであっても、MCAP にはトピックスキーマが記録されます。「登録済みだが一度も出力されていない」状態と「イベントがある状態」を区別する必要があるパイプラインでは、プローブが必要です。上記の「ここに何かあるか」パターンを参照してください。

---

リーダーの行レイアウト：エンティティは行ではなくカラムです

dataset.<filters>.reader(...) の各行は、単一のエンティティ上の単一のイベントに対応します。その行では、ほかのエンティティのカラムは null です。これにより、次の点が重要になります。

• select("rerun_segment_id", "<entity>:<archetype>:<component>") は機能します。SQL でエンティティのカラムを使う場合は、引用符で囲んでください。
• 行属性 rerun_entity_path はありません。行をエンティティに割り当てるには、エンティティを一度に 1 つずつフィルタリングするか、null ではない値のパターンからソースを特定できるエンティティごとのカラム（例：:McapChannel:id）を選択します。
• df.count() は、エンティティごとの件数ではなく、すべてのエンティティのイベント総数を返します。

---

null の構造体にフィールドアクセスすると、null ではなく 0.0 が返ります

構造体メッセージを読み取るパイプラインで問題になる DataFusion の落とし穴です。

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.

構造体をたどる射影は、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)

これは構造体ソースにのみ適用してください。 スカラー／ブロブのカラム（Scalars:scalars、EncodedImage:blob など）では、囲んでも何も変わらないか、悪い場合は下流の結合動作を変える形でプランを書き換えてしまいます。ソースが実際に構造体境界をたどるかどうかに応じて、ガードを適用してください。

---

よくあるデバッグ手順

クエリステージが想定より遅い場合：

1. ラウンドトリップ数を数えます。 各 to_arrow_table()／ collect_column()／count() を time.perf_counter() で囲みます。各呼び出しが 1 回のラウンドトリップです。セグメントクエリが 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]) リーダーを 1 つ使い、下流でエンティティごとに is_not_null() でフィルタリングします）。
5. 終端処理で再利用する前にキャッシュします。 同じリーダーを使う 2 つの count()／collect_column() 呼び出しがある場合は、その間に df.cache() を挟みます。さらに合成される遅延 DataFrame をコンシューマーとして使う場合はキャッシュしないでください。キャッシュすると、プラン全体の最適化が妨げられます。
6. まず時間範囲を絞ります。 ペイロードカラムに触れる射影や集約を行う前に、必ず時間フィルターを適用してください。

---

関連項目

• datafusion-python スキル — DataFrame API、SQL との互換性、式の構築、よくある落とし穴（ブール演算子、不変性など）。 インストールされていませんか？ユーザーにグローバルインストールを依頼してください： npx skills add apache/datafusion-python.