huggingface / huggingface/datatrove

HuggingFaceDatasetWriter reuses already-committed CommitOperationAdd objects when a writer instance is reused across ranks

Open Beginner friendly
#501 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
3.3k
Forks
302
Avg merge
2h 18m
Merged PRs (30d)
2

Description

### Environment

Environment used when finding the issue:
``` bash
datatrove 0.9.0
huggingface_hub 0.36.2
Executor: SlurmPipelineExecutor with tasks_per_job > 1
```
Also tested on the main branch of `datatrove`

# Summary

Using a `HuggingFaceDatasetWriter` while using a `SlurmPipelineExecutor` with `tasks_per_job > 1`, the second (and every subsequent) rank processed within that Slurm array element fails with:
``` bash
ValueError: CommitOperationAdd CommitOperationAdd(path_in_repo='...', path_or_fileobj=b'') has already being committed and cannot be reused. Please create a new CommitOperationAdd object if you want to create a new commit.
```

Root cause:

`SlurmPipelineExecutor.run()` bundles `tasks_per_job` ranks into a single submitted array element and runs them sequentially, in-process, against the same pipeline step instances:

[from lines 219 to 224 of datatrove/executor/slurm.py](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/executor/slurm.py#L219);
``` bash
for rank_to_run in range(*ranks_to_run_range):
...
self._run_for_rank(rank, node_rank=node_rank)
```
[PipelineExecutor._run_for_rank](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/executor/base.py#L102) does not copy `self.pipeline` between iterations of that loop, so every rank in the job reuses the exact same `HuggingFaceDatasetWriter` instance.

[HuggingFaceDatasetWriter.close()](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/pipeline/writers/huggingface.py#L101) accumulates uploaded files onto `self.operations` and commits that list, but never clears it afterwards, **which causes the ValueError** for subsequent ranks:

[summary of the close function of datatrove/pipeline/writers/huggingface.py](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/pipeline/writers/huggingface.py#L101);
``` python
def close(self, rank: int = 0):
filelist = list(self.output_mg.get_open_files().keys())
super().close()
if filelist:
...
self.upload_files(*filelist) # extends self.operations
...
create_commit(self.dataset, operations=self.operations, ...)
```
So when the same instance is reused for the next rank, its `close()` call includes the previous rank's `CommitOperationAdd` objects, which `huggingface_hub` already flagged `_is_committed = True` after that earlier successful commit ([hf_api.py, in `create_commit`](https://github.com/huggingface/huggingface_hub/blob/7d62d95fa01c7e1db9f8680cdf6f96b681f031d7/src/huggingface_hub/hf_api.py#L5215), after the POST succeeds), `create_commit` rejects any operation with `_is_committed` already set, producing the error above.

Note this is specific to `HuggingFaceDatasetWriter`: its sibling [ParquetWriter.close()](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/pipeline/writers/parquet.py#L101) already clears its own per-cycle state on every call (`self._batches.clear()`; `self._writers.clear()`), so it's safe to reuse across ranks. `HuggingFaceDatasetWriter` doesn't extend that same cleanup to `self.operations`.

## Minimal reproduction of the issue

This reproduces the same underlying mechanism directly, without needing a real Slurm cluster — it's exactly what SlurmPipelineExecutor's `tasks_per_job` loop does to a pipeline's writer instance across ranks: reuse the same instance for two sequential write()/close() cycles.

Requires a real HF Hub dataset repository you can write to and a token in scope (e.g. HF_TOKEN environment variable) — see the "Setup" note below.

``` python
import tempfile
from datatrove.data import Document
from datatrove.pipeline.writers import HuggingFaceDatasetWriter
from dotenv import load_dotenv # if you use this you will need to install it; pip install python-dotenv

load_dotenv(".env.test") # sets HF_TOKEN for huggingface_hub's implicit auth

doc_a = Document(text="x", id="1")
doc_b = Document(text="y", id="2")
output_filename = "test.parquet"

with tempfile.TemporaryDirectory() as temp_dir:
writer = HuggingFaceDatasetWriter(
dataset="USERNAME/DATASET-REPO", private=False,
local_working_dir=temp_dir, output_filename=output_filename,
)
with writer:
writer.write(doc_a, rank=0)
# first close() succeeds, commits doc_a

with writer:
writer.write(doc_b, rank=1)
# second close() raises ValueError -- self.operations still holds doc_a's
# already-committed CommitOperationAdd
```

Error output after running the minimal reproduction script

``` bash
2026-08-05 09:03:53.743 | WARNING | datatrove.pipeline.writers.disk_base:__init__:126 - Output filename template 'test.parquet' does not include ${rank}; parallel workers may overwrite outputs.
2026-08-05 09:03:53.952 | INFO | datatrove.pipeline.writers.huggingface:close:102 - Starting upload of 1 files to apmoore1/test-hf-writer
2026-08-05 09:03:54.211 | INFO | datatrove.pipeline.writers.huggingface:upload_files:90 - Uploading 000_test.parquet to the hub...
Processing Files (1 / 1) : 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B, 568B/s
New Data Upload : 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B, 568B/s
...905cz041/000_test.parquet: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B
2026-08-05 09:03:56.140 | INFO | datatrove.pipeline.writers.huggingface:upload_files:92 - Upload of 000_test.parquet to the hub complete!
2026-08-05 09:03:56.546 | INFO | datatrove.pipeline.writers.huggingface:close:102 - Starting upload of 1 files to apmoore1/test-hf-writer
2026-08-05 09:03:56.546 | INFO | datatrove.pipeline.writers.huggingface:upload_files:90 - Uploading 000_test.parquet to the hub...
Processing Files (1 / 1) : 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B, 851B/s
New Data Upload : 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B, 851B/s
...905cz041/000_test.parquet: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 681B / 681B
2026-08-05 09:03:58.253 | INFO | datatrove.pipeline.writers.huggingface:upload_files:92 - Upload of 000_test.parquet to the hub complete!
Traceback (most recent call last):
File "/workspaces/wikipedia-USAS-processing/main.py", line 19, in
with writer:
^^^^^^
File "/workspaces/wikipedia-USAS-processing/.venv/lib/python3.12/site-packages/datatrove/pipeline/writers/disk_base.py", line 164, in __exit__
self.close()
File "/workspaces/wikipedia-USAS-processing/.venv/lib/python3.12/site-packages/datatrove/pipeline/writers/huggingface.py", line 107, in close
create_commit(
File "/workspaces/wikipedia-USAS-processing/.venv/lib/python3.12/site-packages/huggingface_hub/utils/_validators.py", line 114, in _inner_fn
return fn(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^
File "/workspaces/wikipedia-USAS-processing/.venv/lib/python3.12/site-packages/huggingface_hub/hf_api.py", line 1687, in _inner
return fn(self, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/workspaces/wikipedia-USAS-processing/.venv/lib/python3.12/site-packages/huggingface_hub/hf_api.py", line 4197, in create_commit
raise ValueError(
ValueError: CommitOperationAdd CommitOperationAdd(path_in_repo='000_test.parquet', path_or_fileobj=b'') has already being committed and cannot be reused. Please create a new CommitOperationAdd object if you want to create a new commit.
```

### Setup

Create (or reuse) a dataset repository you have write access to (it's auto-created via create_repo(..., exist_ok=True) on first upload if missing), and make sure HF_TOKEN (or HUGGING_FACE_HUB_TOKEN) is set in the environment before running.

Once tested you can delete the HuggingFace dataset repository you used for testing.

## Suggested fix

Clear `self.operations` after a successful commit in [HuggingFaceDatasetWriter.close()](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/pipeline/writers/huggingface.py#L101), mirroring what [ParquetWriter.close()](https://github.com/huggingface/datatrove/blob/9a2b1f9dd85c2178228da418a22f6cc90c013e03/src/datatrove/pipeline/writers/parquet.py#L101) already does for its own state:

``` python
def close(self, rank: int = 0):
filelist = list(self.output_mg.get_open_files().keys())
super().close()
if filelist:
...
self.upload_files(*filelist)
retries = 0
while True:
try:
create_commit(...)
break
except HfHubHTTPError as e:
...
self.operations = []
```
The only change is the addition of `self.operations = []`

This keeps the retry loop's existing correctness (a failed/retryable create_commit never marks operations as committed, so retries still see the full list) and makes a reused instance behave identically to a fresh one for the next rank.

Contributor guide

No contributing guide indexed for this repository

Research direction

Read src/datatrove/pipeline/writers/huggingface.py, especially HuggingFaceDatasetWriter.close(), alongside src/datatrove/executor/slurm.py and src/datatrove/executor/base.py to understand sequential rank reuse. Reproduce the two write()/close() cycles with a writable Hugging Face dataset and HF_TOKEN. Done means a reused writer completes subsequent ranks without reusing committed operations, while the existing commit retry behavior remains intact.

Written by the indexing model from the issue text.

Assessment

Tech stack
huggingface, python
Domain
data-engineering, distributed-systems
Issue type
Bug
Difficulty
2/5
Estimated time
1-3 hours
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
78/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.