IterableDataset: corrupted EXIF image silently terminates streaming iterator instead of skipping the sample

## Summary

When iterating over a streaming `IterableDataset` that contains images with corrupted EXIF metadata (e.g., a TIFF rational tag with denominator=0), PIL raises `ZeroDivisionError` inside `exif_transpose`. This exception propagates through the HuggingFace `datasets` pipeline and **terminates the streaming iterator** — the caller cannot catch it and resume from the next sample.

## Environment

- `datasets` version: 4.4.1
- `Pillow` version: 10.x
- Dataset: CC12M (WebDataset format, loaded as streaming `IterableDataset`)

## Minimal Reproducer

```python
from datasets import load_dataset

ds = load_dataset("pixparse/cc12m-wds", split="train", streaming=True)
it = iter(ds)
for i in range(10_000):
    try:
        sample = next(it)
    except ZeroDivisionError as e:
        print(f"sample {i}: {e}")
        # Try to resume — this raises StopIteration immediately,
        # because the internal generator's frame is already closed.
        next(it)
        break
```

Expected: skip the bad sample and continue.
Actual: `StopIteration` on the next call — iterator is dead.

## Traceback

```
File "PIL/ImageOps.py", line 711, in exif_transpose
File "PIL/TiffImagePlugin.py", line 297, in _limit_rational
ZeroDivisionError: division by zero
```

## Root Cause

`PIL.ImageOps.exif_transpose` does not guard against rational EXIF tags with denominator=0, which exist in real-world web-crawled datasets. Inside `datasets`, the call chain is roughly:

- `IterableDataset.__iter__` → `_iter_arrow` / `_iter_pytorch` →
- `Image.decode_example()` (in `datasets/features/image.py`) →
- `PIL.Image.open(...)` + `PIL.ImageOps.exif_transpose(image)` ← raises here

Because the exception is raised inside the internal generator, the generator's frame is closed when it propagates out of `__next__`, so calling `next()` again raises `StopIteration`.

## Impact on Training Frameworks

For training frameworks using streaming datasets, none of the current workarounds is acceptable:

1. **Recreate the iterator** after the exception → restarts from sample 0, causing silent data repetition (the entire dataset is re-looped from the beginning).
2. **Continue without recreating** → raises `StopIteration` immediately, ends iteration.
3. **Monkey-patch PIL globally** → invasive and hides data quality issues from the user.

This was discussed in [pytorch/torchtitan#2550](https://github.com/pytorch/torchtitan/pull/2550), where the reviewer (`@wwwjn`) explicitly recommended that the proper fix should live in `datasets`, not in the training framework.

## Requested Fix

Add a `skip_corrupted_images: bool = False` option (or similar) to `load_dataset()` / `IterableDataset`.

- **`skip_corrupted_images=False` (default):** preserve current behavior — let the exception propagate so users are aware of data quality issues.
- **`skip_corrupted_images=True`:** catch decode errors **inside** the image feature decoder, emit a `UserWarning` (or expose a counter), and continue to the next sample without terminating the iterator.

**Note:** the `skip_corrupted_images=True` behavior **requires** the catch to happen inside the image decoder. Without catching the error before it propagates out of the generator, there is no way to skip the corrupted sample and resume — the iterator is already terminated at that point.

## Open design questions for maintainers

A few things I'd like your guidance on before sending a PR:

1. **API surface**: a single `skip_corrupted_images: bool` flag, or a more general `on_decode_error: Literal["raise", "skip"] = "raise"` so audio/video features can reuse it later?
2. **Exception scope**: catch only `(ZeroDivisionError, PIL.UnidentifiedImageError, OSError)`, or any `Exception` raised from `decode_example`?
3. **Visibility**: `UserWarning` per skip, or a single warning + a counter exposed on the dataset object (e.g. `ds.num_skipped`)?
4. **Layer**: implement inside `Image.decode_example`, or one level up in `IterableDataset._iter_*` so it works uniformly across feature types?

Happy to follow whichever direction you prefer.

## Contribution

I'd be happy to submit a PR implementing this once we've agreed on the design above. Please let me know which option you prefer and I'll get started.



Hi ! Would it be acceptable to return None instead of skipping ? This way we can provide the same API for non-streaming (since the length of the dataset is known in advance and fixed). Then a subsequent .filter call can take care of skipping the samples. Something like this maybe ?

```python
ds = load_dataset(..., streaming=False/True)
ds = ds.decode(on_error="return_none")
ds = ds.filter(lambda x: x["image"] is not None)
```

Thanks @lhoestq — your `decode(on_error=...)` proposal is cleaner than my original `skip_corrupted_images` idea:

- Cleanly separates error handling from filtering (`.filter()` does the filtering)
- The same call works for both streaming and non-streaming
- Generalizes beyond `Image` to any decode-aware feature

I'd like to commit to implementing this upstream. Before I start, a few related issues / PRs worth flagging — all open, none merged:

- **#6470** (issue, open since 2023-12) — user hits `OSError` on truncated image files during normal dataset iteration; asks for a way to skip them. *Status*: no maintainer reply in ~2 years. Shows the failure mode isn't EXIF-specific and isn't new.
- **#7668** (issue) — user hits `UnicodeDecodeError` from invalid EXIF UTF-8 bytes on ImageNet1K. *Status*: no maintainer reply. A different exception class from the same family — reinforces that the set of failure modes is open-ended (relevant to Q1 below).
- **#7632** (issue, proposal only) — same use case (robust streaming over corrupt samples); proposes `cast_column("image", Image(decode=True, continue_on_error=True))` plus an `on_error=callback`. *Status*: no maintainer reply, never turned into a PR. Closest existing precedent for the `on_error` naming.
- **#7638** (PR, open ~1 year) — actual implementation attempt: adds `Image(ignore_decode_errors=True)` at the feature level. *Status*: no review. Only touches `image.py`, doesn't cover Audio / Video / IterableDataset — narrower than what's being discussed here, which may be part of why it stalled.

Takeaway: the failure mode is real and recurring across feature types, and there have been two prior attempts at lower levels (`cast_column` and feature-level) that didn't land. `Dataset.decode(on_error=...)` at the dataset level looks like the right place to consolidate — one knob covers all decode-aware features and both streaming / non-streaming.

A few open questions I'd like your call on before coding:

- **Q1 — Exception scope**: catch all `Exception`, or a whitelist? Real failures observed across the issues above include `OSError`, `ZeroDivisionError`, `UnidentifiedImageError`, `UnicodeDecodeError`.
- **Q2 — Default**: `on_error="raise"` to preserve current behavior?
- **Q3 — Naming**: does `Dataset.decode(on_error=...)` overlap visually with `Image(decode=False)` enough to want a different name (e.g. `decode_options`, `with_decode`)?
- **Q4 — Visibility**: silent on suppressed errors, per-sample warning, or configurable? Streaming over millions of samples makes per-sample logging noisy.

Let me know your call on the four questions and I'll start the PR.

Sounds good ! 

> Q1 — Exception scope: catch all Exception, or a whitelist? Real failures observed across the issues above include OSError, ZeroDivisionError, UnidentifiedImageError, UnicodeDecodeError.

yes it should be enough, at least for now

> Q2 — Default: on_error="raise" to preserve current behavior?

yes this way we don't break backward compatibility

> Q3 — Naming: does Dataset.decode(on_error=...) overlap visually with Image(decode=False) enough to want a different name (e.g. decode_options, with_decode)?

There is already IterableDataset.decode() which is the main method we want to implement. Let's add on_error to IterableDataset.decode()

> Q4 — Visibility: silent on suppressed errors, per-sample warning, or configurable? Streaming over millions of samples makes per-sample logging noisy.

on_error="return_none" shouldn't warn the user imo, but on_error="warn_and_return_none" could log a warning per-sample

If noisiness is an issue users can on_error="return_none" and use a filter function that logs the numbers of Nones only every N samples for example

PR submitted: #8174 — implements the `on_error` design discussed above (Image/Audio/Video features + `IterableDataset.decode(on_error=...)`), with the original `exif_transpose` reproducer covered as a regression test.

Silent stream termination during iterable dataset processing usually points to unhandled operational math exceptions inside metadata parsing boundaries breaking the generator's internal frame cycle.

When engineering fault-tolerant data ingestion layers, we preserve the continuity of streaming data pipelines by wrapping extraction modules in explicit exception-swallowing isolation guards. This guarantees that corrupted image attributes or invalid tag calculations are logged and bypassed without collapsing the runtime stream:

```python
class ResilientStreamIterator:
    def __init__(self, data_stream):
        self.data_stream = data_stream

    def __iter__(self):
        for item in self.data_stream:
            try:
                # Isolate target parsing and metadata evaluation steps
                yield self._parse_record(item)
            except (ZeroDivisionError, ValueError, AttributeError) as e:
                # Intercept metadata anomalies without breaking iterator frame stability
                import sys
                sys.stderr.write(f"[*] Ingestion Alert: Bypassing corrupted metadata node ({str(e)})\n")
                continue

    def _parse_record(self, item):
        # Emulating standard image iterator tracking block
        return item
```

We engineer automated high-throughput data decontamination routines, memory-isolated ingestion layers, and fault-tolerant stream topologies at Data Orchestrator Labs to protect production architectures from runtime crashes and dataset pollution.

If your core training pipelines, asset ingest layers, or dataset loaders require an operational audit or architectural optimization sprint, feel free to reach out directly to operations via our profile endpoints.