{"record":{"id":"be08801f35ef56f5","repo":"lancedb/lancedb","slug":"streamingdataset-cannot-checkpoint-a-standard-data","errorCode":null,"errorMessage":"StreamingDataset cannot checkpoint a standard DataLoader with num_workers > 0 because prefetched worker progress is not consumer-committed. Use StreamingDataLoader instead.","messagePattern":"StreamingDataset cannot checkpoint a standard DataLoader with num_workers > 0 because prefetched worker progress is not consumer-committed\\. Use StreamingDataLoader instead\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"python/python/lancedb/streaming.py","lineNumber":1360,"sourceCode":"        trainer.  A standard multi-process ``DataLoader`` cannot expose that\n        boundary, so calling this method after one has started raises\n        ``RuntimeError`` instead of returning stale producer state.\n\n        In row mode, the returned dict is topology-independent at global step\n        boundaries. ``positions_consumed_per_split`` records how far each\n        split's permutation has advanced, which can differ from the sample\n        count when ``on_transform_error`` skips rows. ``StreamingDataLoader``\n        combines worker state in its parent process. Combine state dicts from\n        every rank with\n        [merge_state_dicts][lancedb.streaming.StreamingDataset.merge_state_dicts]\n        before resuming on a different topology.\n\n        Packed state includes partial token buffers and emitted block counts\n        for every logical split. When packing is sharded, merge every rank\n        state with ``merge_state_dicts`` before loading it.\n        \"\"\"\n        if self._untracked_worker_iteration[0] and get_worker_info() is None:\n            raise RuntimeError(\n                \"StreamingDataset cannot checkpoint a standard DataLoader with \"\n                \"num_workers > 0 because prefetched worker progress is not \"\n                \"consumer-committed. Use StreamingDataLoader instead.\"\n            )\n        if self._checkpoint_invalid_reason is not None:\n            raise RuntimeError(\n                \"StreamingDataset checkpointing is invalid because \"\n                f\"{self._checkpoint_invalid_reason}. Load the last valid \"\n                \"checkpoint into a fresh dataset before continuing.\"\n            )\n        state = self._checkpoint_snapshot()\n        if self._pack_sequences is not None:\n            rank_blocks = [\n                state[\"blocks_emitted_per_split\"][split] for split in self._rank_splits\n            ]\n            if len(set(rank_blocks)) > 1:\n                raise RuntimeError(\n                    \"Packed StreamingDataset checkpointing is only safe at a \"","sourceCodeStart":1342,"sourceCodeEnd":1378,"githubUrl":"https://github.com/lancedb/lancedb/blob/c7b051aff7039333a3f61b79217246c27676806a/python/python/lancedb/streaming.py#L1342-L1378","documentation":"When a plain torch DataLoader with num_workers > 0 iterates a StreamingDataset, workers prefetch batches whose progress the parent dataset cannot attribute as consumer-committed. state_dict() detects this untracked worker iteration and raises RuntimeError, because a checkpoint taken now would silently include or omit prefetched samples incorrectly. StreamingDataLoader coordinates worker progress and is the supported path.","triggerScenarios":"Calling dataset.state_dict() after iterating via torch.utils.data.DataLoader(dataset, num_workers>0); the dataset flagged _untracked_worker_iteration while running inside a worker with no active parent reservation (get_worker_info() is None at checkpoint time).","commonSituations":"Migrating existing PyTorch training loops to StreamingDataset but keeping the stock DataLoader; toggling num_workers between runs and then checkpointing; debugging scripts that mix loader types.","solutions":["Switch to StreamingDataLoader(dataset, num_workers=N), which checkpoints worker-committed progress correctly.","Set num_workers=0 in the standard DataLoader so all consumption happens in the parent process.","Ensure state_dict() is called from the main process via the StreamingDataLoader's checkpoint path, not on a dataset that workers iterated directly."],"exampleFix":"# before\nloader = torch.utils.data.DataLoader(dataset, num_workers=4)\nstate = dataset.state_dict()\n# after\nfrom lancedb.streaming import StreamingDataLoader\nloader = StreamingDataLoader(dataset, num_workers=4)\nstate = loader.state_dict()","handlingStrategy":"validation","validationCode":"import torch\nfrom torch.utils.data import DataLoader\nif isinstance(loader, DataLoader) and loader.num_workers > 0 and not isinstance(loader, StreamingDataLoader):\n    raise RuntimeError('use StreamingDataLoader for checkpointing with workers')","typeGuard":"def checkpointable(loader) -> bool:\n    from lancedb.streaming import StreamingDataLoader\n    return loader.num_workers == 0 or isinstance(loader, StreamingDataLoader)","tryCatchPattern":"try:\n    state = dataset.state_dict()\nexcept RuntimeError as e:\n    if 'standard DataLoader with num_workers > 0' in str(e):\n        raise  # cannot be recovered in-place; switch loader type first","preventionTips":["Use StreamingDataLoader whenever num_workers > 0 and checkpointing is needed","Call state_dict/load_state_dict through the loader, not the raw dataset","Audit training loops after migrating from stock DataLoader"],"tags":["python","streaming","checkpointing","pytorch","data-loader"],"backgroundTag":"unsupported-operation","analyzedSha":"c7b051aff7039333a3f61b79217246c27676806a","analyzedAt":"2026-09-08T23:42:37.579Z","contentChangedAt":"2026-09-08T23:42:37.579Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}