huggingface / huggingface/datatrove
HuggingFaceDatasetWriter reuses already-committed CommitOperationAdd objects when a writer instance is reused across ranks
- 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