[Go SDK]: Add fileio MatchFiles, MatchAll and ReadMatches transforms

[johannaojeling]

Fixes #25779 and adds the fileio transforms MatchFiles, MatchAll and ReadMatches

---

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.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[johannaojeling]

R: @lostluck

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[codecov]

Codecov Report

Merging #25809 (256ae9c) into master (469101d) will increase coverage by 0.00%.
The diff coverage is 75.57%.

@@           Coverage Diff            @@
##           master   #25809    +/-   ##
========================================
  Coverage   71.42%   71.43%            
========================================
  Files         779      783     +4     
  Lines      102635   102852   +217     
========================================
+ Hits        73308    73469   +161     
- Misses      27867    27911    +44     
- Partials     1460     1472    +12     

Flag	Coverage Δ
go	54.09% <75.57%> (+0.12%)	⬆️

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

Impacted Files	Coverage Δ
sdks/go/pkg/beam/io/fileio/gzip.go	55.55% <55.55%> (ø)
sdks/go/pkg/beam/io/fileio/file.go	69.09% <69.09%> (ø)
sdks/go/pkg/beam/io/fileio/read.go	79.31% <79.31%> (ø)
sdks/go/pkg/beam/io/fileio/match.go	81.39% <81.39%> (ø)

... and 1 file with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[lostluck]

This is great! Thank you! I love that this entirely avoids solving the question of "what type are we reading out?". This

The big counter argument to my suggestion in how options are treated is that enums are trivially serializable, while other approach isn't necessarily trivial to serialize. However, a user who really needs to serialize these things can always manage that process themselves.

Remember, types themselves don't need to be exported to be serialized, only their Fields, so this shouldn't affect using unexported enums for configuring readFn or matchFn.

[johannaojeling]

I haven't provided any option functions for setting what is already the default value, i.e. there is no MatchEmptyAllowIfWildcard(), because it doesn't make much sense? Unless it should be included for clarity/documentation purposes?

[lostluck]

Better to have one for the default if only to explain that it's the default.

Also in complex situations, it might get set multiple times in the list of options, so having the default enables setting it back into the default if desired.

[johannaojeling]

Okay, fixed now. Feel free to merge if approved

[lostluck]

I think longer term when if add additional compression formats, we'd want to do something similar to how the Filesystems are set up, so users don't need to have all possible compressions linked in.

I think it's fine to have Gzip built in by default as it's most common, but if we build in support for more, it should be made modular, and hooked into the "auto" support.

[lostluck]

Thank you very much!

I think I'm a touch surprised that neither of these are Splittable DoFns. I'd expect Match to be able to "sub element split on it's input globs to allow downstream processing of the files to ultimately be split down to each file (if not within the file itself), since we know/can find the size and so forth, to begin to make decent splitting decisions.

But I think we can leave this as is for the moment, since it took me a week longer than desired to get to this review, and this is still a net good that other file based IOs in the SDK can start to build with.

[johannaojeling]

I think I'm a touch surprised that neither of these are Splittable DoFns. I'd expect Match to be able to "sub element split on it's input globs to allow downstream processing of the files to ultimately be split down to each file (if not within the file itself), since we know/can find the size and so forth, to begin to make decent splitting decisions.

Hmm I don't see clearly how the incoming glob element in matchFn should be split into restriction pairs. Only after the List operation do we know how many output elements there are per input element. I guess we could create an initial offsetrange.Restriction based on the total count of files that match the glob pattern from a List invocation, then split that into sub-ranges. The downside is that this would impose additional API calls for the same List operation in ProcessElement and require sorting of the list result to determine which files fall under the current restriction. I'm curious to hear what would be your suggestion for how to implement the Match transform with an SDF?

As I saw it, the matchFn essentially serves the same purpose as expandFn currently does in textio/avroio/parquetio, plus attaches the size. I was thinking that those IOs could potentially be refactored to make use of the fileio Match and Read tranforms to reduce repetition if we want that. At least textio which would utilize both the file handle and the size from a ReadableFile.