[jk-shah]

Addresses #40891

beam run FILE runs a Beam YAML pipeline file like
python -m apache_beam.yaml.main --yaml_pipeline_file=FILE, passing every other flag through as a pipeline option. It reports errors without tracebacks (unless --verbose), uses documented exit codes, cancels the pipeline on Ctrl-C, and names the runner that ran the pipeline.

[github-actions]

📋 Review Summary

This PR introduces an experimental beam CLI entry point (apache_beam.cli.main) that supports executing Python, Java (JAR), and YAML pipelines via a beam run subcommand with automatic pipeline type detection and unified option passing. The implementation is well-structured, cleanly separates argument parsing from execution logic, and includes good unit test coverage.

🔍 General Feedback

• The CLI architecture and pipeline type inference logic are clean and extensible for future subcommands.
• There are a couple of bugs in the Java JAR execution path: non-portable runners like DataflowRunner do not implement run_portable_pipeline directly (expecting run_pipeline), and passing --experiments on the command line will crash due to missing default initialization for known_args.experiments.

---

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

💬 Inline Findings

• sdks/python/apache_beam/cli/main.py (line 227): 🟠 High: known_args.experiments is never registered on parser, so getattr(known_args, 'experiments', None) will always return None and --experiments flags will stay in pipeline_args. However, if --experiments is ever added to parser or known_args is populated with None, known_args.experiments would be None and .append() on line 229 would raise AttributeError: 'NoneType' object has no attribute 'append'. More importantly, because --experiments is in pipeline_args, passing ['--experiments=jar_packages=...'] + pipeline_args to PipelineOptions when pipeline_args also contains --experiments=... can override or conflict depending on how DebugOptions parses multiple --experiments flags. Instead, configure DebugOptions directly on the constructed PipelineOptions object (or initialize known_args.experiments properly).

• sdks/python/apache_beam/cli/main.py (line 258): 🔴 Critical: PipelineRunner.run_portable_pipeline raises NotImplementedError on runners that do not inherit from PortableRunner, notably DataflowRunner (which implements run_pipeline(pipeline, options, pipeline_proto=...) instead of run_portable_pipeline) and FnApiRunner / DirectRunner. Calling runner.run_portable_pipeline(...) directly will fail with NotImplementedError when running a Java JAR on Dataflow or DirectRunner. Using beam.Pipeline.from_runner_api(pipeline_proto, runner, pipeline_options).run() (or runner.run_pipeline(None, pipeline_options, pipeline_proto=pipeline_proto) for Dataflow) will properly route through the runner's execution path.

• sdks/python/apache_beam/cli/main.py (line 122): 🟡 Medium: If sys.argv[1] is an option flag other than -h/--help (for example, beam --runner=DirectRunner pipeline.py or beam --version), this condition treats --runner=DirectRunner as the subcommand name, skips inserting 'run', and fails with an invalid choice error from argparse. Checking not sys.argv[1].startswith('-') would only treat non-flag arguments as subcommand candidates while allowing top-level flags or defaulting to run.

  if (len(sys.argv) > 1 and sys.argv[1] not in KNOWN_COMMANDS and

[github-actions]

Assigning reviewers:

R: @damccorm for label python.
Backup reviewer: jrmccluskey (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).

[damccorm]

Calling beam.Pipeline(...) and pipeline.run() directly without entering Pipeline.__enter__ skips the two context managers set up in Pipeline.__enter__:

• subprocess_server.JavaJarServer.beam_services(options.view_as(CrossLanguageOptions).beam_services)
• subprocess_server.SubprocessServer.cache_subprocesses()

For Beam YAML pipelines that use Java expansion services (e.g. Sql, BigQuery, Kafka, JDBC, Iceberg), SubprocessServer.cache_subprocesses() is required so the expansion service subprocess stays alive between constructor(pipeline) and pipeline.run() (when artifacts are retrieved/staged from the expansion service) and is shared across transforms.

You can still keep the separate try/except blocks for build vs. run by wrapping both in those context managers (or calling pipeline.__enter__() with try ... finally: pipeline._extra_context.close()). Could you also add a unit test verifying that these context managers are active across constructor(pipeline) and pipeline.run()?

[jk-shah]

Good catch, thank you! Updated _run in af1e34b to call pipeline.enter() before constructor(pipeline) and close pipeline._extra_context.close() in a finally block spanning both constructor(pipeline) and pipeline.run(), and added test_pipeline_context_active_across_build_and_run verifying that SubprocessServer.cache_subprocesses() and JavaJarServer.beam_services(...) stay active across both steps and are cleaned up on completion or build failure.

[damccorm]

In Pipeline.__exit__, result.wait_until_finish() is skipped when options.view_as(StandardOptions).no_wait_until_finish is set (for example, when launching a streaming or async job on Dataflow). Should we check options.view_as(StandardOptions).no_wait_until_finish here before calling result.wait_until_finish() and checking state != PipelineState.DONE? Please also add a test covering --no_wait_until_finish.

[jk-shah]

Done in af1e34b — _run now checks options.view_as(StandardOptions).no_wait_until_finish before calling result.wait_until_finish() / checking PipelineState.DONE, prints Submitted ... (not waiting for completion)., and covers this in test_no_wait_until_finish.

[damccorm]

Thanks - I'll merge after checks complete.

We should also follow up with a documentation PR for the website (and maybe even a beam blog) after the 2.79 release in late December or early January

[damccorm]

This is causing the Windows unit test suites to fail with FileNotFoundError: on POSIX \ is a literal filename character, whereas on Windows \ is the directory separator so tricky_input resolves to <self.tmp>\C_\Users\tmp_users.csv whose parent directory C_\Users does not exist yet.

Suggested change

	tricky_input = os.path.join(self.tmp, r'C_\Users\tmp_users.csv')
	shutil.copyfile(self.input, tricky_input)
	tricky_input = os.path.join(self.tmp, r'C_\Users\tmp_users.csv')
	os.makedirs(os.path.dirname(tricky_input), exist_ok=True)
	shutil.copyfile(self.input, tricky_input)

[damccorm]

Could you please fix the failing test before I merge?