Query Performance Tuning

This is a loose collection of considerations when querying Rerun datasets. Over time baseline performance will improve, rendering some of these approaches unnecessary. Since Rerun depends on DataFusion, some of these approaches are observations from our own usage.

First, generate a DataFrame for comparison:

sample_video_path = (
    Path(__file__).parents[4] / "tests" / "assets" / "rrd" / "video_sample"
)
server = rr.server.Server(datasets={"video_dataset": sample_video_path})
# Using OSS server for demonstration but in practice replace with
# the URL of your cloud instance
CATALOG_URL = server.url()
client = rr.catalog.CatalogClient(CATALOG_URL)
dataset = client.get_dataset(name="video_dataset")
df = dataset.filter_contents([
    "/compressed_images/**",
    "/raw_images/**",
]).reader(index="log_time")

Extract Python types from a DataFrame

DataFusion is a streaming query engine, which allows for processing arbitrarily large amounts of data. When working with smaller or filtered-down datasets that fit into memory, you can extract data into Python variables for further post processing. In these examples, we convert DataFrames to PyArrow tables to materialize them in memory. Similar patterns using Polars or Pandas also apply.

Prefer to_numpy

This is technically a PyArrow and general Python detail. For example, when extracting data from a PyArrow table, to_pylist can be multiple orders of magnitude slower, even when using to_numpy(zero_copy_only=False).

table = pa.table(df)
table["log_time"].to_numpy()
# vs.
table["log_time"].to_pylist()

Fine-tune data collection

Similar to the approach described above to collect a DataFusion DataFrame into a PyArrow table, you can instead collect the results in memory and keep them as a DataFrame. Then any operations on this in-memory (cached) DataFrame are typically very fast.

df.count()  # has to pull some data
df.count()  # has to pull same data again
# vs.
cache_df = df.cache()  # materializes table in memory
cache_df.count()  # basically free
cache_df.count()  # basically free

Where a query runs

reader() returns a DataFusion dataframe backed by a Rerun table provider. DataFusion plans the query and asks the provider which filters it can push to the server. The provider pushes segment, entity and time filters. The server uses the chunk index to return the chunks that can match, and DataFusion runs the rest of the plan on those chunks, on the client.

The server applies these filters:

The server filters whole chunks, so the chunks it returns can hold rows outside the filter. DataFusion filters the rows again, so the result is exact.

The server compares only the bare index column with a literal. col("time").cast(pa.int64()) >= t0 is not pushed: the client fetches every chunk of the selected entities and compares the rows.

is_not_null() on a component column makes the reader produce rows only where that component has data. It does not reduce the chunks fetched.

Everything else runs in DataFusion on the fetched rows: filters on data columns, aggregates, joins and sorts. LIMIT stops the fetch early but does not reach the server. With fill_latest_at=True the selected columns no longer narrow entities, because the other entities' timestamps still produce filled rows. Use filter_contents() instead.

segment_table() reads the index and fetches no chunks. Use it for per-segment facts such as duration, counts and property values.

Leverage sparsity to minimize scans

In a write once, read many paradigm adding an additional sparse column can enable cheap access to data of interest via filtering. Read the sparse column on its own first, which fetches only its chunks, then query the heavy data at its timestamps with using_index_values. For example, Mp4Reader logs VideoStream:is_keyframe with only true rows, so on Rerun Hub selecting just that column finds the keyframes without fetching any video samples.

# Create a new sparse layer identifying interesting events
segment_id = dataset.segment_ids()[0]
second_to_last_timestamp = pa.table(df)["log_time"].to_numpy()[-2]
with rr.RecordingStream("rerun_example_layer", recording_id=segment_id) as rec:
    rec.save(RRD_PATH)
    rec.set_log_time_enabled(False)
    rec.set_time("log_time", timestamp=second_to_last_timestamp)
    rec.log("/events", rr.AnyValues(flag=True))

dataset.register([Path(RRD_PATH).as_uri()], layer_name="event_layer")

# Step 1: read the marker on its own, which fetches no image chunks
events = (
    dataset
    .filter_contents("/events")
    .reader(index="log_time")
    .select("rerun_segment_id", "log_time")
)

# Step 2: fetch the images only at the marker's timestamps
images_at_events = dataset.filter_contents([
    "/compressed_images/**",
    "/raw_images/**",
]).reader(index="log_time", using_index_values=events)

# vs. one view with the marker and the images, which fetches every image chunk
df_with_flag = dataset.filter_contents([
    "/compressed_images/**",
    "/raw_images/**",
    "/events/**",
]).reader(index="log_time")
df_with_flag.filter(col("/events:flag").is_not_null())