[Data] Estimate MCAP in-memory size from sampled messages, not file size

[marwan116]

Description

MCAPDatasource does not override estimate_inmemory_data_size(), so it inherits FileBasedDatasource's implementation, which sums on-disk file sizes. The value returned is not an approximation of the in-memory size — it is exactly the file size:

file size on disk       14,302 B
Ray's estimate          14,302 B
actual in memory     1,188,720 B

estimate == on-disk size : True
understated by           : 83x

Two effects compound. MCAP chunks are usually compressed, so the file is smaller on disk than in memory. And the datasource materializes per-message columns — topic, timestamps and, when include_metadata is set, the schema — that have no per-message on-disk equivalent at all. Ray Data derives both the read parallelism and the memory it provisions for the read from this number, so both are wrong for every MCAP read.

Fix. Follow the ParquetDatasource precedent: sample a bounded number of files (1% of the input, clamped to [2, 10], evenly spaced so size-ordered inputs don't bias the sample), read at most 100 messages from each, build a block through the datasource's own _message_to_dict + DelegatingBlockBuilder path, and scale the total on-disk size by the measured in-memory to on-disk ratio.

Accuracy after the change (actual / estimate), across shapes beyond the repro:

uniform 1.00 | varying payload 0.99 | 25 files sampled 1.30 | multi-topic 1.04
topic-filtered 1.02 | include_metadata=False 0.97 | include_paths=True 0.99
uncompressed 1.00

0.97x–1.30x, against 83x before.

Related issues

Related to #65640, #65641 and #65787 — none of them duplicates this. #65787 (batch_size) modifies __init__ and _read_stream only; #65640 and #65641 concern block granularity and batched reads. No open issue or PR covers estimate_inmemory_data_size for MCAP.

Additional information

Two MCAP-specific implementation details worth review.

Measure a materialized block rather than undoing compression. The chunk index reports a ~37x decompression ratio, but the true expansion is 83x — the gap is the per-row metadata columns. Building a real block keeps the estimate correct automatically as include_metadata, include_paths and filters change.

Open sampled files seekable. _open_input_source yields a non-seekable BufferedInputStream, which makes make_reader return a NonSeekingReader whose get_summary() rescans the whole file and is single-use. Sampling uses open_input_file so make_reader returns a SeekingReader and reads the summary from the footer index. Message counts come from statistics.message_count; a topic filter is resolved exactly via channel_message_counts.

Filters the summary cannot express. If the per-file iteration finishes before reaching the 100-message cap, it has seen the whole selection, so the measured size is used unscaled rather than extrapolated through the summary's message count. This matters because time_range and message_types are not representable in the summary. Without it, a time_range selecting 1% of a recording is estimated as if the whole file were selected:

              rows    before    after
no filter      360     1.0x     1.0x
first 50%      180     2.0x     2.0x   (exceeds the sample cap, still extrapolates)
first 10%       36    10.0x     1.0x
first  1%        4    90.0x     1.7x

Failure behaviour. Gated on DataContext.decoding_size_estimation. Every failure path falls back to a default ratio of 5 and logs a warning — over-estimating is the safe direction, consistent with the comment in _estimate_files_encoding_ratio, but a silent fallback would quietly restore the behaviour this change exists to replace.

Two bounded approximations remain, both erring high. Messages are sampled as a prefix of each file rather than spread through it, so a recording whose message size changes markedly from start to end is estimated from its opening; the sample across files is spread, so this is a per-file effect. And MCAP_ENCODING_RATIO_ESTIMATE_LOWER_BOUND floors the ratio at 1, which binds under a filter narrow enough that the selected rows are smaller than the file on disk — the 1.7x above.

Drive-by fix. test_read_mcap_invalid_time_range asserts on the regex "start_time must be less than end_time", but the code raises f"start_time ({self.start_time}) must be less than end_time ({self.end_time})". The pattern can never match, and the test fails on clean master. It is not caught in CI because mcap appears in no requirements file in the repo, so pytestmark = skipif(not MCAP_AVAILABLE) skips the entire module. That gap is worth addressing separately and I'm happy to file it; the one-line assertion fix is included here because this PR adds 10 tests to the same file.

Tests

PYTHONPATH=python/ray/data/tests python -u -m pytest \
    python/ray/data/tests/datasource/test_mcap.py -v

28 passed. Baseline on master is 1 failed, 17 passed.

New tests cover: estimate exceeds on-disk size, estimate tracks include_metadata, estimate respects a topic filter, multi-file aggregation, files with no summary statistics, sampling disabled via DataContext, include_paths (which adds a column downstream of _read_stream, so sampling has to add it too), a narrow time_range, log-time ordering (the fixture is written in descending log time, so passing requires the reader's sort rather than the file layout), and binary payload round-trip.

Lint: black, ruff clean on both changed files.

Self-contained repro (needs only ray[data] and mcap)

import os
import tempfile
from pathlib import Path

import ray
from mcap.writer import CompressionType, Writer

SCHEMA = b"uint8[] data\nstring format\n" * 100  # a realistic ros2msg definition

f = Path(tempfile.mkdtemp()) / "d1.mcap"
with f.open("wb") as fh:
    w = Writer(fh, compression=CompressionType.ZSTD)
    w.start(profile="", library="d1")
    sid = w.register_schema(
        name="foxglove.CompressedVideo", encoding="ros2msg", data=SCHEMA
    )
    cid = w.register_channel(topic="/cam", message_encoding="cdr", schema_id=sid)
    for i in range(360):  # 12 s of 30 fps video
        w.add_message(
            cid,
            log_time=i * 33_000_000,
            publish_time=i * 33_000_000,
            data=b"\x00\x00\x00\x01" + bytes([i % 256]) * 504,
            sequence=i,
        )
    w.finish()

ray.init(num_cpus=2, include_dashboard=False, log_to_driver=False, configure_logging=False)

from ray.data._internal.datasource.mcap_datasource import MCAPDatasource

estimate = MCAPDatasource(paths=[str(f)]).estimate_inmemory_data_size()
on_disk = os.path.getsize(f)
in_memory = ray.data.read_mcap(str(f)).materialize().size_bytes()

print(f"  file size on disk   {on_disk:>10,} B")
print(f"  Ray's estimate      {estimate:>10,} B")
print(f"  actual in memory    {in_memory:>10,} B")
print(f"  estimate == on-disk size : {estimate == on_disk}")
print(f"  understated by           : {in_memory / estimate:.0f}x")

---

AI assistance (Claude) was used for this change. I reviewed every changed line and ran the tests locally.

[gemini-code-assist]

Code Review

This pull request implements in-memory size estimation for the MCAP datasource in Ray Data by sampling files and calculating an encoding ratio, addressing the issue where compressed MCAP files are significantly underestimated. It also adds comprehensive unit tests to validate this estimation logic. The review feedback correctly identifies a bug where an estimated size of zero is ignored due to a truthiness check, and suggests using an explicit is not None check instead.

[gemini-code-assist]

Using if in_memory_size: will evaluate to False when in_memory_size is 0 (which is a valid estimated size when a filter selects zero messages). This causes the estimator to ignore valid zero-size estimates and potentially fall back to the default ratio of 5. Changing this to if in_memory_size is not None: correctly handles 0 as a valid size.

Suggested change

	if in_memory_size:
	ratios.append(in_memory_size / file_size)
	if in_memory_size is not None:
	ratios.append(in_memory_size / file_size)

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 7419b3f. Configure here.}