flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pnowojski <...@git.apache.org>
Subject [GitHub] flink pull request #4722: [FLINK-7683] Iterate over keys in KeyedStateBacken...
Date Tue, 26 Sep 2017 15:59:49 GMT
Github user pnowojski commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4722#discussion_r141090682
  
    --- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
---
    @@ -252,6 +257,43 @@ public RocksDBKeyedStateBackend(
     		LOG.debug("Setting initial keyed backend uid for operator {} to {}.", this.operatorIdentifier,
this.backendUID);
     	}
     
    +	@Override
    +	public <N> Stream<K> getKeys(String field, N namespace) {
    +		Tuple2<ColumnFamilyHandle, ?> columnInfo = kvStateInformation.get(field);
    +		if (columnInfo == null) {
    +			return Stream.empty();
    +		}
    +
    +		RocksIterator iterator = db.newIterator(columnInfo.f0);
    +		iterator.seekToFirst();
    +
    +		Iterator<K> sourceIterator = new Iterator<K>() {
    +			@Override
    +			public boolean hasNext() {
    +				return iterator.isValid();
    +			}
    +
    +			@Override
    +			public K next() {
    +				try {
    +					byte[] key = iterator.key();
    +
    +					DataInputViewStreamWrapper dataInput = new DataInputViewStreamWrapper(
    +						new ByteArrayInputStreamWithPos(key, keyGroupPrefixBytes, key.length - keyGroupPrefixBytes));
    +					K value = keySerializer.deserialize(dataInput);
    +					iterator.next();
    --- End diff --
    
    Added sanity check at the beginning of this method.


---

Mime
View raw message