You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "derrickaw (via GitHub)" <gi...@apache.org> on 2026/05/15 15:05:32 UTC
[PR] add support for matchall [beam]
derrickaw opened a new pull request, #38512:
URL: https://github.com/apache/beam/pull/38512
#38013
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3250425498
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
+
+ matched = patterns | beam.io.fileio.MatchAll(
+ empty_match_treatment=empty_match_treatment)
Review Comment:
ignore
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3263407817
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,60 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
Review Comment:
Will adding this as a source here allow us to make this under YAML transforms in job builder ? I think job builder sources do not support input PCollections (so has to be a Transform there).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3263407817
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,60 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
Review Comment:
Will adding this as a source here allow us to make this available under YAML transforms in job builder ? I think job builder sources do not support input PCollections (so has to be a Transform there).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3250413174
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
+
+ matched = patterns | beam.io.fileio.MatchAll(
+ empty_match_treatment=empty_match_treatment)
+
+ return matched | beam.Map(
+ lambda x: beam.Row(
+ path=str(x.path),
+ size_in_bytes=int(x.size_in_bytes),
+ last_updated_in_seconds=float(x.last_updated_in_seconds), ))
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38512:
URL: https://github.com/apache/beam/pull/38512#issuecomment-4484221682
Assigning reviewers:
R: @jrmccluskey for label python.
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).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw merged PR #38512:
URL: https://github.com/apache/beam/pull/38512
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on PR #38512:
URL: https://github.com/apache/beam/pull/38512#issuecomment-4462499374
/gemini review
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3289418585
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,60 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
Review Comment:
Good call. I moved to standard_providers so I think JB will pick it up as a transform.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #38512:
URL: https://github.com/apache/beam/pull/38512#issuecomment-4462888690
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/38512?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `84.21053%` with `3 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 57.58%. Comparing base ([`9a1a4b1`](https://app.codecov.io/gh/apache/beam/commit/9a1a4b170e4aebbb007db778bc19d7e2f5516b51?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3502cff`](https://app.codecov.io/gh/apache/beam/commit/3502cff58213168e221d3338ea041f6aeb7e1147?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 5 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/38512?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [sdks/python/apache\_beam/yaml/yaml\_io.py](https://app.codecov.io/gh/apache/beam/pull/38512?src=pr&el=tree&filepath=sdks%2Fpython%2Fapache_beam%2Fyaml%2Fyaml_io.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0veWFtbC95YW1sX2lvLnB5) | 84.21% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/38512?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #38512 +/- ##
============================================
+ Coverage 57.54% 57.58% +0.03%
Complexity 5329 5329
============================================
Files 1398 1400 +2
Lines 198769 198943 +174
Branches 4980 4980
============================================
+ Hits 114389 114555 +166
- Misses 80466 80474 +8
Partials 3914 3914
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/38512/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/38512/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.88% <84.21%> (+0.03%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/38512?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3251025253
##########
sdks/python/apache_beam/yaml/yaml_io_test.py:
##########
@@ -18,6 +18,8 @@
import io
import json
import logging
+import os
+import tempfile
import unittest
Review Comment:
mock is already imported
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3250390787
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38512:
URL: https://github.com/apache/beam/pull/38512#issuecomment-4463624835
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces the 'MatchAll' transform to the Apache Beam YAML SDK. This addition enables users to dynamically match files based on patterns provided in the input PCollection, enhancing the file handling capabilities within YAML pipelines. The implementation includes robust schema validation and flexible configuration options for empty match scenarios.
### Highlights
* **New MatchAll Transform**: Added a new 'MatchAll' transform to the Beam YAML API, allowing users to match file patterns within their pipelines.
* **Configuration Support**: Implemented support for specifying the file pattern field and handling empty matches via the 'empty_match_treatment' configuration.
* **Testing**: Added comprehensive unit tests and a new YAML test file to verify the functionality and error handling of the MatchAll transform.
<details>
<summary><b>New Features</b></summary>
🧠 You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3251017243
##########
sdks/python/apache_beam/yaml/yaml_io_test.py:
##########
@@ -18,6 +18,8 @@
import io
import json
import logging
+import os
+import tempfile
import unittest
Review Comment:
The `mock` module is used in the new tests (e.g., line 643) but it is not imported in this file. This will cause a `NameError` when running the tests. Please add the import.
```suggestion
import unittest
from unittest import mock
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [yaml] add support for matchall [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38512:
URL: https://github.com/apache/beam/pull/38512#discussion_r3250359453
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
+
+ matched = patterns | beam.io.fileio.MatchAll(
+ empty_match_treatment=empty_match_treatment)
Review Comment:
`beam.io.fileio.MatchAll` expects an `EmptyMatchTreatment` enum value for the `empty_match_treatment` parameter. Passing a string directly may not work as expected because the internal logic in `FileSystems.match` compares against enum members. It should be converted using the enum.
```suggestion
from apache_beam.io.fileio import EmptyMatchTreatment
matched = patterns | beam.io.fileio.MatchAll(
empty_match_treatment=getattr(EmptyMatchTreatment, empty_match_treatment))
```
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
+
+ matched = patterns | beam.io.fileio.MatchAll(
+ empty_match_treatment=empty_match_treatment)
+
+ return matched | beam.Map(
+ lambda x: beam.Row(
+ path=str(x.path),
+ size_in_bytes=int(x.size_in_bytes),
+ last_updated_in_seconds=float(x.last_updated_in_seconds), ))
Review Comment:
`float(x.last_updated_in_seconds)` will raise a `TypeError` if `last_updated_in_seconds` is `None`. While `MatchAll` usually provides this value, some filesystems might not. It is safer to handle the `None` case. Additionally, the explicit `str()` and `int()` casts are likely redundant if the input `FileMetadata` fields are already of the expected types.
```suggestion
return matched | beam.Map(
lambda x: beam.Row(
path=str(x.path),
size_in_bytes=int(x.size_in_bytes),
last_updated_in_seconds=float(x.last_updated_in_seconds)
if x.last_updated_in_seconds is not None else None))
```
##########
sdks/python/apache_beam/yaml/yaml_io.py:
##########
@@ -723,3 +723,57 @@ def write_to_tfrecord(
num_shards=num_shards,
shard_name_template=shard_name_template,
compression_type=getattr(CompressionTypes, compression_type))
+
+
+@beam.ptransform_fn
+def match_all(
+ pcoll,
+ *,
+ file_pattern: Optional[str] = None,
+ empty_match_treatment: str = 'ALLOW',
+):
+ """Matches file patterns from the input PCollection.
+
+ This transform returns a PCollection of matching files, each represented as a
+ Row with path, size_in_bytes, and last_updated_in_seconds fields.
+
+ Args:
+ file_pattern (str): The name of the field in the input PCollection that contains
+ the file pattern string. If not specified and the input PCollection has
+ exactly one field, that field will be used.
+ empty_match_treatment (str): How to treat empty matches. Possible values are
+ 'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
+ """
+ from apache_beam.typehints import schemas
+
+ try:
+ field_names = [
+ name for name, _ in schemas.named_fields_from_element_type(
+ pcoll.element_type)
+ ]
+ if file_pattern is not None:
+ if file_pattern not in field_names:
+ raise ValueError(
+ f"Field '{file_pattern}' not found in input schema fields: {field_names}"
+ )
+ pattern_field = file_pattern
+ elif len(field_names) == 1:
+ pattern_field = field_names[0]
+ else:
+ raise ValueError(
+ f"Input schema has multiple fields {field_names}. "
+ f"Please specify the 'file_pattern' parameter to select which field "
+ f"contains the file pattern.")
+ patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
+ except Exception:
+ # Fallback for PCollection without a schema (e.g. raw string elements)
+ patterns = pcoll
Review Comment:
The `try-except` block is too broad. It wraps the entire logic for schema-aware PCollections, including explicit `ValueError` raises for missing fields. If a user provides an invalid `file_pattern`, the `ValueError` is caught and the code falls back to treating the entire element as a pattern (line 770), which will likely lead to a confusing error later in `MatchAll` (e.g., an `AttributeError` when trying to treat a Row as a string). The `try-except` should only cover the schema inspection.
```python
try:
field_names = [
name for name, _ in schemas.named_fields_from_element_type(
pcoll.element_type)
]
except Exception:
field_names = None
if field_names:
if file_pattern is not None:
if file_pattern not in field_names:
raise ValueError(
f"Field '{file_pattern}' not found in input schema fields: {field_names}"
)
pattern_field = file_pattern
elif len(field_names) == 1:
pattern_field = field_names[0]
else:
raise ValueError(
f"Input schema has multiple fields {field_names}. "
f"Please specify the 'file_pattern' parameter to select which field "
f"contains the file pattern.")
patterns = pcoll | beam.Map(lambda x: str(getattr(x, pattern_field)))
else:
patterns = pcoll
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org