Add commit-aware Hugging Face dataset writer

[hynky1999]

Problem

Refiner can currently write to hf://datasets/... through the generic file-backed sinks, but this goes through huggingface_hub.HfFileSystem / fsspec. In the current Hugging Face Hub implementation, each opened file is staged to a local temp file and close() calls HfApi.upload_file(...), which creates a Hub commit for that single file.

That means a launched Refiner job writing Parquet directly to hf://datasets/... has these semantics:

• each worker writes directly into the final dataset repository
• each closed output file becomes its own Hub commit
• there is no fsspec transaction support for grouping files
• parallel workers can race on the target revision
• failed workers may already have committed partial output
• cleanup reducers can delete non-finalized files, but those deletes are also additional commits

This is acceptable for small/simple writes, but it is not a robust publish path for large or distributed dataset writes.

Proposal

Add a dedicated Hugging Face dataset writer/sink with commit-aware semantics, instead of relying on direct hf://datasets/... fsspec writes for publication.

A likely API shape:

pipeline.write_hf_dataset(
    repo_id="org/name",
    split="train",
    config="default",
    private=True,
)

or an explicit mode on Parquet output:

pipeline.write_parquet(
    "hf://datasets/org/name/data",
    commit=True,
)

The implementation should use a two-phase model:

1. Worker stage writes Parquet/assets to a staging location, not directly to the final repo paths.
2. Final reducer stage inspects finalized workers only.
3. Reducer builds CommitOperationAdd entries for successful outputs.
4. Reducer uses preupload_lfs_files(...) / Xet-compatible Hub upload APIs for large files.
5. Reducer calls one create_commit(...) for the final dataset publication.
6. Optional: create the repo, write/update README or dataset metadata, organize by split/config, and include delete operations for overwrite modes.

This fits Refiner's existing BaseSink.build_reducer() model. The current Parquet/JSONL sinks already use reducer stages for cleanup, and LeRobot uses a reducer stage for metadata finalization.

Reference behavior

Datatrove has a HuggingFaceDatasetWriter that follows this general approach:

• writes Parquet to a local working directory
• creates the dataset repo if needed
• creates CommitOperationAdd entries
• calls preupload_lfs_files(...) as files are completed/rotated
• calls create_commit(...) at close with accumulated operations
• retries known Hugging Face commit race/transient errors

Datatrove still commits per writer/task in distributed settings. Refiner can improve on that by putting the final create_commit(...) in a single reducer stage so a launched job produces one publication commit for all finalized workers.

Design notes

• Keep direct hf://datasets/... writes available for simple file writes, but document their per-file commit semantics.
• Make commit-aware publication explicit at first; automatically changing all hf://datasets/... writes would alter failure behavior and permissions expectations.
• Use parent_commit where possible to detect unexpected concurrent repo changes.
• Consider commit sizing limits: Hugging Face create_commit(...) has practical limits around number of operations and regular-file payload size. Large jobs may need multiple bounded commits or preupload plus final commit batching.
• Include tests around failed worker cleanup/finalization so partial staged output is not published.

Why this matters

This would give Refiner Spark/Beam-style output-committer semantics for Hugging Face datasets: distributed workers stage outputs, and a single finalizer publishes only successful work. That is safer than direct fsspec writes for large datasets and avoids noisy commit history from one commit per output file.