github-actions[bot] commented on code in PR #41040:
URL: https://github.com/apache/beam/pull/41040#discussion_r4495341923
##########
sdks/python/apache_beam/cli/main.py:
##########
@@ -125,6 +196,39 @@ def _run(file, pipeline_args, verbose):
hint = '\n pipeline options go after FILE, written as --flag=value'
raise _CliError(f'no such file: {file!r}{hint}', EXIT_USAGE)
+ return _run_yaml(['--yaml_pipeline_file=' + file] + pipeline_args, verbose)
+
+
+def _pipe(args, pipeline_args, pipe_argv):
+ """Runs `beam pipe`; `pipe_argv` is its command line, for --dry-run."""
+ try:
+ pipe = pipe_spec.build_pipe_spec(
+ args.source,
+ args.sink,
+ args.from_format or args.format,
+ args.to_format or args.format)
+ except pipe_spec.PipeSpecError as e:
+ raise _CliError(str(e), EXIT_USAGE) from None
+
+ # A dry run makes every check a real run makes, including building the
+ # pipeline, so that it predicts whether `beam run` on its output will work.
+ _refuse_existing_output(pipe.output_prefix)
Review Comment:
**🔴 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.
##########
sdks/python/apache_beam/cli/main.py:
##########
@@ -199,9 +315,81 @@ def _run(file, pipeline_args, verbose):
raise _CliError(f'pipeline finished in state {state}',
EXIT_PIPELINE_FAILED)
print(f'Done. Ran {_describe_runner(options, result)}.')
+ if output_prefix is not None:
+ written = _matching_files(output_prefix)
+ print(f'Wrote {len(written)} file{"" if len(written) == 1 else "s"}:')
+ for path in written:
+ print(' ' + path)
return EXIT_OK
+def _refuse_existing_output(prefix):
+ """Fails if any file starts with the output prefix: pipe never overwrites."""
+ existing = _matching_files(prefix)
+ if existing:
+ shown = '\n'.join(' ' + path for path in existing[:5])
+ more = f'\n ... and {len(existing) - 5} more' if len(existing) > 5 else ''
+ raise _CliError(
+ f'output already exists for {prefix!r}; beam pipe never overwrites '
+ f'data:\n{shown}{more}\n'
+ ' remove these files or choose another --to path',
+ EXIT_USAGE)
+
+
+def _matching_files(prefix):
+ # LocalFileSystem.match calls os.path.dirname(prefix), which is '' for a
+ # bare filename in the current directory and causes _list('') to return no
+ # files. Prepending './' (or '.\' on Windows) gives dirname '.' so files in
+ # the current directory are found.
+ dot = (
+ '.' +
+ os.sep if '://' not in prefix and not os.path.dirname(prefix) else '')
+ try:
+ match = FileSystems.match([dot + prefix + '*'])[0]
+ except ValueError as e:
Review Comment:
**🟠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.
--
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]
