From 79a2c2df9e76b26610c350838264b818260adfdd Mon Sep 17 00:00:00 2001 From: shobhitagnihotri69 Date: Sat, 3 Oct 2026 14:51:02 +0530 Subject: [PATCH 1/2] fix(reader): avoid full scan on empty selection or unknown channels (#669) --- src/hflow/reader.py | 8 ++++ tests/test_multichannel.py | 84 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) diff --git a/src/hflow/reader.py b/src/hflow/reader.py index 3a5a9110..b9548710 100644 --- a/src/hflow/reader.py +++ b/src/hflow/reader.py @@ -251,6 +251,10 @@ def iter_batches( batch_max_bytes: int = DEFAULT_BATCH_MAX_BYTES, ) -> Iterator[MessageBatch]: wanted_channel_ids = frozenset(channel_ids) if channel_ids is not None else None + if wanted_channel_ids is not None and not wanted_channel_ids: + return + if topics is not None and not topics: + return if topics is None and wanted_channel_ids is not None: # Constrain the underlying read to the topics the requested # channels live on, so the MCAP reader can skip unrelated streams @@ -258,6 +262,7 @@ def iter_batches( # that would be discarded below. Topic filtering alone is not # exact -- several channels may share one topic -- so the # channel-id filter in the loop still applies. + has_summary = True try: known_channels = self.channels() except ValueError: @@ -267,6 +272,7 @@ def iter_batches( # case -- channels() raises mcap's RecordLengthLimitExceeded, # and the same error surfaces from iter_messages below with # or without this catch, so damage stays loud. + has_summary = False known_channels = {} derived_topics = sorted( { @@ -275,6 +281,8 @@ def iter_batches( if channel_id in known_channels } ) + if has_summary and not derived_topics: + return if derived_topics: topics = derived_topics topics_by_channel_id: dict[int, str] = {} diff --git a/tests/test_multichannel.py b/tests/test_multichannel.py index 2d9f32ff..690a055d 100644 --- a/tests/test_multichannel.py +++ b/tests/test_multichannel.py @@ -303,6 +303,90 @@ def test_channel_id_read_without_a_summary_falls_back_to_a_full_scan( reader.close() +def test_empty_channel_ids_or_topics_skips_reading_entirely( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An empty channel_ids or empty topics selection yields nothing immediately + without calling the underlying MCAP reader or scanning any messages.""" + path = tmp_path / "empty_selection.mcap" + _write_two_topic_mcap(path, [b"t" * 16], [b"c" * 16]) + + reader = open_reader(path) + try: + topics_passed, topics_yielded = _trace_mcap_iter_messages(reader, monkeypatch) + + assert list(reader.iter_batches(channel_ids=[])) == [] + assert topics_passed == [] + assert topics_yielded == [] + + assert list(reader.iter_batches(topics=[])) == [] + assert topics_passed == [] + assert topics_yielded == [] + finally: + reader.close() + + +def test_indexed_read_with_unknown_channel_id_skips_reading_entirely( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """When an indexed file has a summary section and channel_ids names only + channels not present in the summary, iter_batches yields nothing immediately + without calling the underlying MCAP reader or scanning the file.""" + path = tmp_path / "unknown_channel.mcap" + target_channel_id, camera_channel_id = _write_two_topic_mcap( + path, [b"t" * 16 for _ in range(2)], [b"c" * 16 for _ in range(8)] + ) + unknown_channel_id = max(target_channel_id, camera_channel_id) + 9999 + + reader = open_reader(path) + try: + topics_passed, topics_yielded = _trace_mcap_iter_messages(reader, monkeypatch) + + # Querying an unknown channel id on an indexed file does not scan + assert list(reader.iter_batches(channel_ids=[unknown_channel_id])) == [] + assert topics_passed == [] + assert topics_yielded == [] + + # Combining a known channel with an unknown channel still narrows to the known channel's topic + topics_passed.clear() + topics_yielded.clear() + batches = list(reader.iter_batches(channel_ids=[target_channel_id, unknown_channel_id])) + assert len(batches) == 1 + assert batches[0].channel_id == target_channel_id + assert topics_passed == [["/target"]] + assert topics_yielded == ["/target", "/target"] + finally: + reader.close() + + +def test_unindexed_read_with_unknown_channel_id_falls_back_to_scan( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An unindexed file without a summary section falls back to an unconstrained + scan when channel_ids is given, even if the channel id turns out to be absent.""" + path = tmp_path / "unindexed_absent.mcap" + target_channel_id, camera_channel_id = _write_two_topic_mcap( + path, + [b"t" * 16], + [b"c" * 16], + use_chunking=False, + use_statistics=False, + use_summary_offsets=False, + repeat_channels=False, + repeat_schemas=False, + ) + unknown_channel_id = max(target_channel_id, camera_channel_id) + 9999 + + reader = open_reader(path) + try: + topics_passed, topics_yielded = _trace_mcap_iter_messages(reader, monkeypatch) + assert list(reader.iter_batches(channel_ids=[unknown_channel_id])) == [] + assert topics_passed == [None] + assert set(topics_yielded) == {"/target", "/camera"} + finally: + reader.close() + + def test_episode_streams_several_decoded_channels_in_bounded_batches( dual_channel_source: Path, ) -> None: From 37eb9f187738b096a373c4cf272dd47aedcdecb8 Mon Sep 17 00:00:00 2001 From: shobhitagnihotri69 Date: Sat, 3 Oct 2026 19:14:51 +0530 Subject: [PATCH 2/2] fix(reader): fall back to scan when summary has no channel records --- src/hflow/reader.py | 8 +++++--- tests/test_multichannel.py | 27 +++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/hflow/reader.py b/src/hflow/reader.py index b9548710..cd16bade 100644 --- a/src/hflow/reader.py +++ b/src/hflow/reader.py @@ -262,7 +262,6 @@ def iter_batches( # that would be discarded below. Topic filtering alone is not # exact -- several channels may share one topic -- so the # channel-id filter in the loop still applies. - has_summary = True try: known_channels = self.channels() except ValueError: @@ -272,7 +271,6 @@ def iter_batches( # case -- channels() raises mcap's RecordLengthLimitExceeded, # and the same error surfaces from iter_messages below with # or without this catch, so damage stays loud. - has_summary = False known_channels = {} derived_topics = sorted( { @@ -281,7 +279,11 @@ def iter_batches( if channel_id in known_channels } ) - if has_summary and not derived_topics: + # Only return early when the summary actually lists channels. A file + # written with repeat_channels=False has a summary section but no + # repeated channel records in it (known_channels is empty), so fall + # back to scanning instead of dropping real channels. + if known_channels and not derived_topics: return if derived_topics: topics = derived_topics diff --git a/tests/test_multichannel.py b/tests/test_multichannel.py index 690a055d..617fb030 100644 --- a/tests/test_multichannel.py +++ b/tests/test_multichannel.py @@ -387,6 +387,33 @@ def test_unindexed_read_with_unknown_channel_id_falls_back_to_scan( reader.close() +def test_summary_without_repeated_channels_falls_back_to_scan( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A file with a summary section but no repeated channel records (repeat_channels=False) + returns {} for channels(); iter_batches falls back to scanning instead of dropping real channels.""" + path = tmp_path / "summary_no_channels.mcap" + target_channel_id, _ = _write_two_topic_mcap( + path, + [b"t" * 16], + [b"c" * 16], + use_chunking=False, + repeat_channels=False, + ) + + reader = open_reader(path) + try: + assert reader.channels() == {} + topics_passed, topics_yielded = _trace_mcap_iter_messages(reader, monkeypatch) + batches = list(reader.iter_batches(channel_ids=[target_channel_id])) + assert len(batches) == 1 + assert batches[0].channel_id == target_channel_id + assert topics_passed == [None] + assert set(topics_yielded) == {"/target", "/camera"} + finally: + reader.close() + + def test_episode_streams_several_decoded_channels_in_bounded_batches( dual_channel_source: Path, ) -> None: