Added replaceGcsFilesWithLocalFiles

[liferoad]

Fixes #32531

Improve GCS file handling in DataflowRunner

The proposed changes are only limited to DataflowRunner and are kept simple. When a GCS file is detected in filesToStage, we first download this to a local temp file and replace the GCS file with this local temp file.

Changes

• Added functionality to handle GCS files in replaceGcsFilesWithLocalFiles() method
• Files with GCS paths (gs://) are now downloaded to local temp files before being staged
• Maintains original file names when downloading from GCS
• Handles both files with and without staging names specified
• Adds logging for file downloads and replacements

Notes

• Temp files are created with same names as original GCS files for consistency
• Files are downloaded using FileSystems API for proper GCS integration
• Original functionality for local files is preserved unchanged

Tests

• Add a simple IT to validate the file is downloaded and staged later with a Dataflow job

---

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 or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

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

R: @kennknowles added as fallback since no labels match configuration

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).

[Abacn]

Thanks. This is a working solution.

Alternatively, I am wondering if the client could initiate gcs copy call from source path to the actual staging path when staging happens, this would eliminate the need to download actual file to local. Could be a separate task

[kennknowles]

I don't totally understand the use of this method, but surely you should use Filesystems.copy method that does this:

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileSystems.java

Line 276 in 159c1a4

public static void copy(

[liferoad]

Our Filesystems.copy has too many layers. Here, we simply downloads the file to a given temp file, which is also removed with deleteOnExit. I think we do not need to use our internal tool since it is quite straightforward.

[kennknowles]

What you implemented is just the same as copy but now we have two copies to maintain, really. The layers should be robust.

[liferoad]

Expect srcResourceIds and destResourceIds have the same scheme, but received gs, file.
java.lang.IllegalArgumentException: Expect srcResourceIds and destResourceIds have the same scheme, but received gs, file.

Looks like our copy needs the same scheme. :)

[liferoad]

Will close this PR and create a new one to do this when staging the GCS files.

[liferoad]

some offline discussions with @kennknowles. Adding this to the staging phase needs more time and more places to be changed since the current staging logic relies on local files (e.g., computing hash). So we decided to merge this PR for now. The improvements on the staging phase could be done later.