Implement ordered list state for FnApi. #30317

shunping · 2024-02-14T20:49:36Z

First attempt to implement ordered list state for FnApi. Notice that caching has not been implemented yet.

Reference:
https://docs.google.com/document/d/1U77sAvE6Iy9XsVruRYHxPdFji7nqS6HPi1XU8fhyrxs/

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.

If there are changes on a state after we obtain iterators from calling read() and readRange(), the behavior of these pre-existing iterators were incorrect in the previous implementation. The change introduced here will make sure that these iterators will still work as if no local change is made.

shunping · 2024-02-15T16:53:53Z

Run Python PreCommit 3.9

scwhittle

Didn't get yet to implementation/test

scwhittle · 2024-02-15T19:40:11Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+// A response to the get state request for an ordered list.
+message OrderedListStateGetResponse {
+  bytes continuation_token = 1;
+  bytes data = 2;


how do we return multiple elements if the request was a range?

should we return the sort-key for elements, that seems like part of the user-data for example if it's some id/timestamp the user might want it back instead of having to duplicate it in the payload as well.

should this be repeated OrderedListEntry instead of data?

For efficiency reasons we have generally not split individual elements up into individual field protos, and provided them as contiguous bytes of data. Makes sense to do that here. But the coding should be specified (e.g. is this a concatenation of several encoded KV<sort-key, value>s?

We should also consider if there are considerations in the encoding of sort-key. If they're (typically) timestamps, bigendian might be preferable to varint.

how do we return multiple elements if the request was a range?

We concatenate the encoded entries in a byte array and send them back in chunks with corresponding continuation token.

should this be repeated OrderedListEntry instead of data?

I have considered this option. Besides the efficiency reason @robertwb mentioned, I also find that representing the response as bytes has an advantage of reusing the existing code in https://github.com/apache/beam/blob/3693174c0421d0ff049042ca283db633431892ef/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/StateFetchingIterators.java. This iterator is used to parsed the returned data block (in not only OrderedListState but also Multimap, Bag, etc) and it supports blocks even if the boundary is not aligned with entries/elements. I think this is not achievable with OrderedListEntry.

But the coding should be specified (e.g. is this a concatenation of several encoded KV<sort-key, value>s

That's right. It is basically the concatenation of encoded TimestampedValue, since TimestampedValue is already in use in the sdk interface of OrderedListState

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

Line 29 in 3693174

extends GroupingState<TimestampedValue<T>, Iterable<TimestampedValue<T>>> {

scwhittle · 2024-02-15T19:41:18Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+// timestamp range is not specified, the runner should use [MIN_INT64,
+// MAX_INT64) by default.
+message OrderedListStateGetRequest {
+  bytes continuation_token = 1;


comment on continuation token, ie should be returned by previous response and the key/range should match the previous request generating the token

Do we want to require returning the range if there's a continuation token involved? Or should they be mutually exclusive?

comment on continuation token ...

Good idea! I will add that when I revise the code.

Do we want to require returning the range if there's a continuation token involved? Or should they be mutually exclusive?

I don't think the range is a hard requirement.
In fact in my simple implementation of the fake client, I put the current sort key and the current block index into the continuation token:

beam/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/FakeBeamFnStateClient.java

Line 228 in 5d7cd5e

// The continuation format here is the sort key (long) followed by an index (int)

.
In other implementation, you may need the range, but I think it is implementation-dependent. That's why I hesitate to put a range/sort key as a separate field when continuation token is present.

I thought the continuation token should allow the runner to uniquely identify where to find the next block of data. Is that what you mean by "mutually exclusive"?

Yes, that's what I meant by mutually exclusive. If we allow providing both, we have to figure out what to do when they disagree, and there's no good usecase for that.

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

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

robertwb · 2024-02-15T22:17:08Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+// timestamp range is not specified, the runner should use [MIN_INT64,
+// MAX_INT64) by default.
+message OrderedListStateGetRequest {
+  bytes continuation_token = 1;


Do we want to require returning the range if there's a continuation token involved? Or should they be mutually exclusive?

robertwb · 2024-02-15T22:18:56Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+// A response to the get state request for an ordered list.
+message OrderedListStateGetResponse {
+  bytes continuation_token = 1;
+  bytes data = 2;


For efficiency reasons we have generally not split individual elements up into individual field protos, and provided them as contiguous bytes of data. Makes sense to do that here. But the coding should be specified (e.g. is this a concatenation of several encoded KV<sort-key, value>s?

robertwb · 2024-02-15T22:27:17Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+
+// A request to update an ordered list
+message OrderedListStateUpdateRequest {
+  // when the request is processed, deletes should always happen before inserts.


Should we instead split this up into two separate requests, e.g. OrderedListStateInsertRequest and OrderedListStateDeleteRequest?

Discussion 2 in the design doc talked about this, but there is no conclusion on that https://docs.google.com/document/d/1U77sAvE6Iy9XsVruRYHxPdFji7nqS6HPi1XU8fhyrxs/

I am fine with splitting them though.

It would be helpful to outline the pro/con in the design doc of little decisions like, and note which one was chosen and why.

For example one benefit to splitting the requests is to avoid ordering issues. We would have to spec that either the inserts or deletes happen first, even though they are in one request together. It is a bit confusing. And then if you want them in the other order, you still have to make two requests but each one has an empty field.

And note whether there is an efficiency consideration.

Yep. I am planning to add an addendum to the original design doc to summarize the decisions we make here. We should have that after this round of review completes.

robertwb · 2024-02-15T22:30:27Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+message OrderedListStateUpdateRequest {
+  // when the request is processed, deletes should always happen before inserts.
+  repeated OrderedListRange deletes = 1;
+  repeated OrderedListEntry inserts = 2;


Should there be symmetry between the written and read data? E.g. one writes bytes that are encoded KV<sort_key, data> and then reads them?

robertwb · 2024-02-15T22:37:05Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

+// A response to the get state request for an ordered list.
+message OrderedListStateGetResponse {
+  bytes continuation_token = 1;
+  bytes data = 2;


We should also consider if there are considerations in the encoding of sort-key. If they're (typically) timestamps, bigendian might be preferable to varint.

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/FakeBeamFnStateClient.java

robertwb · 2024-02-15T22:45:24Z

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/FakeBeamFnStateClient.java

+          ;
+          try {
+            if (sortKey < start || sortKey >= end) {
+              throw new IndexOutOfBoundsException("sort key out of range");


OK, this goes back to the idea that range and (opaque) continuation token should be mutually exclusive.

See my comment in a previous thread: #30317 (comment)

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/FakeBeamFnStateClient.java

* The range information is placed in the state key of ordered list * For consistency, we reuse the existing get request and response mesasages of other states like Bag, MultiMap, etc.

* Reuse existing messages of clear and append.

* Replace String::size() > 0 with String::isEmpty() * Return this in readLater and readRangeLater instead of throwing an exception * Remove the added SupressWarnings("unchecked")

Previously, we used a repeated OrderedListEntry field in the AppendRequest particularly for ordered list state. For consistency, we now get rid of that and use the same data field as other states.

github-actions · 2024-03-06T17:36:06Z

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

R: @robertwb for label java.

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

shunping · 2024-03-06T20:51:16Z

Run Java PreCommit

acrites · 2024-03-18T20:50:07Z

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

+                      createOrderedListUserState(key, elemCoder);
+
+                  @Override
+                  public void clear() {


Do we want a separate notion of "clear"? i.e. one that deletes all state associated with this OrderedList and not just clearRange(min, max), which would only delete all the elements from the OrderedList?

The clear() function in OrderedListState is quite similar to its counterpart in MultiMap, and I don't think we ever has the notion of deleting all states associated with a state object. There is one function called "asyncClose()" which will sync the local copy with the remote one and then invalidate any states associated with the state object.

On a different topic, I notice that we have a clear() function in the implementation, which handles the special case of deleting all elements from an OrderedList more efficiently than clearRange(min, max). I should probably use that instead.

acrites · 2024-03-18T20:52:15Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

@@ -1075,6 +1103,12 @@ message StateClearRequest {}
 // A response to clear state.
 message StateClearResponse {}

+// A message describes a sort key range [start, end).


Maybe add a comment about these values needing to be within [BoundedWindow.TIMESTAMP_MIN_VALUE, BoundedWindow.TIMESTAMP_MAX_VALUE)? Should these be the default values for these fields?

acrites · 2024-03-18T20:52:49Z

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

+import org.joda.time.Instant;
+
+/**
+ * An implementation of a bag user state that utilizes the Beam Fn State API to fetch, clear and


ordered list

acrites · 2024-03-18T21:38:52Z

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

+    // Remove items (in a collection) in the specific range from pendingAdds.
+    // The old values of the removed sub map are kept, so that they will still be accessible in
+    // pre-existing iterables even after the sort key is cleared.
+    pendingAdds.subMap(minTimestamp, true, limitTimestamp, false).clear();


I think this is the same as just calling .subMap(minTimestamp, limitTimestamp).

You are right. Will make the change.

acrites · 2024-03-18T21:49:49Z

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

+  public void asyncClose() throws Exception {
+    isClosed = true;
+
+    if (!pendingRemoves.isEmpty()) {


Can processing the pendingRemoves and pendingAdds be combined into a single StateRequest across the FnApi? Fewer roundtrips would probably be better.

Also, it'd be nice if we could clear multiple ranges in a single request.

Can processing the pendingRemoves and pendingAdds be combined into a single StateRequest across the FnApi? Fewer roundtrips would probably be better.

Hmmmm, again this is about whether we want to be consistent with other states and reuse the existing StateRequest OR create a new one particularly for OrderListState.

We had this discussion previously at #30317 (comment), and I think the decision was to follow the existing states and reuse the current StateRequest, which means we will have Append to do the insertion only and Clear to do the deletion.

Also, it'd be nice if we could clear multiple ranges in a single request.

There is no extra field in Clear request message; it is invoked on a state key that contains a single range. That's why we only do one range at a time.

acrites · 2024-03-18T21:54:26Z

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

+    //   (1) a sort key is added to or removed from pendingAdds, or
+    //   (2) a new value is added to an existing sort key
+    ArrayList<PrefetchableIterable<TimestampedValue<T>>> pendingAddsInRange = new ArrayList<>();
+    for (Entry<Instant, Collection<T>> kv :


I'm not super-clear on how Iterables work in Java. Does this actually copy the elements, or is it just wrapping the values in pendingAdds in an Iterable? I'm curious about the statements in the comments saying "the values are kept" so that we can make modifications to the ordered list and not mess up existing Iterables.

Does this actually copy the elements, or is it just wrapping the values in pendingAdds in an Iterable?

Iterables in java work like iterables in other languages. Basically, we use it to traverse an existing collection of elements without making a copy of them.

I'm curious about the statements in the comments saying "the values are kept" so that we can make modifications to the ordered list and not mess up existing Iterables.

The PrefetchableIterable interface in Beam is specifically for iterables that support prefetching. The function PrefetchableIterables.limit() is applied on an iterable, resulting in a new iterable, which can be used to traverse its backend collection until the number of elements reaches a preset limit.

Here is the tricky part.

Note that pendingAddsInRange is a list of iterables. For each of these iterables, when an iteration is performed at any time after its creation, the result will be truncated to the last element we see at the time we initialize the iterable with PrefetchableIterables.limit(). In other words, appending a new element to the end of an existing key of pendingAdds would not affect the outcome above, because the iterable in pendingAddsInRange is designed to be truncated to the size when it is created.

When we add a new key to pendingAdds, it won't change the outcome of traversing an pre-existing pendingAddsInRange either, because the iterable of the new key is not included in pendingAddsInRange.

When we delete an existing key from pendingAdds, we remove the key-value mapping, but not touch its value (i.e. the collection of elements). Therefore, the iterables in pre-existing pendingAddsInRange will still work as the backend collection is accessible. Only when no iterables reference these collections will they be garbage collected.

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/OrderedListUserStateTest.java

acrites · 2024-03-18T22:28:00Z

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/OrderedListUserStateTest.java

+  }
+
+  @Test
+  public void testAdd() throws Exception {


Do we need to have some tests where we're partway through an Iterable when the element gets added? In all of the ones I see here we completely process each Iterable before we do any further operations on it (by calling Iterables.toArray).

Good point! I will add some more tests to cover this.

acrites · 2024-03-20T00:06:47Z

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

Do we also need to add something here:

beam/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto

Line 478 in fb7ba65

// TODO(https://github.com/apache/beam/issues/20486): Add protocol to support OrderedListState

Or are we going to piggy-back on multimap for this? (If so we should delete the TODO.)

Yes. I think we will need a separate one for orderedlist because although orderedlist and multimap have similar semantics, the code paths are totally different.

github-actions · 2024-03-22T12:13:51Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.

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)

github-actions · 2024-03-29T12:14:00Z

Reminder, please take a look at this pr: @kennknowles

github-actions · 2024-04-03T12:13:46Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @Abacn for label java.

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)

Abacn · 2024-04-03T14:43:44Z

waiting on author

github-actions · 2024-05-11T12:13:47Z

Reminder, please take a look at this pr: @Abacn

github-actions · 2024-05-15T12:14:17Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.

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)

shunping · 2024-05-15T13:35:25Z

stop reviewer notifications

github-actions · 2024-05-15T13:36:36Z

Stopping reviewer notifications for this pull request: requested by reviewer

shunping added 8 commits February 13, 2024 15:16

Add request and response proto messages for ordered list state.

b978ca7

Initial implementation of OrderedListState for fnApi.

68abe65

Discard the use of the value coder in FakeBeamFnStateClient.

d06e0a5

Support continuation token for ordered list get request in fake client

f74fc60

Add copyright notices to the new files

f9e48a3

Add binding for ordered list state in fnapi state accessor

a76b523

Clean up comments

8a3b719

shunping marked this pull request as draft February 14, 2024 20:49

github-actions bot added java model labels Feb 14, 2024

Apply spotless and checkStyle to reformat

5d7cd5e

shunping force-pushed the ordered-list-state-2 branch from ac40716 to 5d7cd5e Compare February 15, 2024 15:16

scwhittle requested changes Feb 15, 2024

View reviewed changes

robertwb reviewed Feb 15, 2024

View reviewed changes

shunping added 9 commits February 21, 2024 16:20

Add an encode-only coder for the use in the fake client.

33e748d

Remove request and response messages for ordered list state get.

91b770e

* The range information is placed in the state key of ordered list * For consistency, we reuse the existing get request and response mesasages of other states like Bag, MultiMap, etc.

Remove request and response messages for ordered list state update

4b512ea

* Reuse existing messages of clear and append.

Minor fixes based on feedbacks from reviewers

09db01c

* Replace String::size() > 0 with String::isEmpty() * Return this in readLater and readRangeLater instead of throwing an exception * Remove the added SupressWarnings("unchecked")

Apply spotless

44b6679

Use data field in AppendRequest for ordered list state

868e493

Previously, we used a repeated OrderedListEntry field in the AppendRequest particularly for ordered list state. For consistency, we now get rid of that and use the same data field as other states.

Merge branch 'master' into ordered-list-state-2

ea7ca70

Apply spotless

83bac80

Minor renaming of a variable

90b7f7d

shunping force-pushed the ordered-list-state-2 branch from 5a438a3 to 90b7f7d Compare March 6, 2024 16:40

shunping marked this pull request as ready for review March 6, 2024 16:42

github-actions bot added the Next Action: Reviewers label Mar 6, 2024

acrites reviewed Mar 18, 2024

View reviewed changes

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/OrderedListUserStateTest.java Show resolved Hide resolved

acrites reviewed Mar 18, 2024

View reviewed changes

acrites reviewed Mar 20, 2024

View reviewed changes

github-actions bot removed the slow-review label Mar 22, 2024

github-actions bot added the slow-review label Mar 29, 2024

github-actions bot removed the slow-review label Apr 3, 2024

github-actions bot added Next Action: Author and removed Next Action: Reviewers labels Apr 3, 2024

robertwb mentioned this pull request Apr 24, 2024

FnAPI proto changes for ordered list state. #31092

Merged

3 tasks

Sync with master and apply proto changes from PR 31092

8900646

github-actions bot added model Next Action: Reviewers and removed model Next Action: Author labels May 3, 2024

github-actions bot added the slow-review label May 11, 2024

github-actions bot removed the slow-review label May 15, 2024

Implement ordered list state for FnApi. #30317

Are you sure you want to change the base?

Implement ordered list state for FnApi. #30317

Conversation

shunping commented Feb 14, 2024 • edited

GitHub Actions Tests Status (on master branch)

shunping commented Feb 15, 2024

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping Feb 16, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping Feb 16, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping Feb 23, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Mar 6, 2024

shunping commented Mar 6, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping May 2, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping May 2, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shunping May 2, 2024 • edited

Choose a reason for hiding this comment

github-actions bot commented Mar 22, 2024

github-actions bot commented Mar 29, 2024

github-actions bot commented Apr 3, 2024

Abacn commented Apr 3, 2024

github-actions bot commented May 11, 2024

github-actions bot commented May 15, 2024

shunping commented May 15, 2024

github-actions bot commented May 15, 2024

shunping commented Feb 14, 2024 •

edited

shunping Feb 16, 2024 •

edited

shunping Feb 16, 2024 •

edited

shunping Feb 23, 2024 •

edited

shunping May 2, 2024 •

edited

shunping May 2, 2024 •

edited

shunping May 2, 2024 •

edited