JingsongLi commented on PR #9429: URL: https://github.com/apache/paimon/pull/9429#issuecomment-5467491943
I compared this design with the current PyTorch, Lance, Hugging Face Datasets, WebDataset, and Ray Data implementations. Overall, the direction in this PR looks sound, and I do not see an architectural blocker. The main reasons are: * A plain PyTorch `IterableDataset` must own worker-replica sharding itself (`get_worker_info()`); DDP does not shard inputs, and `DataLoader` rejects a custom sampler for an `IterableDataset`. Therefore, doing process-rank sharding in the dataset and then DataLoader-worker sharding is the correct layer. See the [PyTorch DataLoader source](https://github.com/pytorch/pytorch/blob/main/torch/utils/data/dataloader.py#L359-L369) and [DDP documentation](https://docs.pytorch.org/docs/stable/generated/torch.nn.parallel.DistributedDataParallel.html). * Hugging Face Datasets and WebDataset use the same broad hierarchy: first select the rank/node-local sources, then split those sources among DataLoader workers before opening/decoding them. See [HF distributed.py](https://github.com/huggingface/datasets/blob/main/src/datasets/distributed.py), [HF iterable_dataset.py](https://github.com/huggingface/datasets/blob/main/src/datasets/iterable_dataset.py), and [WebDataset shardlists.py](https://github.com/webdataset/webdataset/blob/main/src/webdataset/shardlists.py). * Keeping `auto_detect_rank=False` by default is appropriate for Paimon. `TableRead.to_torch` receives an already-planned split list and cannot know whether the caller has pre-sharded it. Automatically sharding by default could therefore silently apply sharding twice. Systems such as Ray can safely auto-split because the trainer owns the complete dataset plan and all consumers; that is a different ownership boundary. See [Ray's dataset-shard API](https://docs.ray.io/en/latest/train/api/doc/ray.train.get_dataset_shard.html). * The explicit `sharding_rank` / `sharding_world_size` pair is important for DDP subgroups and hybrid parallelism. Resolving these values in the parent and serializing scalar context into spawned workers is also a good design. One useful comparison is Lance. Lance flattens distributed rank and DataLoader worker id into a global rank and supports fragment- and batch-level sharding. However, based on the current source, its distributed helper only observes a process group initialized in the current process. A spawned DataLoader worker does not retain the parent's initialized process group, while the deprecated explicit `rank/world_size` branch does not additionally shard among worker replicas. Therefore, as a source-level inference, Lance's iterable path does not transparently cover the full `spawn + DDP + num_workers > 0` combination. Paimon's captured parent rank followed by a separate worker split handles this combination more robustly. See [LanceDataset](https://github.com/lance-format/lance/blob/main/python/python/lance/torch/data.py), [Lance distributed helpers](https://github.com/lance-format/lance/blob/main/python/python/lance/torch/dist.py), and [Lance samplers](https://github.com/lance-format/la nce/blob/main/python/python/lance/sampler.py). I would keep the current exact-coverage policy as well: assign every split once, allow uneven rank lengths, and require DDP uneven-input handling. This is preferable to silently duplicating or dropping data. A few non-blocking follow-ups may still be worthwhile: 1. Clarify that `set_epoch()` changes each rank's local buffer-shuffle order, but does not reshuffle split ownership across ranks as PyTorch `DistributedSampler` does. 2. Mention that training loops with additional non-DDP collectives (for example, `SyncBatchNorm`) should use `model.join(throw_on_early_termination=True)`; plain `join()` can only shadow DDP's own collectives. 3. Consider warning when the number of splits is smaller than `world_size * num_workers`, since some ranks/workers will be idle. 4. If equal-step training becomes a requirement later, add an explicit policy such as preserve-all/drop-to-equal/repeat-to-equal rather than changing the current behavior implicitly. Similarly, fragment/split versus batch/weighted sharding can be added as an explicit strategy later without coupling it to rank detection. So my recommendation is to keep the current API boundary and rank -> worker -> reader hierarchy. The comparison with other implementations reinforces this design rather than suggesting that Paimon should copy Lance's defaults. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
