Dalaran Catalog Queries

Dalaran catalog queries

Practical performance patterns for querying a Dalaran catalog from Python (dalaran.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.

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.

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

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(...).

df.cache() is your friend for repeated probes

When the same materialization gets used by multiple downstream filter/count/collect calls, materialize once with DataFrame.cache() and operate on the cached frame:

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.

Cross-segment batching: drop filter_segments, group by dalaran_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 dalaran_segment_id column — group locally:

df = dataset.filter_contents(entities).reader(index=index_col)
cached = df.select(
    "dalaran_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("dalaran_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.

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:

# Anti-pattern: N reader setups, N round-trips.
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:

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.

count() is not free

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":

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.

using_index_values + fill_latest_at is great for resampling, not for presence

.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.

Schema introspection is cheap; use it before probing

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.

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("dalaran_segment_id", "<entity>:<archetype>:<component>") works — quote the entity column when using SQL.
• There is no dalaran_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.

Field access on a null struct returns 0.0, not null

A DataFusion gotcha that bites pipelines reading struct messages:

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.

Wrap struct-walk projections with a null guard:

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.

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 dalaran_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.

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.