You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by bowenli86 <gi...@git.apache.org> on 2017/11/07 00:48:35 UTC
[GitHub] flink pull request #4963: [FLINK-7475] [core][state backend] support update(...
GitHub user bowenli86 opened a pull request:
https://github.com/apache/flink/pull/4963
[FLINK-7475] [core][state backend] support update() in ListState
## What is the purpose of the change
If users want to update the list, they have to do two steps:
```java
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)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bowenli86/flink FLINK-7475
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4963.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4963
----
commit 6ee905066137bafe2619be24446a7ed3439ec65d
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T08:21:20Z
[FLINK-7475] add ListState#update() API and implementations to Flink
commit 390b5656988505c69367118ed295dd83cb7ae784
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T08:59:08Z
add IT tests
commit e63baa1fae56681d3760ba833b5da8c0cfbcae1a
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T09:04:04Z
update doc
commit 03be4982a490df8d4d8da805916c9a29c9174ebd
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T23:17:50Z
format comments
commit 75dcdc134f705a3a997d65d6c547fd3a35d19edd
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T23:23:55Z
checkout empty list
commit 18a952c9978b24287c01aa865d8f28beb4fe31ec
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-27T23:42:19Z
format code
commit c4daba92b04d3789aeb7f72419b7fdd36e11177b
Author: Bowen Li <bo...@gmail.com>
Date: 2017-10-28T05:12:18Z
add update() to FlinkKafkaConsumerBaseTest
----
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/4963
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?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159386696
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java ---
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.List;
+
+/**
+ * Utils to simulate StringAppendTestOperator's merge operations in RocksDB.
+ */
+public class MergeUtils {
+ @VisibleForTesting
+ protected static final byte DELIMITER = ',';
+
+ /**
+ * Merge operands into a single value that can be put directly into RocksDB.
+ */
+ public static byte[] merge(List<byte[]> operands) {
+ if (operands == null) {
+ return null;
+ }
+
+ if (operands.size() < 2) {
+ return operands.get(0);
--- End diff --
...which, in turn, is a corner case that the test should cover.
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by yunfan123 <gi...@git.apache.org>.
Github user yunfan123 commented on the issue:
https://github.com/apache/flink/pull/4963
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?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159547336
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java ---
@@ -120,4 +121,16 @@ public void add(V value) {
a.addAll(b);
return a;
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ map.put(namespace, new ArrayList<>(values));
--- End diff --
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?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159546186
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java ---
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.List;
+
+/**
+ * Utils to simulate StringAppendTestOperator's merge operations in RocksDB.
+ */
+public class MergeUtils {
+ @VisibleForTesting
+ protected static final byte DELIMITER = ',';
--- End diff --
RocksDB doesn't expose that
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159782689
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java ---
@@ -120,4 +121,16 @@ public void add(V value) {
a.addAll(b);
return a;
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ map.put(namespace, new ArrayList<>(values));
--- End diff --
@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](https://issues.apache.org/jira/browse/FLINK-8365) and listed a couple issues I found when trying to relax the type.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159548273
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
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`).
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159631427
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java ---
@@ -33,4 +35,14 @@
* @param <T> Type of values that this list state keeps.
*/
@PublicEvolving
-public interface ListState<T> extends MergingState<T, Iterable<T>> {}
+public interface ListState<T> extends MergingState<T, Iterable<T>> {
+ /**
+ * Updates the state of the current key for the given source namespaces into the state of
+ * the target namespace.
+ *
+ * @param values The target namespace where the merged state should be stored.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void update(List<T> values) throws Exception;
--- End diff --
Ok, if the outcome for both cases should be the same, wouldn't it be nicer to always return the empty iterator and avoid `null`s if we can? This might be a small thing but safe users from debugging some NPEs :-)
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159387626
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java ---
@@ -497,6 +497,15 @@ public String toString() {
return partitionOffsets;
}
+
+ @Override
+ public void update(List<S> values) throws Exception {
+ internalList.clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
nit: I think the second check is not required and makes the statement more verbose for no good reason.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159582753
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java ---
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.benchmark;
+
+import org.apache.flink.contrib.streaming.state.PredefinedOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.contrib.streaming.state.util.MergeUtils;
+import org.apache.flink.testutils.junit.RetryOnFailure;
+import org.apache.flink.testutils.junit.RetryRule;
+import org.apache.flink.util.TestLogger;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.WriteOptions;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test that validates that the performance of APIs of RocksDB's ListState is as expected.
+ *
+ * Benchmarking
+ * Computer: MacbookPro (Mid 2015), Flash Storage, Processor 2.5GHz Intel Core i7, Memory 16GB 1600MHz DDR3
+ *
+ * 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
--- End diff --
I'm not sure why the formatting is a few spaces behind others on github. They are aligned on my computer
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159386188
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java ---
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.List;
+
+/**
+ * Utils to simulate StringAppendTestOperator's merge operations in RocksDB.
+ */
+public class MergeUtils {
+ @VisibleForTesting
+ protected static final byte DELIMITER = ',';
+
+ /**
+ * Merge operands into a single value that can be put directly into RocksDB.
+ */
+ public static byte[] merge(List<byte[]> operands) {
+ if (operands == null) {
+ return null;
+ }
+
+ if (operands.size() < 2) {
+ return operands.get(0);
--- End diff --
This will fail if size == 0.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159565552
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
in that case, for all state backends, it returns `null`. I will add a unit test to verify and enforce it.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159387906
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/util/MergeUtilsTest.java ---
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for MergeUtils.
+ */
+public class MergeUtilsTest {
+ @Test
+ public void testMerge() {
--- End diff --
This test should cover more corner cases, like empty list, singleton list, ...
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@StefanRRichter Thanks! I will
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@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.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159549032
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java ---
@@ -33,4 +35,14 @@
* @param <T> Type of values that this list state keeps.
*/
@PublicEvolving
-public interface ListState<T> extends MergingState<T, Iterable<T>> {}
+public interface ListState<T> extends MergingState<T, Iterable<T>> {
+ /**
+ * Updates the state of the current key for the given source namespaces into the state of
+ * the target namespace.
+ *
+ * @param values The target namespace where the merged state should be stored.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void update(List<T> values) throws Exception;
--- End diff --
We should define and document what happens if this is called with `null` or empty list. Does this lead to different results for `get()` ?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159549233
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java ---
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.List;
+
+/**
+ * Utils to simulate StringAppendTestOperator's merge operations in RocksDB.
+ */
+public class MergeUtils {
+ @VisibleForTesting
+ protected static final byte DELIMITER = ',';
--- End diff --
Ok, that is unfortunate.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159387760
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java ---
@@ -120,4 +121,16 @@ public void add(V value) {
a.addAll(b);
return a;
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ map.put(namespace, new ArrayList<>(values));
--- End diff --
Why do we need to make a defensive copy of the list?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/4963
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159388932
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ try {
+ writeCurrentKeyWithGroupAndNamespace();
+ byte[] key = keySerializationStream.toByteArray();
+ DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
+
+ List<byte[]> bytes = new ArrayList<>(values.size());
+ for (V value : values) {
+ keySerializationStream.reset();
+ valueSerializer.serialize(value, out);
+ bytes.add(keySerializationStream.toByteArray());
+ }
+
+ backend.db.put(columnFamily, writeOptions, key, MergeUtils.merge(bytes));
--- End diff --
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.
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@yunfan123 sorry I can't find where `PredefinedOptions` defines it. Isn't Flink using RocksDB's `StringAppendTESTOperator` in `RocksDBKeyedStateBackend`?
Hi @aljoscha , do you have insightful suggestions?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159582105
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java ---
@@ -497,6 +497,15 @@ public String toString() {
return partitionOffsets;
}
+
+ @Override
+ public void update(List<S> values) throws Exception {
+ internalList.clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
it makes a difference as we chatted about empty list case
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@yunfan123 @aljoscha @StefanRRichter I chose the "shallow" simulation. What do you guys think?
The build failure seems to be because one build profile timed out.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159546670
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ try {
+ writeCurrentKeyWithGroupAndNamespace();
+ byte[] key = keySerializationStream.toByteArray();
+ DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
+
+ List<byte[]> bytes = new ArrayList<>(values.size());
+ for (V value : values) {
+ keySerializationStream.reset();
+ valueSerializer.serialize(value, out);
+ bytes.add(keySerializationStream.toByteArray());
+ }
+
+ backend.db.put(columnFamily, writeOptions, key, MergeUtils.merge(bytes));
--- End diff --
good idea. I will do some benchmarking
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4963
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.
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4963
@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.
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@yunfan123 @aljoscha @StefanRRichter feedbacks are appreciated
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159388106
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/util/MergeUtils.java ---
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.util;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.List;
+
+/**
+ * Utils to simulate StringAppendTestOperator's merge operations in RocksDB.
+ */
+public class MergeUtils {
+ @VisibleForTesting
+ protected static final byte DELIMITER = ',';
--- End diff --
It would be great if we could initialize this from the RocksDB API somehow. But not sure if this is possible.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159762966
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java ---
@@ -33,4 +35,14 @@
* @param <T> Type of values that this list state keeps.
*/
@PublicEvolving
-public interface ListState<T> extends MergingState<T, Iterable<T>> {}
+public interface ListState<T> extends MergingState<T, Iterable<T>> {
+ /**
+ * Updates the state of the current key for the given source namespaces into the state of
+ * the target namespace.
+ *
+ * @param values The target namespace where the merged state should be stored.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void update(List<T> values) throws Exception;
--- End diff --
@StefanRRichter agree. The correct way to do should be changing `ListState#get()`. I filed [FLINK-8364](https://issues.apache.org/jira/browse/FLINK-8364) and will work on it after this PR is checked in
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159388540
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
Is it expected behaviour that adding an empty list will not result in an entry? Maybe that is correct, but we should double check.
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
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?
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159618800
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java ---
@@ -120,4 +121,16 @@ public void add(V value) {
a.addAll(b);
return a;
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ map.put(namespace, new ArrayList<>(values));
--- End diff --
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)
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159565645
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java ---
@@ -33,4 +35,14 @@
* @param <T> Type of values that this list state keeps.
*/
@PublicEvolving
-public interface ListState<T> extends MergingState<T, Iterable<T>> {}
+public interface ListState<T> extends MergingState<T, Iterable<T>> {
+ /**
+ * Updates the state of the current key for the given source namespaces into the state of
+ * the target namespace.
+ *
+ * @param values The target namespace where the merged state should be stored.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void update(List<T> values) throws Exception;
--- End diff --
Let me document it. They are of the same result - return `null` when given an empty list
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159545896
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java ---
@@ -120,4 +121,16 @@ public void add(V value) {
a.addAll(b);
return a;
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ map.put(namespace, new ArrayList<>(values));
--- End diff --
because `values` is of `List` and `map` only is of `ArrayList`, need to enforce strict subclass
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on the issue:
https://github.com/apache/flink/pull/4963
@yunfan123 @aljoscha @StefanRRichter feedbacks are appreciated
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159631572
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java ---
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state.benchmark;
+
+import org.apache.flink.contrib.streaming.state.PredefinedOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.contrib.streaming.state.util.MergeUtils;
+import org.apache.flink.testutils.junit.RetryOnFailure;
+import org.apache.flink.testutils.junit.RetryRule;
+import org.apache.flink.util.TestLogger;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+import org.rocksdb.WriteOptions;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test that validates that the performance of APIs of RocksDB's ListState is as expected.
+ *
+ * Benchmarking
+ * Computer: MacbookPro (Mid 2015), Flash Storage, Processor 2.5GHz Intel Core i7, Memory 16GB 1600MHz DDR3
+ *
+ * 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
--- End diff --
Ok, that's an improvement that would justify the more complex approach. Nice!
---
[GitHub] flink issue #4963: [FLINK-7475] [core][DataStream API] support update() in L...
Posted by yunfan123 <gi...@git.apache.org>.
Github user yunfan123 commented on the issue:
https://github.com/apache/flink/pull/4963
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.
---
[GitHub] flink pull request #4963: [FLINK-7475] [core][DataStream API] support update...
Posted by bowenli86 <gi...@git.apache.org>.
Github user bowenli86 commented on a diff in the pull request:
https://github.com/apache/flink/pull/4963#discussion_r159546631
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java ---
@@ -158,4 +158,28 @@ public void mergeNamespaces(N target, Collection<N> sources) throws Exception {
throw new Exception("Error while merging state in RocksDB", e);
}
}
+
+ @Override
+ public void update(List<V> values) throws Exception {
+ clear();
+
+ if (values != null && !values.isEmpty()) {
--- End diff --
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
---