Episode.channel() reads unrelated MCAP topics before channel filtering

[Kaileshwar16]

Episode.channel() currently selects a channel by channel_id, but that information is not propagated to the underlying MCAP reader as a topic filter.

Because of this, the MCAP reader can yield messages from unrelated topics, and HFlow only removes them afterwards in Python.

For episodes containing large camera/image streams, reading a small state/action channel may therefore cause unnecessary MCAP reads and decoding.

Where the issue comes from

The read path is roughly:

Episode.channel()
    ↓
PythonMcapEpisodeReader.iter_batches(channel_ids=[...])
    ↓
MCAP.iter_messages(topics=None)
    ↓
messages from all topics are yielded
    ↓
HFlow filters by channel.id

1. Episode.channel() only passes the channel ID

In src/hflow/episode.py, Episode.channel() resolves the requested channel and calls the reader using its channel ID:

self._reader.iter_batches(
    channel_ids=[info.channel_id],
)

At this point HFlow already knows the channel's topic through info, but the topic is not passed down.

2. PythonMcapEpisodeReader.iter_batches() only forwards topics to MCAP

In src/hflow/reader.py, the underlying MCAP reader is called roughly as:

self._reader.iter_messages(
    topics=list(topics) if topics is not None else None,
    ...
)

Since Episode.channel() did not provide topics, this becomes:

iter_messages(topics=None)

3. channel_ids are filtered only after MCAP yields the message

The channel-ID check happens afterwards in HFlow:

if wanted_channel_ids is not None and channel.id not in wanted_channel_ids:
    continue

So the channel filtering is correct from a result perspective, but it happens too late to help the underlying MCAP reader avoid unrelated data.

Reproduction

I reproduced this using an MCAP containing:

• /target: 1 message
• /camera: 8 unrelated messages

The messages were made large enough, with a small MCAP chunk size, to keep the streams in separate chunks.

I then instrumented the underlying MCAP reader to record:

1. what topics HFlow passes to iter_messages()
2. which messages MCAP yields before HFlow applies its channel_id filter

Reproduction script:

from collections import Counter
from pathlib import Path

from mcap.writer import CompressionType, Writer
from hflow.episode import Episode

path = Path("/tmp/hflow-channel-filter-probe.mcap")

with path.open("wb") as f:
    writer = Writer(
        f,
        chunk_size=64 * 1024,
        compression=CompressionType.NONE,
    )
    writer.start()

    target = writer.register_channel(
        topic="/target",
        message_encoding="json",
        schema_id=0,
    )

    camera = writer.register_channel(
        topic="/camera",
        message_encoding="json",
        schema_id=0,
    )

    writer.add_message(
        channel_id=target,
        log_time=1,
        publish_time=1,
        data=b"x" * 100_000,
    )

    for i in range(8):
        writer.add_message(
            channel_id=camera,
            log_time=10 + i,
            publish_time=10 + i,
            data=b"y" * 100_000,
        )

    writer.finish()


ep = Episode(path)

hflow_reader = ep._reader
mcap_reader = hflow_reader._reader

original_iter_messages = mcap_reader.iter_messages

calls = []
seen_topics = []


def traced_iter_messages(*args, **kwargs):
    topics = kwargs.get(
        "topics",
        args[0] if args else None,
    )

    calls.append(topics)

    for schema, channel, message in original_iter_messages(*args, **kwargs):
        seen_topics.append(channel.topic)
        yield schema, channel, message


mcap_reader.iter_messages = traced_iter_messages

result = ep.channel("/target")

print("requested channel: /target")
print("messages returned by HFlow:", len(result))
print("topics passed to MCAP:", calls)
print("messages yielded by MCAP before HFlow filter:")
print(Counter(seen_topics))

ep.close()
path.unlink()

Observed output:

requested channel: /target
messages returned by HFlow: 1
topics passed to MCAP: [None]
messages yielded by MCAP before HFlow filter:
Counter({'/camera': 8, '/target': 1})

So although only /target was requested, the underlying MCAP reader yielded all 8 /camera messages as well.

HFlow eventually returns the correct result, but the unrelated messages have already crossed the underlying reader boundary.

Why this matters

This is potentially significant for multimodal robotics episodes, where channel sizes can be very different.

For example:

/joint_states      50 MB
/actions           20 MB
/front_camera       6 GB
/wrist_camera       4 GB
/depth_camera       5 GB

Code such as:

joints = episode.channel("/joint_states")

should ideally constrain the underlying MCAP read to the relevant topic instead of allowing unrelated camera streams to be yielded and then discarded.

The overhead may become more noticeable when several individual channels are accessed independently.

Expected behavior

When HFlow already knows the topic associated with the requested channel, it should propagate that information to the MCAP reader so unrelated topics can be filtered as early as possible.

Conceptually, the read should become:

Episode.channel("/target")
    ↓
iter_batches(
    topics=["/target"],
    channel_ids=[target_id],
)
    ↓
MCAP.iter_messages(topics=["/target"])
    ↓
only messages from the relevant topic reach HFlow
    ↓
channel_id filtering preserves exact channel semantics

Possible fixes

Option 1: Optimize Episode.channel()

Since Episode.channel() already has the channel metadata, it could pass both the topic and channel ID:

self._reader.iter_batches(
    topics=[info.topic],
    channel_ids=[info.channel_id],
)

This is probably the smallest change.

The channel-ID filter should remain because multiple MCAP channels may share the same topic.

Option 2: Optimize PythonMcapEpisodeReader.iter_batches()

A more general solution could be to derive the corresponding topic(s) whenever channel_ids are supplied.

For example, given:

iter_batches(channel_ids=[1, 4])

the reader could resolve those channel IDs to their topics and pass those topics to:

self._reader.iter_messages(topics=[...])

while still applying the existing channel-ID filter afterwards.

This would benefit any caller using channel_ids, rather than only Episode.channel().

Important correctness case

Topic filtering alone should not replace channel-ID filtering.

MCAP can have multiple channels associated with the same topic, so:

topics=["/target"]

may identify a broader set than:

channel_ids=[specific_channel_id]

The optimization should therefore use topic filtering to reduce the underlying read and retain the channel-ID check for exact selection.

Suggested regression test

A regression test could create an MCAP containing two topics in separate chunks:

/target
/camera

Then request only /target through Episode.channel() and instrument/mock the underlying MCAP reader.

The test should verify that MCAP receives:

topics=["/target"]

rather than:

topics=None

A second test would be useful for two channels sharing the same topic to ensure the existing channel-ID semantics are preserved.

Expected outcome

After the fix:

episode.channel("/target")

should still return exactly the same channel data, but the underlying MCAP call should be constrained to the relevant topic instead of reading unrelated topics and discarding them afterwards.