[Bugfix][Omni-DuplexEval] Load local Hugging Face dataset layouts in duplex eval loader (#7331)

Signed-off-by: RyanYun09 <RyanYun09@users.noreply.github.com>
Co-authored-by: RyanYun09 <RyanYun09@users.noreply.github.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/benchmarks/duplex/test_omni_duplex_eval.py

@@ -64,6 +64,177 @@ def test_hugging_face_subset_names_are_splits(monkeypatch):
     assert sample.task_type == "correction"
 
 
+def test_local_hf_dataset_directory_is_routed_to_hf_loader(tmp_path, monkeypatch):
+    # An existing local Hugging Face dataset layout (directory with a
+    # ``data/*.parquet`` file) must be loaded through ``datasets.load_dataset``
+    # instead of being parsed as a JSON/JSONL manifest.
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    (data_dir / "train-00000-of-00001.parquet").write_bytes(b"PAR1")
+    calls = []
+
+    def fake_load_dataset(name, **kwargs):
+        calls.append((name, kwargs))
+        return [{"id": "pr", "split": "PR_correction", "question_text": "Correct this."}]
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    samples = load_samples(hf_dir)
+    assert calls == [(str(hf_dir), {})]
+    assert len(samples) == 1
+    assert samples[0].task_type == "correction"
+
+
+def test_local_parquet_file_is_routed_to_hf_loader(tmp_path, monkeypatch):
+    parquet = tmp_path / "samples.parquet"
+    parquet.write_bytes(b"PAR1")
+    calls = []
+
+    def fake_load_dataset(name, **kwargs):
+        calls.append((name, kwargs))
+        return [{"id": "pr", "split": "PR_correction", "question_text": "Correct this."}]
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    samples = load_samples(str(parquet))
+    assert calls == [("parquet", {"data_files": str(parquet)})]
+    assert len(samples) == 1
+    assert samples[0].family == "pr"
+
+
+def test_hf_collapse_to_train_keeps_row_split_identity(tmp_path, monkeypatch):
+    # datasets.load_dataset collapses a config-less local mirror (a bare data/
+    # directory) into a single generic ``train`` split. Rows that already carry
+    # their own split/subset identity must keep it so family inference is not
+    # poisoned by the generic name (regression for the direct-read path).
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    (data_dir / "train-00000-of-00021.parquet").write_bytes(b"PAR1")
+
+    def fake_load_dataset(name, **kwargs):
+        return {
+            "train": [
+                {"id": "pr", "split": "PR_correction", "question_text": "Correct this."},
+                {"id": "pr-subset", "subset": "PR_event_reminder", "question_text": "Remind me."},
+                {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."},
+            ]
+        }
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    samples = load_samples(hf_dir)
+    by_id = {sample.id: sample for sample in samples}
+    assert by_id["pr"].split == "PR_correction"
+    assert by_id["pr"].family == "pr"
+    assert by_id["pr"].task_type == "correction"
+    assert by_id["pr-subset"].split == "PR_event_reminder"
+    assert by_id["pr-subset"].family == "pr"
+    assert by_id["pr-subset"].task_type == "proactive_reminder"
+    assert by_id["rtd"].split == "RTD_OCR"
+    assert by_id["rtd"].family == "rtd"
+
+
+def test_hf_config_layout_stamps_split_name_when_row_has_none(tmp_path, monkeypatch):
+    # With a proper config-per-split mirror the loader still stamps the Hugging
+    # Face split name onto rows that do not identify themselves (unchanged).
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    (data_dir / "RTD_OCR-00000-of-00001.parquet").write_bytes(b"PAR1")
+    (data_dir / "PR_correction-00000-of-00001.parquet").write_bytes(b"PAR1")
+
+    def fake_load_dataset(name, **kwargs):
+        return {
+            "RTD_OCR": [{"id": "rtd", "question_text": "Read this."}],
+            "PR_correction": [{"id": "pr", "question_text": "Correct this."}],
+        }
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    samples = load_samples(hf_dir)
+    by_id = {sample.id: sample for sample in samples}
+    assert by_id["rtd"].split == "RTD_OCR"
+    assert by_id["rtd"].family == "rtd"
+    assert by_id["pr"].split == "PR_correction"
+    assert by_id["pr"].family == "pr"
+    assert by_id["pr"].task_type == "correction"
+
+
+def test_hf_collapse_to_train_without_row_identity_raises_actionable_error(tmp_path, monkeypatch):
+    # A row that carries no split/family/task_type and lands in a generic
+    # collapsed ``train`` split cannot be routed; the loader must raise a clear,
+    # actionable ValueError instead of the bare family_for_split message.
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    (data_dir / "train-00000-of-00021.parquet").write_bytes(b"PAR1")
+
+    def fake_load_dataset(name, **kwargs):
+        return {"train": [{"id": "x", "question_text": "What is this?"}]}
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    with pytest.raises(ValueError, match="Hugging Face splits"):
+        load_samples(hf_dir)
+
+
+def test_non_manifest_file_raises_clear_error(tmp_path):
+    # A plain non-JSON file (e.g. a README) must surface a clear ValueError
+    # listing the accepted inputs instead of a raw JSONDecodeError.
+    bad = tmp_path / "README.md"
+    bad.write_text("# readme\nnot a manifest", encoding="utf-8")
+    with pytest.raises(ValueError, match="JSON/JSONL manifest"):
+        load_samples(str(bad))
+
+
+def _write_parquet(path, rows):
+    pyarrow = pytest.importorskip("pyarrow")
+    pyarrow.parquet.write_table(pyarrow.Table.from_pylist(rows), str(path))
+
+
+def test_local_parquet_split_filters_rows_by_identity(tmp_path):
+    # Regression: a single ``.parquet`` file is exposed by ``load_dataset`` as a
+    # generic ``train`` physical split, so ``--split RTD_OCR`` used to raise
+    # ``Unknown split "RTD_OCR"``. The requested split must instead filter the
+    # rows by their preserved identity.
+    pytest.importorskip("datasets")
+    parquet = tmp_path / "samples.parquet"
+    _write_parquet(
+        parquet,
+        [
+            {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."},
+            {"id": "pr", "split": "PR_correction", "question_text": "Correct this."},
+        ],
+    )
+    assert {sample.id for sample in load_samples(str(parquet))} == {"rtd", "pr"}
+    rtd = load_samples(str(parquet), split="RTD_OCR")
+    assert [sample.id for sample in rtd] == ["rtd"]
+    assert rtd[0].family == "rtd"
+    pr = load_samples(str(parquet), split="PR_correction")
+    assert [sample.id for sample in pr] == ["pr"]
+    assert pr[0].family == "pr"
+    assert pr[0].task_type == "correction"
+
+
+def test_local_directory_named_splits_resolve_without_row_identity(tmp_path):
+    # Regression: a single-configuration mirror whose data files are named after
+    # the benchmark splits must resolve each split for real (no fake loader) and
+    # without any per-row identity.
+    pytest.importorskip("datasets")
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    _write_parquet(data_dir / "RTD_OCR-00000-of-00001.parquet", [{"id": "rtd", "question_text": "Read this."}])
+    _write_parquet(
+        data_dir / "PR_correction-00000-of-00001.parquet",
+        [{"id": "pr", "question_text": "Correct this."}],
+    )
+    by_id = {sample.id: sample for sample in load_samples(hf_dir)}
+    assert by_id["rtd"].split == "RTD_OCR"
+    assert by_id["rtd"].family == "rtd"
+    assert by_id["pr"].split == "PR_correction"
+    assert by_id["pr"].family == "pr"
+    assert by_id["pr"].task_type == "correction"
+    assert [sample.id for sample in load_samples(hf_dir, split="RTD_OCR")] == ["rtd"]
+
+
 def test_response_aliases_and_clock_guard():
     assert split_text("One. Two!") == ["One.", "Two!"]
     assert normalize_response_items({"chunks": [{"text": "x", "current_time": 800}]}) == [
@@ -141,3 +312,113 @@ def test_frame_sample_content_passes_frames(tmp_path, monkeypatch):
         judge_video_mode="frame-sample",
     )
     assert score["content"]["frame_count"] == 2
+
+
+def test_local_directory_collapsed_train_with_row_identity_and_split_filter(tmp_path):
+    # Reviewer-pointed gap: a collapsed local mirror (data/train-*.parquet)
+    # whose rows carry their own split identity must correctly filter by
+    # --split RTD_OCR.
+    pytest.importorskip("datasets")
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    _write_parquet(
+        data_dir / "train-00000-of-00001.parquet",
+        [
+            {"id": "rtd", "split": "RTD_OCR", "question_text": "Read this."},
+            {"id": "pr", "split": "PR_correction", "question_text": "Correct this."},
+        ],
+    )
+    all_samples = load_samples(hf_dir)
+    assert {s.id for s in all_samples} == {"rtd", "pr"}
+    rtd = load_samples(hf_dir, split="RTD_OCR")
+    assert [s.id for s in rtd] == ["rtd"]
+    assert rtd[0].family == "rtd"
+
+
+def test_local_directory_mistyped_split_raises_clear_error(tmp_path):
+    # A named-shard mirror with --split RTD_TYPO must raise ValueError
+    # listing the observed splits instead of silently returning nothing.
+    pytest.importorskip("datasets")
+    hf_dir = tmp_path / "Omni-DuplexEval"
+    data_dir = hf_dir / "data"
+    data_dir.mkdir(parents=True)
+    _write_parquet(
+        data_dir / "RTD_OCR-00000-of-00001.parquet",
+        [{"id": "rtd", "question_text": "Read this."}],
+    )
+    _write_parquet(
+        data_dir / "PR_correction-00000-of-00001.parquet",
+        [{"id": "pr", "question_text": "Correct this."}],
+    )
+    with pytest.raises(ValueError) as exc_info:
+        load_samples(hf_dir, split="RTD_TYPO")
+    message = str(exc_info.value)
+    assert "matched zero" in message
+    assert "RTD_OCR" in message or "PR_correction" in message
+
+
+def test_manifest_without_split_identity_split_override_still_works(tmp_path):
+    # Manifest rows without split/subset/config identity must keep the
+    # base override behaviour: --split RTD_OCR stamps the split so rows
+    # are kept instead of rejected (regression test for P2).
+    manifest = tmp_path / "samples.json"
+    manifest.write_text(
+        json.dumps([{"id": "a", "video": "clip.mp4"}, {"id": "b", "video": "clip2.mp4"}]),
+        encoding="utf-8",
+    )
+    samples = load_samples(manifest, split="RTD_OCR")
+    assert len(samples) == 2
+    assert all(s.split == "RTD_OCR" for s in samples)
+    assert all(s.family == "rtd" for s in samples)
+
+
+def test_iterable_without_split_identity_split_override_still_works():
+    # Iterable rows without split/subset/config identity must keep the
+    # base override behaviour (regression test for F5/P2).
+    data = [{"id": "a", "video": "clip.mp4"}, {"id": "b", "video": "clip2.mp4"}]
+    samples = load_samples(data, split="RTD_OCR")
+    assert len(samples) == 2
+    assert all(s.split == "RTD_OCR" for s in samples)
+    assert all(s.family == "rtd" for s in samples)
+
+
+def test_audio_feature_decode_false_preserves_bytes_key(tmp_path):
+    # When a parquet file carries an Audio feature column, the loader
+    # must recast it with decode=False so rows contain the ``bytes`` key
+    # instead of ``array``/``sampling_rate``.
+    pytest.importorskip("datasets")
+    pytest.importorskip("pyarrow")
+    import datasets
+
+    audio_bytes = b"\x00\x01\x02"
+    ds = datasets.Dataset.from_dict(
+        {
+            "id": ["a"],
+            "question_text": ["Hello"],
+            "task_type": ["pr_correction"],
+            "audio": [{"bytes": audio_bytes, "path": None}],
+        },
+        features=datasets.Features(
+            {
+                "id": datasets.Value("string"),
+                "question_text": datasets.Value("string"),
+                "task_type": datasets.Value("string"),
+                "audio": datasets.Audio(sampling_rate=16000),
+            }
+        ),
+    )
+    parquet = tmp_path / "audio_sample.parquet"
+    ds.to_parquet(str(parquet))
+
+    samples = load_samples(str(parquet))
+    assert len(samples) == 1
+    raw = samples[0].raw
+    assert raw is not None
+    audio_val = raw.get("audio")
+    assert audio_val is not None
+    # The Audio feature was decoded with decode=False, so the struct
+    # keeps the ``bytes`` key that materialize_media needs.
+    assert isinstance(audio_val, dict) and "bytes" in audio_val
+    assert audio_val["bytes"] == audio_bytes
+    assert "path" in audio_val

tests/benchmarks/duplex/test_omni_duplex_eval_media.py

@@ -0,0 +1,67 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from __future__ import annotations
+
+import pytest
+
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_media import materialize_media
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_dict_with_bytes_materializes_a_real_file(tmp_path):
+    # A Hugging Face Audio/Video feature value {"bytes": ..., "path": ...} used
+    # to short-circuit on ``path`` (the artifact name inside the dataset, which
+    # does not exist on disk), so downstream ffmpeg saw a missing file. Inline
+    # bytes must win and be written to a real file whose content equals bytes.
+    payload = b"\x00\xff\x01RIFFfake-content"
+    resolved = materialize_media({"bytes": payload, "path": "question_audio.wav"}, tmp_path, "s1_question", ".wav")
+    assert resolved.is_file()
+    assert resolved.read_bytes() == payload
+    assert resolved.parent == tmp_path
+    assert resolved.name == "s1_question.wav"
+
+
+def test_dict_extension_comes_from_path_safely(tmp_path):
+    payload = b"flac-ish-bytes"
+    resolved = materialize_media({"bytes": payload, "path": "audio/clip.FLAC"}, tmp_path, "s1", ".wav")
+    assert resolved.name == "s1.flac"
+    assert resolved.read_bytes() == payload
+    # A path with no usable extension falls back to the caller's suffix.
+    resolved = materialize_media({"bytes": payload, "path": "audio/clip"}, tmp_path, "s2", ".wav")
+    assert resolved.name == "s2.wav"
+    assert resolved.read_bytes() == payload
+
+
+def test_materialization_is_idempotent_and_updated(tmp_path):
+    payload = b"abc"
+    first = materialize_media({"bytes": payload, "path": "q.wav"}, tmp_path, "s", ".wav")
+    second = materialize_media({"bytes": payload, "path": "q.wav"}, tmp_path, "s", ".wav")
+    assert first == second
+    assert first.read_bytes() == payload
+    replacement = b"xyz"
+    third = materialize_media({"bytes": replacement, "path": "q.wav"}, tmp_path, "s", ".wav")
+    assert third == first
+    assert third.read_bytes() == replacement
+
+
+def test_str_path_and_path_only_mapping_returned_as_is(tmp_path):
+    existing = tmp_path / "clip.mp4"
+    existing.write_bytes(b"video")
+    assert materialize_media(str(existing), tmp_path, "s", ".mp4") == existing
+    assert materialize_media({"path": str(existing)}, tmp_path, "s", ".mp4") == existing
+
+
+def test_raw_bytes_value_is_written(tmp_path):
+    payload = b"raw-media"
+    resolved = materialize_media(payload, tmp_path, "r", ".bin")
+    assert resolved.name == "r.bin"
+    assert resolved.read_bytes() == payload
+
+
+def test_unresolvable_value_raises_clear_error(tmp_path):
+    with pytest.raises(ValueError, match="cannot materialize"):
+        materialize_media(None, tmp_path, "s", ".wav")
+    with pytest.raises(ValueError, match="cannot materialize"):
+        materialize_media({"path": None}, tmp_path, "s", ".wav")

vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py

@@ -3,15 +3,36 @@
 
 """Dataset and split normalization for Omni-DuplexEval.
 
-The loader accepts a Hugging Face dataset id, a JSON/JSONL manifest, or an
+The loader accepts a Hugging Face dataset id, a local Hugging Face dataset
+layout (a directory or a ``.parquet`` file), a JSON/JSONL manifest, or an
 already materialized iterable.  Media is deliberately resolved at use time so
 the benchmark remains usable in air-gapped environments.
+
+Local directory mirrors are expected to use a *single* Hugging Face
+configuration whose data files are named after the benchmark splits
+(``data/<SPLIT>-00000-of-00001.parquet``, with ``<SPLIT>`` one of the RTD_*/PR_*
+names), so ``datasets.load_dataset`` exposes one physical split per benchmark
+split.  A single ``.parquet`` file and a mirror that Hugging Face collapses into
+a generic ``train`` split are also accepted, but then the requested ``split`` is
+applied as a filter on each row's preserved ``split``/``subset``/``config``
+identity instead of as a physical split name.  Rows that identify neither their
+split nor their family/task raise a clear error that explains the required
+layout.
+
+A requested ``split`` that matches zero rows (for example a mistyped name) is
+reported loudly: :func:`load_samples` raises a ``ValueError`` that lists the
+split identities actually observed in the loaded rows instead of silently
+returning an empty selection.  Rows coming from a JSON/JSONL manifest or an
+already materialized iterable that carry no split/subset/config identity keep
+the base override behavior: ``--split`` is stamped onto them so they are kept
+rather than rejected.
 """
 
 from __future__ import annotations
 
+import importlib
 import json
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -68,6 +89,23 @@ def family_for_split(split: str, task_type: str | None = None) -> str:
     raise ValueError(f"cannot infer benchmark family from split {split!r}")
 
 
+def _identity_error(split: str, sample_id: str, *, family: bool) -> str:
+    """Build an actionable error for a row whose family/task cannot be derived."""
+    subject = "family" if family else "proactive-reminder task type"
+    label = f" for sample {sample_id!r}" if sample_id else ""
+    detail = (
+        "the split is not an RTD_*/PR_* name and no per-row 'task_type' is present"
+        if family
+        else "no per-row 'task_type' is present"
+    )
+    return (
+        f"cannot infer the Omni-DuplexEval {subject}{label} from split {split!r}: {detail}. "
+        "Load a local Omni-DuplexEval mirror whose data files are named after the "
+        "Hugging Face splits (RTD_*/PR_*) or add per-row "
+        "'family'/'task_type'/'split' columns."
+    )
+
+
 def _value(row: dict[str, Any], *keys: str, default: Any = None) -> Any:
     for key in keys:
         if key in row and row[key] is not None:
@@ -100,14 +138,27 @@ class DuplexSample:
     raw: dict[str, Any] | None = None
 
     @classmethod
-    def from_row(cls, row: dict[str, Any], *, split: str | None = None, media_root: Path | None = None) -> DuplexSample:
-        chosen_split = str(split or _value(row, "split", "subset", "config", default=""))
+    def from_row(cls, row: dict[str, Any], *, media_root: Path | None = None) -> DuplexSample:
+        chosen_split = str(_value(row, "split", "subset", "config", default=""))
+        sample_id = str(_value(row, "id", "sample_id", "uid", "name", default=""))
         task_value = _value(row, "task_type", "task", "type")
-        family = str(_value(row, "family", default="") or family_for_split(chosen_split, task_value))
+        family_value = _value(row, "family", default="")
+        if family_value:
+            family = str(family_value)
+        else:
+            try:
+                family = family_for_split(chosen_split, task_value)
+            except ValueError as exc:
+                raise ValueError(_identity_error(chosen_split, sample_id, family=True)) from exc
         task = None
         if family == "pr":
-            task = canonical_task_type(str(task_value)) if task_value else task_type_for_split(chosen_split)
-        sample_id = str(_value(row, "id", "sample_id", "uid", "name", default=""))
+            if task_value:
+                task = canonical_task_type(str(task_value))
+            else:
+                try:
+                    task = task_type_for_split(chosen_split)
+                except ValueError as exc:
+                    raise ValueError(_identity_error(chosen_split, sample_id, family=False)) from exc
         video = _value(row, "video", "video_path", "video_file")
         audio = _value(row, "question_audio", "audio", "question_wav")
         if media_root:
@@ -148,6 +199,150 @@ def _read_manifest(path: Path) -> list[dict[str, Any]]:
     return payload
 
 
+def _has_split_identity(row: dict[str, Any]) -> bool:
+    """Whether a row already identifies the benchmark split it belongs to."""
+    return any(row.get(key) for key in ("split", "subset", "config"))
+
+
+def _stamp_split(rows: Iterable[dict[str, Any]], name: str) -> list[dict[str, Any]]:
+    """Stamp a physical Hugging Face split name onto rows that lack identity.
+
+    Hugging Face collapses a config-less local mirror (a bare ``data/``
+    directory) or a single ``.parquet`` file into a single generic split
+    (``train``).  A row's own split identity (``split``/``subset``/``config``)
+    must win so family inference is not poisoned by that generic name; the
+    physical split name is only stamped when the row does not identify itself.
+    """
+    stamped: list[dict[str, Any]] = []
+    for row in rows:
+        item = dict(row)
+        if name and not _has_split_identity(item):
+            item["split"] = name
+        stamped.append(item)
+    return stamped
+
+
+def _override_split(rows: list[dict[str, Any]], split: str | None) -> list[dict[str, Any]]:
+    """Apply the base split-override semantics to manifest/iterable rows.
+
+    Rows that lack split/subset/config identity are stamped with the requested
+    split so they are kept, matching the pre-override behavior; rows that
+    already carry identity are left untouched so the requested split stays a
+    filter. Copies are returned, never mutating the caller's original dicts.
+    """
+    if not split or split == "all":
+        return rows
+    return [row if _has_split_identity(row) else {**row, "split": split} for row in rows]
+
+
+def _recast_media_features(loaded: Any) -> Any:
+    """Recast Audio/Video features to decode=False so rows carry {bytes, path}.
+
+    Hugging Face ``Audio``/``Video`` features default to ``decode=True``,
+    producing ``{"array", "path", "sampling_rate"}`` dicts without the
+    ``bytes`` key that :func:`materialize_media` needs. ``decode=False`` keeps
+    the raw parquet struct instead. Must run before row iteration
+    (``cast_column`` is lazy; decoding happens at row access).
+    """
+    try:
+        features_mod = importlib.import_module("datasets.features")
+    except ImportError:
+        return loaded
+    audio_cls = getattr(features_mod, "Audio", None)
+    video_cls = getattr(features_mod, "Video", None)
+    if audio_cls is None:
+        return loaded
+
+    if isinstance(loaded, dict):
+        # A DatasetDict -- or a plain dict returned by a fake loader whose
+        # values are lists -- recurses one level and is returned untouched by
+        # the features guard below.
+        return {name: _recast_media_features(table) for name, table in loaded.items()}
+
+    feats = getattr(loaded, "features", None)
+    if feats is None:
+        return loaded
+    for column, feature in feats.items():
+        if isinstance(feature, audio_cls):
+            loaded = loaded.cast_column(column, audio_cls(decode=False))
+        elif video_cls is not None and isinstance(feature, video_cls):
+            loaded = loaded.cast_column(column, video_cls(decode=False))
+    return loaded
+
+
+def _read_hf(load_dataset: Callable[..., Any], dataset: str, want: str | None) -> Any:
+    """Call ``datasets.load_dataset`` for a local layout or a remote id.
+
+    The physical split is only requested when one was asked for; ``datasets``
+    then returns a single ``Dataset`` (an iterable of rows) instead of a
+    ``DatasetDict`` (a mapping of split name to such an iterable). The result is
+    normalized by :func:`_rows_from_hf`.
+    """
+    if Path(dataset).suffix.lower() == ".parquet":
+        # A single parquet file is always exposed as one generic ``train``
+        # split; the logical benchmark split is filtered from row identity
+        # afterwards. The ``want`` branch was unreachable (the only call site
+        # passes ``None``) and implied physical split support that does not
+        # exist, so it was removed.
+        return load_dataset("parquet", data_files=dataset)
+    if want:
+        return load_dataset(dataset, split=want)
+    return load_dataset(dataset)
+
+
+def _rows_from_hf(dataset: str, *, split: str | None) -> list[dict[str, Any]]:
+    """Load rows from a Hugging Face dataset id or a local dataset layout.
+
+    ``datasets.load_dataset`` accepts a remote id, a local dataset directory
+    (``data/<SPLIT>-00000-of-00001.parquet``) and, via the ``parquet`` builder,
+    a single ``.parquet`` file, which keeps the benchmark usable offline with a
+    locally mirrored copy of the dataset.
+
+    The logical benchmark split (``RTD_OCR``, ``PR_correction``, ...) is not
+    forwarded blindly as a physical Hugging Face split: a single ``.parquet``
+    file and a collapsed local mirror only expose a generic ``train`` split, so
+    passing the logical name there raises ``Unknown split``.  The physical
+    dataset is loaded instead and the requested split is applied later, on the
+    preserved per-row identity (see :func:`load_samples`).
+    """
+    try:
+        from datasets import load_dataset
+    except ImportError as exc:
+        raise RuntimeError(
+            "datasets is required to load a Hugging Face dataset id or a local "
+            "dataset directory/parquet file; use a local JSON/JSONL manifest instead"
+        ) from exc
+
+    requested = split if split and split != "all" else None
+    is_local = Path(dataset).exists()
+    if Path(dataset).suffix.lower() == ".parquet":
+        # A single parquet file is always exposed as one generic ``train``
+        # split; the logical benchmark split is filtered from row identity
+        # afterwards rather than forwarded as a physical split name.
+        loaded = _read_hf(load_dataset, dataset, None)
+    else:
+        try:
+            loaded = _read_hf(load_dataset, dataset, requested)
+        except ValueError as exc:
+            # A collapsed *local* mirror (bare ``data/`` directory) does not
+            # expose the logical split as a physical Hugging Face split; fall
+            # back to the whole physical dataset and filter by row identity.
+            # A remote id keeps the original error so a mistyped split still
+            # fails loudly instead of silently selecting nothing.
+            if not is_local or requested is None or "Unknown split" not in str(exc):
+                raise
+            loaded = _read_hf(load_dataset, dataset, None)
+            requested = None
+
+    loaded = _recast_media_features(loaded)
+    if isinstance(loaded, dict):
+        rows: list[dict[str, Any]] = []
+        for name, table in loaded.items():
+            rows.extend(_stamp_split((dict(row) for row in table), name))
+        return rows
+    return _stamp_split((dict(row) for row in loaded), requested or "")
+
+
 def load_samples(
     dataset: str | Path | Iterable[dict[str, Any]] = DEFAULT_DATASET,
     *,
@@ -157,37 +352,54 @@ def load_samples(
     limit: int | None = None,
     ids: Iterable[str] | None = None,
 ) -> list[DuplexSample]:
-    if isinstance(dataset, (str, Path)) and Path(str(dataset)).exists():
-        rows = _read_manifest(Path(str(dataset)))
-    elif not isinstance(dataset, (str, Path)):
-        rows = list(dataset)
-    else:
-        try:
-            from datasets import load_dataset
-        except ImportError as exc:
-            raise RuntimeError("datasets is required for Hugging Face loading; use a local manifest instead") from exc
-        kwargs: dict[str, Any] = {}
-        if split and split != "all":
-            kwargs["split"] = split
-        loaded = load_dataset(str(dataset), **kwargs)
-        if isinstance(loaded, dict):
-            rows = []
-            for name, table in loaded.items():
-                rows.extend({**dict(row), "split": name} for row in table)
+    if isinstance(dataset, str | Path):
+        path = Path(str(dataset))
+        if path.exists():
+            # A directory or a ``.parquet`` file is a Hugging Face dataset
+            # layout (e.g. a locally mirrored Omni-DuplexEval snapshot); only
+            # single JSON/JSONL files are parsed as manifests.  Routing by path
+            # shape instead of mere existence lets local Hugging Face datasets
+            # load through ``datasets.load_dataset``.
+            if path.is_dir() or path.suffix.lower() == ".parquet":
+                rows = _rows_from_hf(str(path), split=split)
+            else:
+                try:
+                    rows = _override_split(_read_manifest(path), split)
+                except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+                    raise ValueError(
+                        f"cannot load dataset from {str(path)!r}: expected a "
+                        "JSON/JSONL manifest file, a local Hugging Face dataset "
+                        "directory, or a single .parquet file"
+                    ) from exc
         else:
-            rows = [dict(row) for row in loaded]
+            rows = _rows_from_hf(str(dataset), split=split)
+    else:
+        rows = _override_split(list(dataset), split)
     wanted = set(str(item) for item in ids) if ids else None
     root = Path(media_root).expanduser() if media_root else None
     result = []
+    split_matched = 0
     for row in rows:
-        sample = DuplexSample.from_row(row, split=None if split in (None, "all") else split, media_root=root)
+        # The requested split is a *filter* over each row's own split identity;
+        # never pass it as an override, which would relabel every row as the
+        # requested split (turning a collapsed local dataset into a full match
+        # instead of a filter).
+        sample = DuplexSample.from_row(row, media_root=root)
         if split and split != "all" and sample.split != split:
             continue
+        split_matched += 1
         if family != "all" and sample.family != family:
             continue
         if wanted is not None and sample.id not in wanted:
             continue
         result.append(sample)
+    if split and split != "all" and split_matched == 0 and rows:
+        observed = sorted({str(_value(row, "split", "subset", "config", default="")) or "<none>" for row in rows})
+        raise ValueError(
+            f"--split {split!r} matched zero samples. "
+            f"Split identities observed in the loaded rows: {observed}. "
+            "Check the split name for a typo."
+        )
     if limit is not None:
         result = result[: max(0, limit)]
     return result

vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py

@@ -5,7 +5,9 @@
 
 from __future__ import annotations
 
+import os
 import subprocess
+import tempfile
 import wave
 from collections.abc import Iterator
 from pathlib import Path
@@ -20,20 +22,57 @@ def _run_ffmpeg(args: list[str]) -> bytes:
 
 
 def materialize_media(value: Any, output_dir: str | Path, stem: str, suffix: str) -> Path:
-    """Resolve common Hugging Face media values to a local file."""
+    """Resolve common Hugging Face media values to a local file.
+
+    ``str``/``Path`` values are returned as-is. Inline byte payloads (a
+    ``bytes``/``bytearray`` or a Hugging Face ``Audio``/``Video``/``Image``
+    feature mapping ``{"bytes": ..., "path": ...}``) are written to a real file
+    under ``output_dir`` and that existing path is returned. For such mappings
+    the ``path`` key is the original artifact name inside the dataset, which is
+    usually *not* present on the local disk, so it is only used to pick a
+    content extension and never treated as an existing file. A mapping or
+    object carrying only a ``path`` (no inline ``bytes``) is returned as an
+    already-materialized file reference.
+    """
     if isinstance(value, str | Path):
         return Path(value)
     path = value.get("path") if isinstance(value, dict) else getattr(value, "path", None)
-    if path:
-        return Path(path)
-    payload = value.get("bytes") if isinstance(value, dict) else None
+    payload = value.get("bytes") if isinstance(value, dict) else getattr(value, "bytes", None)
     if payload is None and isinstance(value, bytes | bytearray):
         payload = value
-    if payload is None:
-        raise ValueError(f"cannot materialize media value for {stem!r}")
-    destination = Path(output_dir) / f"{stem}{suffix}"
-    destination.parent.mkdir(parents=True, exist_ok=True)
-    destination.write_bytes(bytes(payload))
+    if payload is not None:
+        return _write_media_bytes(payload, output_dir, stem, _media_extension(path, suffix))
+    if path:
+        return Path(path)
+    raise ValueError(f"cannot materialize media value for {stem!r}")
+
+
+def _media_extension(path: str | Path | None, fallback: str) -> str:
+    """Pick a safe lower-case file extension from ``path`` when available."""
+    extension = Path(str(path)).suffix.lower() if path else ""
+    if extension.startswith(".") and len(extension) <= 8 and extension[1:].isalnum():
+        return extension
+    return fallback
+
+
+def _write_media_bytes(payload: bytes | bytearray, output_dir: str | Path, stem: str, extension: str) -> Path:
+    data = bytes(payload)
+    root = Path(output_dir)
+    root.mkdir(parents=True, exist_ok=True)
+    destination = root / f"{stem}{extension}"
+    if destination.exists() and destination.stat().st_size == len(data) and destination.read_bytes() == data:
+        return destination
+    fd, temporary = tempfile.mkstemp(dir=root, prefix=f".{destination.name}.", suffix=".part")
+    try:
+        with os.fdopen(fd, "wb") as handle:
+            handle.write(data)
+        os.replace(temporary, destination)
+    except OSError:
+        try:
+            os.unlink(temporary)
+        except OSError:
+            pass
+        raise
     return destination
 
 

vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py

@@ -28,8 +28,28 @@ from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase
 
 
 def _common(parser: argparse.ArgumentParser) -> None:
-    parser.add_argument("--dataset", default=DEFAULT_DATASET)
-    parser.add_argument("--split", default="all")
+    parser.add_argument(
+        "--dataset",
+        default=DEFAULT_DATASET,
+        help=(
+            "Hugging Face dataset id, a JSON/JSONL manifest, or a local Hugging Face "
+            "dataset mirror. A local directory must use a single configuration whose "
+            "data files are named after the RTD_*/PR_* splits "
+            "(data/<SPLIT>-00000-of-00001.parquet); a single .parquet file is also "
+            "accepted. --split filters the rows by their preserved split identity, so "
+            "rows without identity fail loudly."
+        ),
+    )
+    parser.add_argument(
+        "--split",
+        default="all",
+        help=(
+            "Restrict to one split (e.g. RTD_OCR) or 'all'. A mistyped split name "
+            "raises a clear error listing the splits actually observed in the data. "
+            "Rows from a manifest or an iterable without split/subset/config identity "
+            "keep the requested split as an override rather than being rejected."
+        ),
+    )
     parser.add_argument("--family", choices=("all", "rtd", "pr"), default="all")
     parser.add_argument("--media-root")
     parser.add_argument("--limit", type=int)
@@ -100,6 +120,8 @@ def run(args: argparse.Namespace) -> int:
         limit=args.limit,
         ids=args.ids,
     )
+    if not samples:
+        raise ValueError("no samples selected; check --dataset/--split/--family/--ids/--limit")
     if args.action == "generate":
         if args.concurrency < 1:
             raise ValueError("--concurrency must be at least 1")