You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/01/19 11:06:06 UTC
flink git commit: [FLINK-7938] Introduce addAll() to ListState
Repository: flink
Updated Branches:
refs/heads/master 1f9c2d974 -> 14840809b
[FLINK-7938] Introduce addAll() to ListState
This closes #5281.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/14840809
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/14840809
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/14840809
Branch: refs/heads/master
Commit: 14840809b13f4f03a487cc4483b5f76ec98b4f7b
Parents: 1f9c2d9
Author: Bowen Li <bo...@gmail.com>
Authored: Fri Jan 12 00:14:55 2018 -0800
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Fri Jan 19 12:05:16 2018 +0100
----------------------------------------------------------------------
docs/dev/stream/state/state.md | 2 +-
.../kafka/FlinkKafkaConsumerBaseTest.java | 7 +++-
.../kinesis/FlinkKinesisConsumerTest.java | 7 +++-
.../streaming/state/RocksDBListState.java | 42 +++++++++++++++-----
.../RocksDBListStatePerformanceTest.java | 6 ++-
.../flink/api/common/state/AppendingState.java | 3 +-
.../flink/api/common/state/ListState.java | 20 ++++++++--
.../client/state/ImmutableListState.java | 5 +++
.../state/DefaultOperatorStateBackend.java | 5 +++
.../runtime/state/UserFacingListState.java | 5 +++
.../flink/runtime/state/heap/HeapListState.java | 18 +++++++++
.../state/internal/InternalListState.java | 22 ++++++++--
.../runtime/state/StateBackendTestBase.java | 17 +++++++-
13 files changed, 136 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/docs/dev/stream/state/state.md
----------------------------------------------------------------------
diff --git a/docs/dev/stream/state/state.md b/docs/dev/stream/state/state.md
index 2002138..7d37691 100644
--- a/docs/dev/stream/state/state.md
+++ b/docs/dev/stream/state/state.md
@@ -94,7 +94,7 @@ for each key that the operation sees). The value can be set using `update(T)` an
`T value()`.
* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
-over all currently stored elements. Elements are added using `add(T)`, the Iterable can
+over all currently stored elements. Elements are added using `add(T)` or `addAll(List<T>)`, the Iterable can
be retrieved using `Iterable<T> get()`. You can also override the existing list with `update(List<T>)`
* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
index 7185ee8..933891c 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
@@ -705,7 +705,12 @@ public class FlinkKafkaConsumerBaseTest {
public void update(List<T> values) throws Exception {
clear();
- if (values != null && !values.isEmpty()) {
+ addAll(values);
+ }
+
+ @Override
+ public void addAll(List<T> values) throws Exception {
+ if (values != null) {
list.addAll(values);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
index 3b0cb5c..379b640 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
@@ -548,7 +548,12 @@ public class FlinkKinesisConsumerTest {
public void update(List<T> values) throws Exception {
list.clear();
- if (values != null || !values.isEmpty()) {
+ addAll(values);
+ }
+
+ @Override
+ public void addAll(List<T> values) throws Exception {
+ if (values != null) {
list.addAll(values);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
index 436fec2..950058b 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
@@ -167,24 +167,48 @@ public class RocksDBListState<K, N, V>
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());
+ byte[] premerge = getPreMergedValue(values);
+ if (premerge != null) {
+ backend.db.put(columnFamily, writeOptions, key, premerge);
+ } else {
+ throw new IOException("Failed pre-merge values in update()");
}
+ } catch (IOException | RocksDBException e) {
+ throw new RuntimeException("Error while updating data to RocksDB", e);
+ }
+ }
+ }
+
+ @Override
+ public void addAll(List<V> values) throws Exception {
+ if (values != null && !values.isEmpty()) {
+ try {
+ writeCurrentKeyWithGroupAndNamespace();
+ byte[] key = keySerializationStream.toByteArray();
- byte[] premerge = MergeUtils.merge(bytes);
+ byte[] premerge = getPreMergedValue(values);
if (premerge != null) {
- backend.db.put(columnFamily, writeOptions, key, premerge);
+ backend.db.merge(columnFamily, writeOptions, key, premerge);
} else {
- throw new IOException("Failed pre-merge values");
+ throw new IOException("Failed pre-merge values in addAll()");
}
} catch (IOException | RocksDBException e) {
throw new RuntimeException("Error while updating data to RocksDB", e);
}
}
}
+
+ private byte[] getPreMergedValue(List<V> values) throws IOException {
+ 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());
+ }
+
+ return MergeUtils.merge(bytes);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java
index 88c74fd..153584f 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/RocksDBListStatePerformanceTest.java
@@ -45,6 +45,9 @@ import java.util.List;
*
* <p>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()
+ * 10 236875 ns 17048 ns 13.90x
+ * 50 312332 ns 14281 ns 21.87x
+ * 100 393791 ns 18360 ns 21.45x
* 500 978703 ns 55397 ns 17.66x
* 1000 3044179 ns 89474 ns 34.02x
* 5000 9247395 ns 305580 ns 30.26x
@@ -83,7 +86,7 @@ public class RocksDBListStatePerformanceTest extends TestLogger {
final byte[] valueBytes = value.getBytes(StandardCharsets.UTF_8);
// The number of values added to ListState. Can be changed for benchmarking
- final int num = 1000;
+ final int num = 10;
try (
final Options options = new Options()
@@ -103,7 +106,6 @@ public class RocksDBListStatePerformanceTest extends TestLogger {
// ----- add() API -----
log.info("begin add");
- System.out.println("begin add");
final long beginInsert1 = System.nanoTime();
for (int i = 0; i < num; i++) {
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-core/src/main/java/org/apache/flink/api/common/state/AppendingState.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/AppendingState.java b/flink-core/src/main/java/org/apache/flink/api/common/state/AppendingState.java
index dd070a9..bdf238c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/state/AppendingState.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/state/AppendingState.java
@@ -64,8 +64,7 @@ public interface AppendingState<IN, OUT> extends State {
* to the list of values. The next time {@link #get()} is called (for the same state
* partition) the returned state will represent the updated list.
*
- * @param value
- * The new value for the state.
+ * @param value The new value for the state.
*
* @throws IOException Thrown if the system cannot access the state.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java b/flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java
index 97aaa7f..74f275b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/state/ListState.java
@@ -37,14 +37,28 @@ import java.util.List;
@PublicEvolving
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.
+ * Updates the operator state accessible by {@link #get()} by updating existing values to
+ * to the given list of values. The next time {@link #get()} is called (for the same state
+ * partition) the returned state will represent the updated list.
*
* If `null` or an empty list is passed in, the state value will be null
*
- * @param values The target namespace where the merged state should be stored.
+ * @param values The new values for the state.
*
* @throws Exception The method may forward exception thrown internally (by I/O or functions).
*/
void update(List<T> values) throws Exception;
+
+ /**
+ * Updates the operator state accessible by {@link #get()} by adding the given values
+ * to existing list of values. The next time {@link #get()} is called (for the same state
+ * partition) the returned state will represent the updated list.
+ *
+ * If `null` or an empty list is passed in, the state value remains unchanged
+ *
+ * @param values The new values to be added to the state.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void addAll(List<T> values) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java
index dd780a8..9f1465e 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/state/ImmutableListState.java
@@ -72,4 +72,9 @@ public final class ImmutableListState<V> extends ImmutableState implements ListS
public void update(List<V> values) throws Exception {
throw MODIFICATION_ATTEMPT_ERROR;
}
+
+ @Override
+ public void addAll(List<V> values) throws Exception {
+ throw MODIFICATION_ATTEMPT_ERROR;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index da05a8b..1f3e017 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -502,6 +502,11 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
public void update(List<S> values) throws Exception {
internalList.clear();
+ addAll(values);
+ }
+
+ @Override
+ public void addAll(List<S> values) throws Exception {
if (values != null && !values.isEmpty()) {
internalList.addAll(values);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-runtime/src/main/java/org/apache/flink/runtime/state/UserFacingListState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/UserFacingListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/UserFacingListState.java
index b93e4d3..71f5aa5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/UserFacingListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/UserFacingListState.java
@@ -60,4 +60,9 @@ class UserFacingListState<T> implements ListState<T> {
public void update(List<T> values) throws Exception {
originalState.update(values);
}
+
+ @Override
+ public void addAll(List<T> values) throws Exception {
+ originalState.addAll(values);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java
index fbf2087..4ce3c4d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapListState.java
@@ -133,4 +133,22 @@ public class HeapListState<K, N, V>
map.put(namespace, new ArrayList<>(values));
}
}
+
+ @Override
+ public void addAll(List<V> values) throws Exception {
+ if (values != null && !values.isEmpty()) {
+ final N namespace = currentNamespace;
+ final StateTable<K, N, ArrayList<V>> map = stateTable;
+
+ ArrayList<V> list = map.get(currentNamespace);
+
+ if (list == null) {
+ list = new ArrayList<>();
+ }
+
+ list.addAll(values);
+
+ map.put(namespace, list);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-runtime/src/main/java/org/apache/flink/runtime/state/internal/InternalListState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/internal/InternalListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/internal/InternalListState.java
index ac17f8e..1e22dc6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/internal/InternalListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/internal/InternalListState.java
@@ -32,12 +32,28 @@ import java.util.List;
*/
public interface InternalListState<N, T> extends InternalMergingState<N, T, Iterable<T>>, ListState<T> {
/**
- * Updates the state of the current key for the given source namespaces into the state of
- * the target namespace.
+ * Updates the operator state accessible by {@link #get()} by updating existing values to
+ * to the given list of values. The next time {@link #get()} is called (for the same state
+ * partition) the returned state will represent the updated list.
*
- * @param values The target namespace where the merged state should be stored.
+ * If `null` or an empty list is passed in, the state value will be null
+ *
+ * @param values The new values for the state.
*
* @throws Exception The method may forward exception thrown internally (by I/O or functions).
*/
void update(List<T> values) throws Exception;
+
+ /**
+ * Updates the operator state accessible by {@link #get()} by adding the given values
+ * to existing list of values. The next time {@link #get()} is called (for the same state
+ * partition) the returned state will represent the updated list.
+ *
+ * If `null` or an empty list is passed in, the state value remains unchanged
+ *
+ * @param values The new values to be added to the state.
+ *
+ * @throws Exception The method may forward exception thrown internally (by I/O or functions).
+ */
+ void addAll(List<T> values) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/14840809/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
index 2b4de54..f39cf3f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
@@ -1286,7 +1286,7 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
}
@Test
- public void testListStateAddUpdateAndGet() throws Exception {
+ public void testListStateAPIs() throws Exception {
AbstractKeyedStateBackend<String> keyedBackend = createKeyedBackend(StringSerializer.INSTANCE);
@@ -1318,7 +1318,22 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
keyedBackend.setCurrentKey("g");
assertNull(state.get());
+ state.addAll(null);
+ assertNull(state.get());
+ state.addAll(new ArrayList<>());
+ assertNull(state.get());
+ state.addAll(Arrays.asList(3L, 4L));
+ assertThat(state.get(), containsInAnyOrder(3L, 4L));
+ state.addAll(null);
+ assertThat(state.get(), containsInAnyOrder(3L, 4L));
+ state.addAll(new ArrayList<>());
+ assertThat(state.get(), containsInAnyOrder(3L, 4L));
+ state.addAll(new ArrayList<>());
+ assertThat(state.get(), containsInAnyOrder(3L, 4L));
+ state.addAll(Arrays.asList(5L, 6L));
+ assertThat(state.get(), containsInAnyOrder(3L, 4L, 5L, 6L));
state.update(Arrays.asList(1L, 2L));
+ assertThat(state.get(), containsInAnyOrder(1L, 2L));
keyedBackend.setCurrentKey("def");
assertThat(state.get(), containsInAnyOrder(10L, 16L));