flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [4/8] flink git commit: [FLINK-7700] Fix RocksDB ListState merging
Date Thu, 28 Sep 2017 15:09:09 GMT
[FLINK-7700] Fix RocksDB ListState merging

Before, the merged state was not cleared.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/93456ff7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/93456ff7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/93456ff7

Branch: refs/heads/master
Commit: 93456ff7c51a7184c385ce0326a8e190ecc9348b
Parents: fc9bc2f
Author: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Authored: Wed Sep 27 11:37:09 2017 +0200
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Thu Sep 28 17:02:55 2017 +0200

----------------------------------------------------------------------
 .../org/apache/flink/contrib/streaming/state/RocksDBListState.java  | 1 +
 1 file changed, 1 insertion(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/93456ff7/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 9d3e97e..f8ed244 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
@@ -146,6 +146,7 @@ public class RocksDBListState<K, N, V>
 
 					byte[] sourceKey = keySerializationStream.toByteArray();
 					byte[] valueBytes = backend.db.get(columnFamily, sourceKey);
+					backend.db.delete(columnFamily, sourceKey);
 
 					if (valueBytes != null) {
 						backend.db.merge(columnFamily, writeOptions, targetKey, valueBytes);


Mime
View raw message