Cache total size in file based source

[Abacn]

In some use case it is found excessive match request (up to 6 times) during splitting file based source (e.g. in BigQueryTableSource) and causing split request taking lots of time when there are lots of files.

This value can be cached at minimum cost and avoid actual match api call.

Please add a meaningful description for your change here

---

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.

[Abacn]

Some sources e.g. BigQueryTableSource already implemented cache:

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryTableSource.java

Lines 71 to 73 in 7caadb2

	Long maybeNumBytes = tableSizeBytes.get();
	if (maybeNumBytes != null) {
	return maybeNumBytes;

[Abacn]

R: @johnjcasey

[github-actions]

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

[Abacn]

Run Java_Examples_Dataflow PreCommit

[Abacn]

Run Java_GCP_IO_Direct PreCommit

[Abacn]

Run Java PreCommit

[johnjcasey]

Is this a file resource level cache?

[Abacn]

No, it's just a Long value

[johnjcasey]

Can we add a test case for this?

[Abacn]

Done

[Abacn]

Did some test, reading from tpcds_1T.web_sales:

• on master, runner v1

11562 logs searching for pattern gs://.../temp/BigQueryExtractTemp/42fa897c43704ed7998868ebf83e9198/

• this branch, runner v1

5781 logs for gs://.../temp/BigQueryExtractTemp/ebd9c20b02f142439a837fbf99d72e25/ basically each file is matched only 3 times (2 during split, 1 before delete)

That is a decrease by half (meaning making half of the match request)

• on master, runner v2

3908 logs searching for pattern "gs://.../temp/BigQueryExtractTemp/8568e876f4404544a69b996c5ff47ca4/" basically each file is matched only 2 times (during split, before delete)

Tested on 10 workers (n1s1); runner v1 has 520k record/sec throughput, while runner v2 has 600k record/sec throughput.

Summary:

• This change significantly reduced the number of List API request to gcs by half on Dataflow runner v1
• No change on Dataflow runner v2 as the runner already avoided redundant call for estimateSize.
• It is found that runner v2 is more efficient for this use case (BigQueryIO EXPORT read), having higher throughput under same environment.

[Abacn]

Java PreCommit known flake unrelated: #21333