flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From StefanRRichter <...@git.apache.org>
Subject [GitHub] flink pull request #5518: [FLINK-8679][State Backends]Fix RocksDBKeyedBacken...
Date Mon, 19 Feb 2018 13:45:56 GMT
Github user StefanRRichter commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5518#discussion_r169074775
  
    --- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
---
    @@ -1991,43 +1999,71 @@ public int numStateEntries() {
     		return count;
     	}
     
    +	/**
    +	 * This class is not thread safety.
    +	 */
     	private static class RocksIteratorWrapper<K> implements Iterator<K> {
     		private final RocksIterator iterator;
     		private final String state;
     		private final TypeSerializer<K> keySerializer;
     		private final int keyGroupPrefixBytes;
    +		private final byte[] namespaceBytes;
    +		private K nextKey;
     
     		public RocksIteratorWrapper(
     				RocksIterator iterator,
     				String state,
     				TypeSerializer<K> keySerializer,
    -				int keyGroupPrefixBytes) {
    +				int keyGroupPrefixBytes,
    +				byte[] namespaceBytes) {
     			this.iterator = Preconditions.checkNotNull(iterator);
     			this.state = Preconditions.checkNotNull(state);
     			this.keySerializer = Preconditions.checkNotNull(keySerializer);
     			this.keyGroupPrefixBytes = Preconditions.checkNotNull(keyGroupPrefixBytes);
    +			this.namespaceBytes = Preconditions.checkNotNull(namespaceBytes);
    +			this.nextKey = null;
     		}
     
     		@Override
     		public boolean hasNext() {
    -			return iterator.isValid();
    +			final int namespaceBytesLength = namespaceBytes.length;
    +			while (nextKey == null && iterator.isValid()) {
    +				try {
    +					boolean namespaceValid = true;
    +					byte[] key = iterator.key();
    +					if (key.length >= namespaceBytesLength + keyGroupPrefixBytes) {
    +						for (int i = 1; i <= namespaceBytesLength; ++i) {
    +							if (key[key.length - i] != namespaceBytes[namespaceBytesLength - i]) {
    +								namespaceValid = false;
    +								break;
    +							}
    +						}
    +						if (namespaceValid) {
    +							DataInputViewStreamWrapper dataInput = new DataInputViewStreamWrapper(
    --- End diff --
    
    This creates a lot of short-lived objects, and the `ByteArrayInputStreamWithPos` can also
grow internally. I wonder if we could not just reuse always the same `ByteArrayInputStreamWithPos`
at least? This could also let the internal array eventually be at good size. 


---

Mime
View raw message