Add MultimapState API by zhengbuqian · Pull Request #23491 · apache/beam

zhengbuqian · 2022-10-05T06:20:33Z

Adds the API and and in-memory implementation for multimap state. This is the first PR for #22831. Doc - Beam MultimapState.

#23492 is the PR for the windmill impl.

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

codecov · 2022-10-05T06:56:49Z

Codecov Report

Merging #23491 (e1d765b) into master (7f54035) will decrease coverage by 0.71%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #23491      +/-   ##
==========================================
- Coverage   73.33%   72.62%   -0.72%     
==========================================
  Files         719      736      +17     
  Lines       95792    97267    +1475     
==========================================
+ Hits        70250    70638     +388     
- Misses      24231    25318    +1087     
  Partials     1311     1311

Flag	Coverage Δ
python	`81.83% <ø> (-1.23%)`	⬇️

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

Impacted Files	Coverage Δ
...python/apache_beam/examples/complete/distribopt.py	`0.00% <0.00%> (-98.57%)`	⬇️
...hon/apache_beam/examples/cookbook/mergecontacts.py	`23.21% <0.00%> (-73.22%)`	⬇️
...he_beam/examples/cookbook/multiple_output_pardo.py	`37.50% <0.00%> (-58.34%)`	⬇️
...python/apache_beam/examples/wordcount_debugging.py	`38.77% <0.00%> (-57.15%)`	⬇️
...python/apache_beam/examples/dataframe/wordcount.py	`38.46% <0.00%> (-53.85%)`	⬇️
...s/python/apache_beam/examples/wordcount_minimal.py	`40.74% <0.00%> (-51.86%)`	⬇️
sdks/python/apache_beam/examples/wordcount.py	`48.27% <0.00%> (-44.83%)`	⬇️
...apache_beam/examples/cookbook/custom_ptransform.py	`53.65% <0.00%> (-41.58%)`	⬇️
.../apache_beam/examples/cookbook/group_with_coder.py	`51.16% <0.00%> (-34.89%)`	⬇️
sdks/python/apache_beam/examples/complete/tfidf.py	`73.43% <0.00%> (-24.95%)`	⬇️
... and 95 more

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

zhengbuqian · 2022-10-12T03:28:12Z

R: @reuvenlax @scwhittle @y1chi

github-actions · 2022-10-12T03:53:19Z

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

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

y1chi · 2022-10-12T18:20:57Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

why not just clear the instantiated contents?

See

beam/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

Line 411 in af6a04c

// Even though we're clearing we can't remove this from the in-memory state map, since

I think we should create a new container instead of clear(), this is also how Bag/Map/Set do clear().

The +1 was for your response that we need to keep stuff consistent for past reads that were returned.

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/FnApiStateAccessor.java

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

y1chi

LGTM, added some minor comments.

sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/FnApiStateAccessor.java

zhengbuqian · 2022-10-25T05:21:53Z

Gentle ping for review.

R: @reuvenlax @scwhittle

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

runners/flink/flink_runner.gradle

sdks/java/core/src/main/java/org/apache/beam/sdk/state/MultimapState.java

lukecwik · 2022-10-25T05:38:55Z

sdks/java/core/src/main/java/org/apache/beam/sdk/state/MultimapState.java

+  /** Returns an {@link Iterable} over the keys contained in this multimap. */
+  ReadableState<Iterable<K>> keys();
+
+  /** Returns an {@link Iterable} over all key-values pairs contained in this multimap. */


Suggested change

/** Returns an {@link Iterable} over all key-values pairs contained in this multimap. */

/** Returns an {@link Iterable} over all key-value pairs contained in this multimap. */

What I meant is in each pair there may be multiple values, WDYT?

I missed that we are returning Entry<K, Iterable<V>> which is why I suggested the swap to key-value over key-values.

Note that other map APIs use Entry<K, V>, see:
https://guava.dev/releases/23.0/api/docs/com/google/common/collect/ArrayListMultimap.html#entries--
https://commons.apache.org/proper/commons-collections/apidocs/org/apache/commons/collections4/MultiValuedMap.html#entries--

It would likely make sense to match what existing common libraries do on this front so we are idiomatic for Java users.

Agreed, done.

…keys.

sdks/java/core/src/main/java/org/apache/beam/sdk/state/MultimapState.java

lukecwik · 2022-11-30T20:09:37Z

sdks/java/core/src/main/java/org/apache/beam/sdk/state/MultimapState.java

+  /** Returns an {@link Iterable} over the keys contained in this multimap. */
+  ReadableState<Iterable<K>> keys();
+
+  /** Returns an {@link Iterable} over all key-values pairs contained in this multimap. */


I missed that we are returning Entry<K, Iterable<V>> which is why I suggested the swap to key-value over key-values.

Note that other map APIs use Entry<K, V>, see:
https://guava.dev/releases/23.0/api/docs/com/google/common/collect/ArrayListMultimap.html#entries--
https://commons.apache.org/proper/commons-collections/apidocs/org/apache/commons/collections4/MultiValuedMap.html#entries--

It would likely make sense to match what existing common libraries do on this front so we are idiomatic for Java users.

lukecwik · 2022-11-30T20:10:46Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

lukecwik · 2022-11-30T20:13:16Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

+
+    @Override
+    public ReadableState<Iterable<V>> get(K key) {
+      return CollectionViewState.of(contents.get(keyCoder.structuralValue(key)));


This returns a stale view in this order of operations:

mm.put("A", 1); ReadableState<Iterable<V>> iter = mm.get("A"); mm.clear(); iter.read(); <-- will contain "1"

I believe there are other combinations where you won't get what you want. Please ensure that ParDoTest covers:

get before and after remove/put/clear.

containsKey before and after remove/put/clear

entries before and after remove/put/clear

isEmpty before and after remove/put/clear

keys fore and after remove/put/clear

Currently I see your covering entries before and after put/remove.

Done, updated ParDoTest to include those as well.

sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java

sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesMultimapState.java

Co-authored-by: Lukasz Cwik <lcwik@google.com>

…<K, Iterable<V>>

…t tests

zhengbuqian · 2022-12-05T21:08:06Z

gentle ping

lukecwik

Please add coverage for get/containsKey for the structural multimap test and then LGTM

lukecwik · 2022-12-06T23:57:08Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java

The +1 was for your response that we need to keep stuff consistent for past reads that were returned.

lukecwik · 2022-12-07T00:12:29Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+                assertEquals(4, Iterables.size(entries));
+                assertEquals(5, Iterables.size(entriesView.read()));
+                assertEquals(5, Iterables.size(state.entries().read()));
+


please add coverage for containsKey/get since these accessors are important and need to do the structural key conversion.

Done. Can you help merge the PR if LGTY? thanks!

…ting structural values.

lukecwik · 2022-12-15T22:00:48Z

Run Java PreCommit

lukecwik · 2022-12-15T22:00:57Z

Run Java_Spark3_Versions PreCommit

…work(that input elements are processed in the same order as in Create.of) is made.

zhengbuqian · 2022-12-16T00:02:15Z

The unit test testMultimapStateRemove failed without commit e1d765b because I made the assumption that the test framework will process the following 4 elements in order, but seems that it's not the case, for example b, 33 may be the 1st/2nd/3rd/4th processed element. testMultimapStateRemove was the only unit test that relies on such ordering and has been fixed, all other tests verify the result only after all elements have been processed thus have no issue.

pipeline
              .apply(
                  Create.of(
                      KV.of("hello", KV.of("a", 97)), KV.of("hello", KV.of("a", 97)),
                      KV.of("hello", KV.of("a", 98)), KV.of("hello", KV.of("b", 33))))
              .apply(ParDo.of(fn));

zhengbuqian · 2022-12-16T02:21:26Z

Run Java_GCP_IO_Direct PreCommit

zhengbuqian · 2022-12-16T02:21:45Z

Run Java_Examples_Dataflow_Java11 PreCommit

* Add Multimap API * Update based on commit * Fix format * Fix typo and adopt comment suggestion * Fix spotless check * Update based on comments * Change Multimap in memory implementation to use structural value for keys. * Address comments * Apply suggestions from code review Co-authored-by: Lukasz Cwik <lcwik@google.com> * Update MultimapState.entries() to return Entry<K, V> instead of Entry<K, Iterable<V>> * Update MultimapState inmemory runner impl and added more thorough unit tests * In MultimapState ParDoTest also test against get/containsKey when testing structural values. * Fix flaky unit test, in which incorrect assumptions of the test framework(that input elements are processed in the same order as in Create.of) is made. Co-authored-by: Lukasz Cwik <lcwik@google.com>

zhengbuqian marked this pull request as draft October 5, 2022 06:26

github-actions bot added core dataflow direct flink java jet model portability runners samza spark labels Oct 5, 2022

zhengbuqian force-pushed the multimapstate-api branch from 6618e8a to 7dcb816 Compare October 12, 2022 00:16

zhengbuqian changed the title ~~[DRAFT] Add MultimapState API~~ Add MultimapState API Oct 12, 2022

zhengbuqian marked this pull request as ready for review October 12, 2022 03:28

Add Multimap API

306db0f

zhengbuqian force-pushed the multimapstate-api branch from 7dcb816 to 306db0f Compare October 12, 2022 18:48

y1chi reviewed Oct 12, 2022

View reviewed changes

zhengbuqian added 2 commits October 12, 2022 14:35

Update based on commit

4bd6e61

Fix format

43831b2

y1chi approved these changes Oct 12, 2022

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateSpecs.java Outdated Show resolved Hide resolved

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/FnApiStateAccessor.java Outdated Show resolved Hide resolved

zhengbuqian added 2 commits October 12, 2022 16:29

Fix typo and adopt comment suggestion

bc8685e

Fix spotless check

3339044

lukecwik requested changes Oct 25, 2022

View reviewed changes

zhengbuqian added 2 commits November 1, 2022 22:45

Update based on comments

1debe2b

Change Multimap in memory implementation to use structural value for …

838d18a

…keys.

zhengbuqian requested review from lukecwik and reuvenlax November 28, 2022 21:46

lukecwik requested changes Nov 30, 2022

View reviewed changes

zhengbuqian and others added 3 commits November 30, 2022 18:32

Apply suggestions from code review

4c23ec6

Co-authored-by: Lukasz Cwik <lcwik@google.com>

Update MultimapState.entries() to return Entry<K, V> instead of Entry…

e92f14d

…<K, Iterable<V>>

Update MultimapState inmemory runner impl and added more thorough uni…

a826d8c

…t tests

zhengbuqian requested review from lukecwik and reuvenlax and removed request for lukecwik and reuvenlax December 1, 2022 07:55

lukecwik requested changes Dec 7, 2022

View reviewed changes

In MultimapState ParDoTest also test against get/containsKey when tes…

ce8291f

…ting structural values.

zhengbuqian requested a review from lukecwik December 14, 2022 00:21

lukecwik approved these changes Dec 15, 2022

View reviewed changes

Fix flaky unit test, in which incorrect assumptions of the test frame…

e1d765b

…work(that input elements are processed in the same order as in Create.of) is made.

lukecwik approved these changes Dec 16, 2022

View reviewed changes

lukecwik merged commit 7d5d62e into apache:master Dec 16, 2022

zhengbuqian deleted the multimapstate-api branch December 19, 2022 23:52

tvalentyn mentioned this pull request Dec 21, 2022

[Failing Test]: Exclude MultimapState tests from Dataflow VR suites #24750

Closed

15 tasks

zhengbuqian mentioned this pull request Dec 21, 2022

Exclude MultimapState related tests in Dataflow legacy worker validates runner tests #24758

Merged

3 tasks

rakeshcusat mentioned this pull request Jun 15, 2025

Update state doc 34905 #35209

Merged

3 tasks

	/** Returns an {@link Iterable} over all key-values pairs contained in this multimap. */
	/** Returns an {@link Iterable} over all key-value pairs contained in this multimap. */

Conversation

zhengbuqian commented Oct 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

codecov bot commented Oct 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

zhengbuqian commented Oct 12, 2022

Uh oh!

github-actions bot commented Oct 12, 2022

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

y1chi left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

zhengbuqian commented Oct 25, 2022

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zhengbuqian commented Dec 5, 2022

Uh oh!

lukecwik left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhengbuqian Dec 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik commented Dec 15, 2022

Uh oh!

lukecwik commented Dec 15, 2022

Uh oh!

zhengbuqian commented Dec 16, 2022

Uh oh!

zhengbuqian commented Dec 16, 2022

zhengbuqian commented Oct 5, 2022 •

edited

Loading

codecov bot commented Oct 5, 2022 •

edited

Loading

zhengbuqian Dec 14, 2022 •

edited

Loading