New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add Windmill support for MultimapState #23492

Merged

y1chi merged 6 commits into apache:master from zhengbuqian:multimap-windmill

Jul 5, 2023

Contributor

zhengbuqian commented Oct 5, 2022 •

edited

Loading

This PR adds windmill legacy worker support for MultimapState.

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.

zhengbuqian marked this pull request as draft

October 5, 2022 06:39

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

codecov bot commented Oct 5, 2022 •

edited

Loading

Codecov Report

Merging #23492 (add1c3f) into master (7db76df) will increase coverage by 0.50%.
The diff coverage is n/a.

❗ Current head add1c3f differs from pull request most recent head 52f6338. Consider uploading reports for the commit 52f6338 to get more accurate results

@@            Coverage Diff             @@
##           master   #23492      +/-   ##
==========================================
+ Coverage   72.82%   73.32%   +0.50%     
==========================================
  Files         775      719      -56     
  Lines      102928    95799    -7129     
==========================================
- Hits        74958    70249    -4709     
+ Misses      26515    24239    -2276     
+ Partials     1455     1311     -144

Flag	Coverage Δ
python	`83.04% <ø> (+1.07%)`	⬆️

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

see 274 files with indirect coverage changes

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

zhengbuqian force-pushed the multimap-windmill branch from bd6791a to 81bdf90 Compare

October 12, 2022 00:16

zhengbuqian mentioned this pull request

Add MultimapState API #23491

Merged

4 tasks

zhengbuqian force-pushed the multimap-windmill branch from 81bdf90 to f48eeab Compare

October 12, 2022 18:48

zhengbuqian force-pushed the multimap-windmill branch from f48eeab to add1c3f Compare

December 1, 2022 08:02

zhengbuqian force-pushed the multimap-windmill branch from add1c3f to 14a2006 Compare

December 17, 2022 02:54

github-actions bot removed java jet core portability samza flink model direct spark labels

zhengbuqian changed the title ~~[Draft] Add Windmill support for MultimapState~~ Add Windmill support for MultimapState

zhengbuqian marked this pull request as ready for review

December 19, 2022 19:16

Contributor

github-actions bot commented Dec 19, 2022

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

scwhittle requested changes

View reviewed changes

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated

+                                    keyState.existence = KeyExistence.KNOWN_EXIST;
+                                    keyState.values.extendWith(entry.getValue());
+                                    // We can't set keyState.valuesCached to true here, because there may be more
+                                    // paginated values that should not be filtered out in above if statement.

Contributor

scwhittle Feb 8, 2023

this isn't a paginated read, can we set valuesCached?
Or shoudl comment be updated that there may be additional entries for the key.

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated

-                            return Iterables.unmodifiableIterable(
-                                Iterables.concat(mergedCachedEntries(), fromWindmill));
+                            MultimapIterables<K, V> cachedEntries = mergedCachedEntries();
+                            entries.forEach(

Contributor

scwhittle Feb 8, 2023

I think this is back to loading all the pages into the memory before iterating

Could you instead make it lazy by returning windmill non-cached data first (filtering cached or added to cached) then return the rest?

Iterables.concat(Iterables.transform( logic in entries.forEach),
some iterable that lazily calls mergedCacheEntries and iterates);

Maybe you can make some test for this by having the test have some iterable that would return GBs of data total which should be streamed through and not cause test to OOM?

zhengbuqian requested a review from scwhittle

February 10, 2023 00:14

zhengbuqian force-pushed the multimap-windmill branch 2 times, most recently from 56a57e7 to 230a03f Compare

February 28, 2023 05:28


          Add windmill APIs proto and windmill legacy worker implementations fo…

8fa3d3f

…r multimap state

zhengbuqian force-pushed the multimap-windmill branch from 230a03f to 8fa3d3f Compare

February 28, 2023 22:09

zhengbuqian added 2 commits

March 1, 2023 23:53


          Always set valuesCached to true when removing key from multimap

7139a85


          Fix spotless check error

2110bd0

scwhittle requested changes

View reviewed changes

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated Show resolved Hide resolved

...w-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateReader.java

                     consumeSortedList(sorted_list, stateTag);
                   }
+                  for (Windmill.TagMultimapFetchResponse tagMultimap : response.getTagMultimapsList()) {
+                    // First check if it's keys()/entries()

Contributor

scwhittle Mar 23, 2023

Can always optimize it later


          Address comments

79f5e49

scwhittle requested changes

View reviewed changes

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java

+                  private enum KeyExistence {
+                    // this key is known to exist, it has at least 1 value in either localAdditions or windmill
+                    KNOWN_EXIST,
+                    // this key is known to be nonexistent, it has 0 value in both localAdditions and windmill

Contributor

scwhittle Mar 24, 2023

s/0 value/no values/

Contributor Author

zhengbuqian Mar 29, 2023

Done

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java

+                        if (allKeysKnown) {
+                          keyState = keyStateMap.get(structuralKey);
+                          if (keyState == null || keyState.existence == KeyExistence.UNKNOWN_EXISTENCE) {
+                            if (keyState != null) keyStateMap.remove(structuralKey);

Contributor

scwhittle Mar 24, 2023

I think it would be simpler to not remove in this read if we're not inserting.

It seems like we shouldn't have UNKNOWN_EXISTENCE in the map except when we construct new entries. But if that's not the case for some reason it seems simpler to just leave it as is here.

Contributor Author

zhengbuqian Mar 29, 2023

UNKNOWN_EXISTENCE is needed in situations like when a key doesn't exist in the map, and we need to check windmill to find out its existence. Before the windmill read returns result, we need to mark it as UNKNOWN_EXISTENCE.

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated

+                        ByteString encodedKey = keyStream.toByteStringAndReset();
+                        Windmill.TagMultimapEntry.Builder entryBuilder = builder.addUpdatesBuilder();
+                        entryBuilder.setEntryName(encodedKey);
+                        entryBuilder.setDeleteAll(keyState.removedLocally);

Contributor

scwhittle Mar 24, 2023

I meant leave proto to default of false if false
if (keyState.removedLocally) entryBuilder.setDeleteAll(true);

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated

+                      keyState.valuesSize = 0;
+                      keyState.existence = KeyExistence.KNOWN_NONEXISTENT;
+                    } else {
+                      // no data in windmill, deleting from local cache is sufficient.

Contributor

scwhittle Mar 24, 2023

Thanks, I think that inversion is easier to read, can you use it and swap the cases?
But shouldn't such a key be KNOWN_NONEXISTENT and handled above?
Maybe the special handling can just be removed and can assert that it is in the expected KNOWN state?

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternals.java Outdated

+                                  keyState.existence = KeyExistence.KNOWN_EXIST;
+                                  keyState.values.extendWith(entry.getValue());
+                                  keyState.valuesSize += Iterables.size(entry.getValue());
+                                  // We can't set keyState.valuesCached to true here, because there may be more

Contributor

scwhittle Mar 24, 2023

That is the values within the bag page. But the iterable which you are examining here is either

those values and thus weighted if a single page
or a PagingIterable (which is not weighted)

...va/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateReaderTest.java

+                    entry.getValue().forEach(expectedMap.get(key)::add);
+                  }
+                  for (Map.Entry<ByteString, List<Integer>> entry : actual) {
+                    assertThat(

Contributor

scwhittle Mar 24, 2023

explicit check that expectedMap contains the key?
to avoid weird possiblities that returning null from remove possibly matches some weird actual state.

Contributor Author

zhengbuqian Mar 29, 2023

Done

...va/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateReaderTest.java Outdated

+                  Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest2.build());
+                  Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest3.build());
+                  Mockito.verifyNoMoreInteractions(mockWindmill);
+                  // NOTE: The future will still contain a reference to the underlying reader.

Contributor

scwhittle Mar 24, 2023

ditto

Contributor Author

zhengbuqian Mar 29, 2023

Done

...worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternalsTest.java Outdated

@@ @@ -645,6 +655,1163 @@ public void testMapComplexPersist() throws Exception { @@
                   assertEquals(0, commitBuilder.getValueUpdatesCount());
                 }
+                private static <T> ByteString encodeWithCoder(T key, Coder<T> coder) throws IOException {

Contributor

scwhittle Mar 24, 2023

just rethrow as RuntimeException to avoid having to add IOException to everything when not expected

Contributor Author

zhengbuqian Mar 29, 2023

Done

...worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternalsTest.java Outdated

+                }
+                @Test
+                public void testMultimapCachedPartialEntryCannotCachePolled() throws IOException {

Contributor

scwhittle Mar 24, 2023

test name is a little unclear
how about a comment explaining the case

Contributor Author

zhengbuqian Mar 29, 2023

Renamed the test.

...worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternalsTest.java

+                      ByteArrayCoder.of().decode(entryUpdate.getEntryName().newInput(), Context.OUTER);
+                  assertTrue(Arrays.equals(key1, decodedKey));
+                  assertTrue(entryUpdate.getDeleteAll());
+                }

Contributor

scwhittle Mar 24, 2023

these look good but with such a complex class it is hard to cover everything. A fuzz test might help ensure that we're not missing something. You could do something like:

have two maps and in a loop

randomly make a modification to both maps (add value to 100 random keys, remove 100 random keys, clear map)
perform operations like persist, read that shouldn't affect viewed entries on one of the maps randomly

And then at the end or periodically verify that both maps have the same observed state by iterating over both. Would be a nice sanity check that things are always consistent and we covered everything.

Contributor Author

zhengbuqian Apr 19, 2023

Added a fuzz test to perform different modifications and verify the state.

Contributor

scwhittle commented Apr 18, 2023

@zhengbuqian Any updates on this PR?


          Address comments

47b9294

zhengbuqian requested a review from scwhittle

April 19, 2023 23:15

scwhittle reviewed

View reviewed changes

...worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillStateInternalsTest.java Outdated

+                    }
+                    Windmill.WorkItemCommitRequest.Builder commitBuilder =
+                        Windmill.WorkItemCommitRequest.newBuilder();
+                    underTest.persist(commitBuilder);

Contributor

scwhittle Apr 20, 2023

can you also recreate the multimap after some rounds so that it has to initialize from persisted state as necessary? As is it is fully cached all the time and thus is likely not merging in-memory or on-disk state.

Contributor Author

zhengbuqian Apr 25, 2023

Done. Now it clears cache after 100 rounds, and ran another 100 rounds. In between it calls keys() and get() to rebuild cache.


          Address comments

52f6338

zhengbuqian requested a review from scwhittle

April 25, 2023 08:45

scwhittle approved these changes

View reviewed changes

Contributor

scwhittle left a comment

Thanks Buqian for all your work on this!
I'm not a committer so adding
R: @reuvenlax

Contributor

scwhittle commented Apr 26, 2023

Run Java PreCommit

Contributor

scwhittle commented Apr 26, 2023

precommit failure looked unrelated so rerunning

Contributor

scwhittle commented Jun 2, 2023

Run Java PreCommit

Contributor

scwhittle commented Jun 7, 2023

@reuvenlax can you merge?

Contributor Author

zhengbuqian commented Jun 7, 2023

thanks @scwhittle! If more changes are needed pls let me know, I'll work on fixes!

y1chi merged commit 2a12765 into apache:master

y1chi mentioned this pull request

Fix windmill proto #27361

Merged

3 tasks

andreydevyatkin pushed a commit to akvelon/beam that referenced this pull request


          Add Windmill support for MultimapState (apache#23492)

af05af1

aleksandr-dudko pushed a commit to aleksandr-dudko/beam that referenced this pull request


          Add Windmill support for MultimapState (apache#23492)

b2c7a69

aleksandr-dudko pushed a commit to aleksandr-dudko/beam that referenced this pull request


          Add Windmill support for MultimapState (apache#23492)

4c32e56

bullet03 pushed a commit to akvelon/beam that referenced this pull request


          Add Windmill support for MultimapState (apache#23492)

e52da7a

cushon pushed a commit to cushon/beam that referenced this pull request


          Add Windmill support for MultimapState (apache#23492)

fcd8a19

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

dataflow runners