[jk-shah]

beam pipe --from SOURCE --to SINK reads CSV, JSON, Parquet or Avro files and writes them as CSV, JSON or Avro by building a two-step Beam YAML pipeline. The format comes from the file extension (also behind a glob or shard suffix such as users.parquet*) unless a flag sets it. Existing output is never overwritten, the files written are listed, and --dry-run prints the pipeline as YAML to save and run with beam run.

[github-actions]

📋 Review Summary

This PR cleanly adds beam pipe --from SOURCE --to SINK (including --dry-run, format inference/overrides, and overwrite protection) on top of the existing Beam YAML CLI runner. However, there are two issues in how _refuse_existing_output and _matching_files interact with FileSystems for remote URIs (gs://, s3://, azfs://, hdfs://).

🔍 General Feedback

• _refuse_existing_output currently runs before PipelineOptions are parsed and registered with FileSystems.set_options(...), leaving FileSystems._pipeline_options as None in a fresh CLI process.
• _matching_files only catches ValueError from FileSystems.match, whereas missing filesystem extras ([gcp], [aws], [azure], [hadoop]) and remote I/O failures raise BeamIOError.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🔴 Critical: _refuse_existing_output(pipe.output_prefix) is called before _run_yaml parses pipeline_args into PipelineOptions and constructs beam.Pipeline(options=options, ...) (which calls FileSystems.set_options(self._options)).

In a fresh CLI invocation, FileSystems._pipeline_options is still None at this point, which breaks remote filesystems even when their extras are installed:

• s3://: S3IO.__init__(options=None) raises ValueError('Must provide one of client or options'), which _matching_files catches and misreports as EXIT_MISSING_PACKAGE.
• azfs://: BlobStorageIO.__init__(pipeline_options=None) raises AttributeError: 'NoneType' object has no attribute 'view_as'.
• hdfs://: HadoopFileSystem.__init__(pipeline_options=None) raises ValueError('pipeline_options is not set'), ignoring --hdfs_host/--hdfs_port/--hdfs_user in pipeline_args.
• gs://: GcsIO falls back to an empty PipelineOptions(), ignoring GCP flags in pipeline_args (such as --project or --impersonate_service_account).

Move _refuse_existing_output(output_prefix) into _run_yaml after pipeline = beam.Pipeline(options=options, display_data=display_data) (re-raising _CliError before except Exception in that block) so FileSystems is configured with the parsed PipelineOptions first.

[github-actions]

🟠 High: _matching_files only catches ValueError, but FileSystems.match raises BeamIOError both when a remote filesystem extra ([gcp], [aws], [azure], [hadoop]) is not installed and when a remote listing fails (e.g., auth errors or nonexistent buckets):

• GCSFileSystem, S3FileSystem, BlobStorageFileSystem, and HadoopFileSystem are all imported unconditionally in apache_beam/io/filesystems.py because their underlying modules (gcsio, s3io, blobstorageio, hadoopfilesystem) guard third-party imports with try...except ImportError.
• When the extra is missing at runtime, GcsIO/S3IO/BlobStorageIO raise RuntimeError('<X> dependencies are not installed...') (or NameError) inside _list(), and HadoopFileSystem.__init__ raises ImportError('Failed to import hdfs...'), all of which FileSystem.match / FileSystems.get_filesystem wrap into BeamIOError.

Catch BeamIOError here as well (inspecting its nested exception details or message to distinguish missing dependencies -> EXIT_MISSING_PACKAGE from general I/O errors -> EXIT_USAGE/EXIT_PIPELINE_FAILED, while respecting --verbose) so FileSystems.match failures do not crash with an unhandled traceback.

[github-actions]

Assigning reviewers:

R: @shunping for label python.
Backup reviewer: damccorm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.47%. Comparing base (8970af1) to head (593b045).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41040      +/-   ##
============================================
+ Coverage     57.44%   57.47%   +0.02%     
  Complexity     2492     2492              
============================================
  Files          1203     1204       +1     
  Lines        171109   171201      +92     
  Branches       2103     2103              
============================================
+ Hits          98294    98392      +98     
+ Misses        70161    70155       -6     
  Partials       2654     2654              

Flag	Coverage Δ
python	80.21% <100.00%> (+0.02%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.