feat(get): --workers streams to S3 across compute nodes (0.25.0)

docs/commands.md

@@ -206,9 +206,10 @@ Download DVC-tracked data without creating tracking files.
 ```bash
 dt get <repository> <path> [-o <output>] [--rev REV] [--link TYPES] [-j N] [-f] [-v]
 dt get <repository> --csv <file> [--path-col COL] [-o <dir>]
+dt get <repository> --csv <file> -o s3://<bucket>/<prefix>/ [-w N] [--dest-profile NAME]
 ```
 
-The `dvc get` counterpart to `dt import`: materialises data and writes no `.dvc` file, for handing a subset of a dataset to someone outside the group. A path may name a subdirectory *inside* a tracked directory output, so you can take part of a large dataset without transferring all of it. With `--csv` the source is cloned once and every row resolved against that clone, rather than re-cloning per path. [Full documentation →](get.md)
+The `dvc get` counterpart to `dt import`: materialises data and writes no `.dvc` file, for handing a subset of a dataset to someone outside the group. A path may name a subdirectory *inside* a tracked directory output, so you can take part of a large dataset without transferring all of it. With `--csv` the source is cloned once and every row resolved against that clone, rather than re-cloning per path. `-o s3://…` streams from the source's remote straight into object storage, and `-w N` spreads that stream across N compute nodes via qxub — network-bound work, so nodes add bandwidth. [Full documentation →](get.md)
 
 ---
 

docs/config_options.md

@@ -190,7 +190,7 @@ dt config set index.retry_interval 10
 
 ## qxub Options
 
-These options configure the jobs `dt` submits via [qxub](https://github.com/swarbricklab/qxub): `dt push --workers`, `dt add`, and (via the `archive.qxub_*` fallbacks) `dt remote archive stage --via-qxub`.
+These options configure the jobs `dt` submits via [qxub](https://github.com/swarbricklab/qxub): `dt push --workers`, `dt get --workers`, `dt remote verify --workers`, `dt add`, and (via the `archive.qxub_*` fallbacks) `dt remote archive stage --via-qxub`.
 
 ### `qxub.env`
 
@@ -284,7 +284,8 @@ dt push -w 16
 ```
 
 `dt pull` has no `--workers` flag; it pulls in-process. The `qxub.*` settings
-apply to `dt push -w` and to `dt add` jobs submitted via qxub.
+apply to `dt push -w`, `dt get -w` (`s3://` destinations only), `dt remote
+verify -w`, and to `dt add` jobs submitted via qxub.
 
 ## auth Options
 

docs/get.md

@@ -113,6 +113,8 @@ S3 destinations only (see [To an S3 bucket](#to-an-s3-bucket)):
 - `--dest-region <region>`: Destination region. Required if the profile's region is `auto`.
 - `--dest-account-id <id>`: Abort unless the destination credentials resolve to this account
 - `--chunk-size <MiB>`: Streaming chunk size (default: 8)
+- `-w, --workers <n>`: Distribute across n compute nodes via qxub — see [Across several compute nodes](#across-several-compute-nodes---workers)
+- `--no-wait`: With `--workers`, submit the jobs and exit without waiting
 
 ## To an S3 bucket
 
@@ -172,6 +174,71 @@ keep accruing storage charges. Add a lifecycle rule expiring incomplete multipar
 Not supported: `ssh://` destinations, and `--link` (there is nothing to link — an S3 destination is
 always a stream).
 
+## Across several compute nodes: `--workers`
+
+`-w/--workers N` spreads an `s3://` transfer over N compute nodes via [qxub](https://github.com/swarbricklab/qxub), instead of N threads in one process.
+
+```bash
+# 82 samples out to a collaborator's bucket, over 8 nodes
+dt get bcarc_wts --csv samples.csv -o s3://their-bucket/fastqs/ -w 8 --dest-profile their-aws
+
+# Submit and go; collect the result later with qxub monitor
+dt get bcarc_wts --csv samples.csv -o s3://their-bucket/fastqs/ -w 8 --no-wait
+```
+
+### Why only `s3://`
+
+`--workers` is an error on a local destination, and that is deliberate rather than unfinished:
+
+| | bound by | second node adds |
+|---|---|---|
+| `-o s3://…` | the network | **bandwidth** — a separate NIC |
+| `-o some/dir/` | the filesystem both ends share | contention |
+
+Streaming into object storage is network-bound, so nodes multiply throughput. A local destination on Lustre is limited by the filesystem, not the node, so a second node competes with the first rather than helping. Raise `-j` there.
+
+On NCI there is a second reason: `copyq` is the queue with outbound network access, and it is already the default for `dt` transfer workers (`qxub.queue`).
+
+> **Worth measuring before you scale up.** If your *source* remote is on `/g/data` rather than object storage, the read side is a shared filesystem again and the same argument applies to it. Time a single-node `-j 8` run over a few samples and watch whether throughput saturates before adding nodes.
+
+### The two axes
+
+`--jobs` and `--workers` are not alternatives, and they compose — `-w 8 -j 4` is 8 nodes running 4 streams each.
+
+| | `-w, --workers` | `-j, --jobs` |
+|---|---|---|
+| Unit | compute nodes via qxub | threads in one process |
+| Partitioned | up front, by size | not at all — a shared queue |
+
+The difference in the second row is the whole reason `--workers` needs more machinery than a bigger `-j`. Separate processes on separate machines cannot steal work from each other, so the split has to be decided before they start; a thread pool hands tasks out from a shared queue and self-corrects as it goes.
+
+### How the work is split
+
+Files are partitioned by **size**, largest-first onto the least-loaded worker (greedy LPT bin-packing — the same code `dt push --workers` uses). Fastq sample sizes are genuinely uneven, which is exactly the case this is for: a split by row count would hand one node a 40 GiB sample and another a 4 GiB one.
+
+The unit is a *file*, not a row, so one sample directory may be spread over several nodes. Reporting is unaffected — each file carries its row, and the per-row `✓`/`✗` report comes back in CSV order as usual:
+
+```
+Destination: s3://their-bucket/fastqs
+Writing as:  account 123456789012 as arn:aws:iam::123456789012:user/handoff
+Partition byte balance across 8 active worker(s): min=41.2G, max=46.8G, total=358G
+✓ data/fq/AF013-A: 2 files -> s3://their-bucket/fastqs/AF013-A
+✓ data/fq/AF013-B: 2 files -> s3://their-bucket/fastqs/AF013-B
+```
+
+Read the balance line: `min` far below `max` means one file dominates a partition and more nodes will not help.
+
+### What each worker needs
+
+The submitting node resolves every row once — it has to, because the byte sizes the partitioning needs come from that resolve — and ships the resolved md5, size and destination key for each file in the manifest under `.dt/tmp/get/<job-id>/`. Consequences worth knowing:
+
+- **No worker runs `dvc list`.** DVC keeps a SQLite state database for the repository it reads, and pointing N compute nodes at the shared clone's copy of it is how you get `database is locked`. Each worker opens the clone once, read-only, for one thing only: the source remote's filesystem and credentials out of its `.dvc/config`.
+- **The manifest directory and the clone must be visible from the compute nodes.** On `/scratch` or `/g/data` they are. This is the same requirement `dt push --workers` has.
+- **Each worker preflights for itself**, so `--dest-account-id` is enforced on the node that actually writes — a credential chain can resolve differently on a compute node than on a login node.
+- **No key material is written to the manifest.** The profile *name*, endpoint and region travel; secrets stay in `~/.aws`, which the compute nodes read for themselves.
+
+A worker killed mid-partition — walltime, usually — reports its rows as failed rather than silently short. Re-run with `--resume` to send only what is missing.
+
 ## Examples
 
 ```bash

dt/__init__.py

@@ -1,3 +1,3 @@
 """DVC Tools - Convenient tools for working with DVC in HPC environments."""
 
-__version__ = "0.24.0"
+__version__ = "0.25.0"

dt/cli.py

@@ -838,6 +838,89 @@ def _report_rows(results, verb):
         raise SystemExit(1)
 
 
+def _get_distributed(repository, path, out, csv_path, path_col, owner, rev,
+                     num_workers, jobs, force, resume, check, refresh,
+                     dest_config, chunk_bytes, to_s3, wait, verbose):
+    """Run ``dt get --workers``: submit qxub workers and report what landed.
+
+    Split out of ``get_cmd`` because it is a third top-level mode alongside
+    single-path and ``--csv``, not a variation on either -- it consumes both
+    kinds of target list and then does something structurally different with
+    them.
+    """
+    if num_workers < 1:
+        raise click.UsageError("--workers must be at least 1.")
+
+    if not to_s3:
+        raise click.UsageError(
+            "--workers only applies to an s3:// destination.\n"
+            "Streaming to object storage is bound by the network, so more nodes "
+            "mean more bandwidth. A local destination is bound by the filesystem "
+            "both ends share, where a second node adds contention rather than "
+            "throughput -- raise -j instead."
+        )
+
+    # Rows land under the root prefix by name, matching the non-distributed
+    # path: a trailing slash on -o collects each row under its own basename,
+    # and a single path without one is used verbatim.
+    if csv_path:
+        if path:
+            raise click.UsageError("Do not provide PATH when using --csv.")
+        try:
+            rows = utils.read_csv_targets(csv_path, path_col, None)
+        except ValueError as e:
+            raise click.ClickException(str(e))
+        if not rows:
+            raise click.ClickException(f"No rows selected from {csv_path}")
+        targets = [
+            (row_path, row_out or Path(row_path or '.').name)
+            for row_path, row_out in rows
+        ]
+    else:
+        if not path:
+            raise click.UsageError("PATH is required (unless using --csv).")
+        targets = [(path, Path(path).name if out.endswith('/') else None)]
+
+    # Preflight on the submitting node before spending a queue slot: a bad
+    # profile or a missing s3:PutObject grant should fail here, not eight times
+    # over in eight PBS jobs.
+    root = get_dest.S3Dest(out, dest_config)
+    identity = root.preflight()
+    click.echo(f"Destination: {root}")
+    click.echo(f"Writing as:  {identity}")
+
+    try:
+        job_ids, manifest_dir, results = get_mod.distributed_upload_to_s3(
+            repository=repository,
+            targets=targets,
+            root=root,
+            num_workers=num_workers,
+            owner=owner,
+            rev=rev,
+            jobs=jobs,
+            force=force,
+            resume=resume,
+            check=check,
+            chunk_size=chunk_bytes,
+            refresh=refresh,
+            path_col=path_col,
+            wait=wait,
+            verbose=verbose,
+        )
+    except get_mod.GetError as e:
+        raise click.ClickException(str(e))
+
+    if results is None:
+        click.echo(f"Submitted {len(job_ids)} job(s):")
+        for job_id in job_ids:
+            click.echo(f"  {job_id}")
+        click.echo(f"Manifest: {manifest_dir}")
+        click.echo(f"\nMonitor with: qxub monitor {' '.join(job_ids)}")
+        return
+
+    _report_rows(results, "uploaded")
+
+
 def _bool_config(key, default):
     """Read a boolean config value, tolerating yes/no/true/false strings."""
     raw = cfg.get_value(key)
@@ -4574,7 +4657,7 @@ def import_cmd(repository, path, out, owner, no_checkout, no_refresh, no_downloa
 
 
 @cli.command('get')
-@click.argument('repository')
+@click.argument('repository', required=False)
 @click.argument('path', required=False)
 @click.option('-o', '--out', help='Destination path. With --csv, the directory rows are collected under')
 @click.option('--owner', help='Override the GitHub owner for short names')
@@ -4593,11 +4676,19 @@ def import_cmd(repository, path, out, owner, no_checkout, no_refresh, no_downloa
 @click.option('--dest-region', default=None, help='Region for an s3:// destination. Required if the profile\'s region is "auto".')
 @click.option('--dest-account-id', default=None, help='Abort unless the destination credentials resolve to this AWS account. For unattended runs.')
 @click.option('--chunk-size', type=int, default=None, help='Streaming chunk size in MiB for s3:// destinations (default: 8)')
+@click.option('--workers', '-w', type=int, default=None,
+              help='Distribute an s3:// transfer across N compute nodes via qxub.')
+@click.option('--no-wait', is_flag=True,
+              help='With --workers: submit jobs and exit without waiting.')
+@click.option('--worker', type=int, default=None,
+              help='Worker ID (internal, used by submitted jobs).')
+@click.option('--manifest', type=click.Path(exists=True), default=None,
+              help='Manifest directory (internal, used by submitted jobs).')
 @click.option('-v', '--verbose', is_flag=True, help='Show detailed progress')
 def get_cmd(repository, path, out, owner, rev, csv_path, path_col,
             jobs, link, no_refresh, remote_fallback, resume, check, force,
             dest_profile, dest_endpoint_url, dest_region, dest_account_id,
-            chunk_size, verbose):
+            chunk_size, workers, no_wait, worker, manifest, verbose):
     """Download DVC-tracked data without creating tracking files.
 
     Wraps the idea of ``dvc get``: materialises data from a source repository
@@ -4617,14 +4708,48 @@ def get_cmd(repository, path, out, owner, rev, csv_path, path_col,
     Unlike a loop over ``dvc get``, the source repository is cloned once and
     every row is resolved against that clone.
 
+    \b
+    Two kinds of parallelism, and they are not alternatives:
+        -j, --jobs      threads inside one process
+        -w, --workers   compute nodes via qxub; s3:// destinations only
+
+    ``--workers`` exists because streaming into object storage is bound by the
+    network, so more nodes mean more bandwidth. A local destination is bound by
+    the filesystem both ends share, where a second node adds contention rather
+    than throughput -- widen ``-j`` there instead. Files are partitioned across
+    workers by size, so each gets a comparable share of the bytes.
+
     \b
     Examples:
         dt get my-registry data/fq/AF013-A -o fastqs/
         dt get my-registry --csv samples.csv -o fastqs/
         dt get my-registry --csv samples.csv --path-col fq_dir -o fastqs/
         dt get my-registry data/ref.fa --link copy -o ./
         dt get my-registry data/fq/AF013-A -o s3://bucket/fastqs/ --dest-profile aws
+        dt get my-registry --csv samples.csv -o s3://bucket/fastqs/ -w 8
+        dt get my-registry --csv samples.csv -o s3://bucket/fastqs/ -w 8 --no-wait
     """
+    # Worker mode: everything this needs is in the manifest, including the
+    # destination and the clone, so no REPOSITORY argument is required or read.
+    if worker is not None:
+        if manifest is None:
+            raise click.UsageError("--worker requires --manifest.")
+        try:
+            written, failed = get_mod.worker_upload_to_s3(
+                manifest_dir=Path(manifest),
+                worker_id=worker,
+                verbose=verbose,
+            )
+        except get_mod.GetError as e:
+            raise click.ClickException(str(e))
+        click.echo(f"{written} uploaded, {failed} failed")
+        if failed:
+            raise SystemExit(1)
+        return
+
+    if not repository:
+        raise click.UsageError("REPOSITORY is required.")
+
     if force and resume:
         raise click.UsageError(
             "--force and --resume are contradictory: one re-fetches everything, "
@@ -4662,6 +4787,20 @@ def get_cmd(repository, path, out, owner, rev, csv_path, path_col,
     chunk_bytes = (chunk_size * 1024 * 1024) if chunk_size \
         else get_dest.DEFAULT_CHUNK_SIZE
 
+    if no_wait and workers is None:
+        raise click.UsageError("--no-wait only means anything with --workers.")
+
+    if workers is not None:
+        _get_distributed(
+            repository=repository, path=path, out=out, csv_path=csv_path,
+            path_col=path_col, owner=owner, rev=rev, num_workers=workers,
+            jobs=jobs, force=force, resume=resume, check=check,
+            refresh=not no_refresh, dest_config=dest_config,
+            chunk_bytes=chunk_bytes, to_s3=to_s3, wait=not no_wait,
+            verbose=verbose,
+        )
+        return
+
     if csv_path:
         if path:
             raise click.UsageError("Do not provide PATH when using --csv.")