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.


Comments:
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