[BEAM-9877] Estimate sizes of group-by-key values behind a key lazily only.

[tudorm]

Perform lazy estimation only for such iterables since they can have enormous sizes and size estimates would cause reading the data multiple times.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• 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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[tudorm]

R: @kennknowles
R: @lukecwik

[lukecwik]

retest this please

[lukecwik]

How will this impact PCollection size estimation shown in the Dataflow UI?

[tudorm]

In general, the Dataflow UI ought to use the exact counter GBK has for effecting the sanity checking, if it doesn't do that already. However, I believe the estimate in this case is for the PCollection output one step down from the read from shuffle, at which point I don't know how useful any estimation is at all when reading from a GBK (except for accounting the amount of data read and possibly reduced / transformed).

…

On Mon, May 4, 2020 at 8:38 AM Lukasz Cwik ***@***.***> wrote: How will this impact PCollection size estimation shown in the Dataflow UI? — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#11598 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABRMIED6PZHHSF32BM24HSLRP3OKTANCNFSM4MYNL2QA> .

[tudorm]

Updated the change to perform lazy estimation instead of side-stepping the estimation altogether.

[lukecwik]

retest this please

[ibzib]

LGTM

[ibzib]

It's odd that ElementByteSizeObservableIterable::iterator adds observers within the method body. I assume this is for historic reasons, since it doesn't seem to do anything now, and the comment documenting references a setObserver method that doesn't exist. Anyway, your change looks fine. But we should consider cleaning this up.

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/util/common/ElementByteSizeObservableIterable.java

Lines 49 to 61 in 6453e85

	/**
	* Returns a new iterator for this iterable. If an observer was set in a previous call to
	* setObserver(), it will observe the iterator returned.
	*/
	@Override
	public InputT iterator() {
	InputT iterator = createIterator();
	for (Observer observer : observers) {
	iterator.addObserver(observer);
	}
	observers.clear();
	return iterator;
	}

[ibzib]

I filed BEAM-9878 as a wish item.

[tudorm]

LGTM