[FLINK-7475] [core][DataStream API] support update() in ListState #4963

bowenli86 · 2017-11-07T00:48:31Z

What is the purpose of the change

If users want to update the list, they have to do two steps:

listState.clear() 
for (Element e : myList) { 
    listState.add(e); 
}

We should enable users to do such actions by providing an API listState.update(myNewList)

Brief change log

Added update() API to ListState
Added and updated unit/IT tests
Updated Flink doc

update() can actually be an API in AppendingState, or at least in MapState. We can consider it later.

Verifying this change

This change added tests and can be verified in testListStateAddUpdateAndGet()

Does this pull request potentially affect one of the following parts:

The public API, i.e., is any changed class annotated with @Public(Evolving): (yes)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (docs and JavaDocs)

Performance Benchmarking

Computer: MacbookPro (Mid 2015), Flash Storage, Processor 2.5GHz Intel Core i7, Memory 16GB 1600MHz DDR3

Number of values added | time for add() API | time for update() API | perf improvement of update() over add()

500 978703 ns 55397 ns 17.66x
1000 3044179 ns 89474 ns 34.02x
5000 9247395 ns 305580 ns 30.26x
10000 16416442 ns 605963 ns 27.09x
50000 84311205 ns 5691288 ns 14.81x
100000 195103310 ns 12914182 ns 15.11x
500000 1223141510 ns 70595881 ns 17.33x

In summary, update() API which pre-merges all values gives users 15-35x performance improvements.
For most frequent use cases where there are a few hundreds to a few thousands values per key, users can get 30x - 35x performance improvement!

yunfan123 · 2017-11-10T10:04:18Z

In the rocksdb implement, it still use db.merge to add each element to the rocksdb.
I think we can get the specific MergeOperator class from rocksdb config and merge the list as a string.
So we can only use db.put once to update the list.

bowenli86 · 2017-11-11T23:18:05Z

Hi @yunfan123 Thanks for the feedback, and I agree with you.

Well, here's the problem: Flink seems to be using the StringAppendTestOperator as merge operator from RocksDB, which added a ',' between values. But I didn't find how a java class for StringAppendTestOperator.

Shall we mimic StringAppendTestOperator and concatenate the values with ',' ourselves? Do you have any more suggestions?

yunfan123 · 2017-11-13T03:07:51Z

Flink use StringAppendOperator as merge operator. It used in org.apache.flink.contrib.streaming.state.PredefinedOptions.
It use native java method.
Can we use the native method to merge our list?

bowenli86 · 2017-11-13T04:40:49Z

@yunfan123 sorry I can't find where PredefinedOptions defines it. Isn't Flink using RocksDB's StringAppendTESTOperator in RocksDBKeyedStateBackend? And the StringAppendOperator java class in rocksdbjni doesn't expose any methods for us to use.

Hi @aljoscha , do you have insightful suggestions?

aljoscha · 2017-11-13T09:44:05Z

I don't know if there's a way of retrieving the class behind the "stringappendtest" merge operator. If not, we should just simulate the behaviour of appending with commas in Java code. @StefanRRichter what do you think?

StefanRRichter · 2017-11-13T09:50:47Z

@aljoscha IIRC there is no class for StringAppendTESTOperator in the RocksDB Java API. You can only pass a string that is used inside some factory in the native RocksDB code to instantiate the append operator. I would also suggest to simulate it, either "shallow" by adding commas or "deeper" by using a dummy RocksDB instance internally to actually perform the merge and query for the result.

bowenli86 · 2017-11-14T21:18:24Z

@yunfan123 @aljoscha @StefanRRichter I chose the "shallow" simulation. What do you guys think about the new code?

The build failure seems to be because one build profile timed out.

bowenli86 · 2018-01-03T05:18:04Z

@yunfan123 @aljoscha @StefanRRichter feedbacks are appreciated

StefanRRichter · 2018-01-03T09:39:51Z

...ebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java

+		}
+
+		if (operands.size() < 2) {
+			return operands.get(0);


This will fail if size == 0.

...which, in turn, is a corner case that the test should cover.

StefanRRichter · 2018-01-03T09:48:12Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java

+		public void update(List<S> values) throws Exception {
+			internalList.clear();
+
+			if (values != null && !values.isEmpty()) {


nit: I think the second check is not required and makes the statement more verbose for no good reason.

it makes a difference as we chatted about empty list case

StefanRRichter · 2018-01-03T09:49:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java

+			final N namespace = currentNamespace;
+			final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+			map.put(namespace, new ArrayList<>(values));


Why do we need to make a defensive copy of the list?

because values is of List and map only is of ArrayList, need to enforce strict subclass

Then, I wonder if changing the generic type of the map to List is not the better solution? Maybe @StephanEwen had the intention to enforce a somewhat efficient list implementation here?

Correct, I wanted to enforce using ArrayList in the internal state when possible, because users never interacted with the list directly. We don't necessarily have to keep that, could relax it to List to avoid extra copies. That should be a ground rule in all of Flink's runtime code: No extra work to work around current code.

Background: I initially strongly typed to ArrayList to make it clear that we want a compact and efficient list implementation. Because I have seen too many times that LinkedList (which is for most cases so much slower) is used as the default list (I blame this on University education, which talks about how theoretical complexity of LinkedList is lower, but fails to actually take processor architecture into account)

@StephanEwen @StefanRRichter I tried to relax the type but the change scope is much bigger than I thought. I recommend handle that as a separate task. I filed FLINK-8365 and listed a couple issues I found when trying to relax the type.

StefanRRichter · 2018-01-03T09:49:56Z

...kend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/util/MergeUtilsTest.java

+ */
+public class MergeUtilsTest {
+	@Test
+	public void testMerge() {


This test should cover more corner cases, like empty list, singleton list, ...

StefanRRichter · 2018-01-03T09:51:02Z

...ebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java

+ */
+public class MergeUtils {
+	@VisibleForTesting
+	protected static final byte DELIMITER = ',';


It would be great if we could initialize this from the RocksDB API somehow. But not sure if this is possible.

RocksDB doesn't expose that

Ok, that is unfortunate.

StefanRRichter · 2018-01-03T09:53:22Z

...backend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java

+	public void update(List<V> values) throws Exception {
+		clear();
+
+		if (values != null && !values.isEmpty()) {


Is it expected behaviour that adding an empty list will not result in an entry? Maybe that is correct, but we should double check.

I believe it's the behavior in rocksdb. Good to have a second eye on it https://github.com/facebook/rocksdb/blob/master/utilities/merge_operators/string_append/stringappend2.cc since I never wrote c++ before

What I meant, is not a RocksDB question, but a question if the list state should return null or empty iterator if after the user stored an empty list under a key. That might make a difference and right now, I the caller updates with empty list, the get is identical to a non-existing mapping (null). Returning an empty list would require to make an entry in rocksdb.

in that case, for all state backends, it returns null. I will add a unit test to verify and enforce it.

StefanRRichter · 2018-01-03T09:55:33Z

...backend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java

+					bytes.add(keySerializationStream.toByteArray());
+				}
+
+				backend.db.put(columnFamily, writeOptions, key, MergeUtils.merge(bytes));


Do we have some numbers that this custom merging is better than multiple calls to RocksDB merging state? I can see that it is potentially a lot better, but just making sure it is worth the additional complexity.

good idea. I will do some benchmarking

StefanRRichter · 2018-01-03T23:00:28Z

flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java

+	 *
+	 * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+	 */
+	void update(List<T> values) throws Exception;


We should define and document what happens if this is called with null or empty list. Does this lead to different results for get() ?

Let me document it. They are of the same result - return null when given an empty list

Ok, if the outcome for both cases should be the same, wouldn't it be nicer to always return the empty iterator and avoid nulls if we can? This might be a small thing but save users from debugging some NPEs :-)

@StefanRRichter agree. The correct way to do should be changing ListState#get(). I filed FLINK-8364 and will work on it after this PR is checked in

bowenli86 · 2018-01-04T05:20:27Z

@StefanRRichter I added a perf benchmarking unit test, you can find the perf results in both this PR's description and the comment of that test. The perf improvement is pretty magnificent, between 15x -35x faster.

bowenli86 · 2018-01-04T05:25:30Z

...java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java

+ *
+ * Number of values added | time for add()   |  time for update() | perf improvement of update() over add()
+ * 500						978703 ns			55397 ns			17.66x
+ * 1000						3044179 ns			89474 ns			34.02x


I'm not sure why the formatting is a few spaces behind others on github. They are aligned on my computer

Ok, that's an improvement that would justify the more complex approach. Nice!

StefanRRichter · 2018-01-09T13:18:11Z

Alright, I think this looks good to me now. @bowenli86 would you also like to work on the followup issues that you created?

I will merge this today or tomorrow.

bowenli86 · 2018-01-09T17:12:44Z

@StefanRRichter Thanks! I will

bowenli86 changed the title ~~[FLINK-7475] [core][state backend] support update() in ListState~~ [FLINK-7475] [core][DataStream API] support update() in ListState Nov 8, 2017

StefanRRichter reviewed Jan 3, 2018

View reviewed changes

bowenli86 added 13 commits January 3, 2018 17:34

update local branch

e06ac7e

remove sh

8df3cf1

[FLINK-7475] add ListState#update() API and implementations to Flink

25e9429

add IT tests

c01be2b

update doc

1ad59b3

format comments

b565478

checkout empty list

a148b1a

format code

7dfc24e

add update() to FlinkKafkaConsumerBaseTest

a430024

fix build

f8a5bda

update impl

793c056

simulate StringAppendTESTOperator

5a73127

add more tests

c37d926

bowenli86 added 4 commits January 3, 2018 17:35

remove imports

831159e

enforce update(emptyList) remains the value null

9afa713

add perf unit test

795ea1d

change default number of values

97d7cfa

format comment

ee00a8a

bowenli86 commented Jan 4, 2018

View reviewed changes

fix checkstyle

a1bde36

asfgit closed this in 438e4e3 Jan 10, 2018

bowenli86 mentioned this pull request Jan 12, 2018

[FLINK-7938] [State Backend] support addAll() in ListState #5281

Closed

rmetzger added component=API/DataStream component=Runtime/StateBackends labels Mar 18, 2019

[FLINK-7475] [core][DataStream API] support update() in ListState #4963

[FLINK-7475] [core][DataStream API] support update() in ListState #4963

Conversation

bowenli86 commented Nov 7, 2017 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Performance Benchmarking

yunfan123 commented Nov 10, 2017

bowenli86 commented Nov 11, 2017

yunfan123 commented Nov 13, 2017

bowenli86 commented Nov 13, 2017 • edited

aljoscha commented Nov 13, 2017

StefanRRichter commented Nov 13, 2017

bowenli86 commented Nov 14, 2017 • edited

bowenli86 commented Jan 3, 2018

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

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

StefanRRichter Jan 3, 2018 • 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

StefanRRichter Jan 4, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bowenli86 commented Jan 4, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter commented Jan 9, 2018

bowenli86 commented Jan 9, 2018

bowenli86 commented Nov 7, 2017 •

edited

bowenli86 commented Nov 13, 2017 •

edited

bowenli86 commented Nov 14, 2017 •

edited

StefanRRichter Jan 3, 2018 •

edited

StefanRRichter Jan 4, 2018 •

edited