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 2019/03/06 15:45:41 UTC
[flink] branch master updated: [hotfix][statebackend] Reduce and
simplify code for column creation in RocksDB backend
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 953a5ff [hotfix][statebackend] Reduce and simplify code for column creation in RocksDB backend
953a5ff is described below
commit 953a5ffcbdae4115f7d525f310723cf8770779df
Author: azagrebin <az...@users.noreply.github.com>
AuthorDate: Wed Mar 6 16:45:33 2019 +0100
[hotfix][statebackend] Reduce and simplify code for column creation in RocksDB backend
This closes #7830.
---
.../streaming/state/RocksDBKeyedStateBackend.java | 16 +----
.../state/RocksDBKeyedStateBackendBuilder.java | 13 ++--
.../streaming/state/RocksDBOperationUtils.java | 82 ++++++++++++----------
.../state/RocksDBPriorityQueueSetFactory.java | 5 +-
.../restore/AbstractRocksDBRestoreOperation.java | 19 ++---
.../state/restore/RocksDBFullRestoreOperation.java | 9 +--
.../RocksDBIncrementalRestoreOperation.java | 36 +++-------
.../state/restore/RocksDBNoneRestoreOperation.java | 7 +-
.../state/ttl/RocksDbTtlCompactFiltersManager.java | 31 ++------
9 files changed, 78 insertions(+), 140 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index c6d3863..4a6bd3c 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -126,9 +126,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
RocksDBKeyedStateBackend<K> backend) throws Exception;
}
- /** String that identifies the operator that owns this backend. */
- private final String operatorIdentifier;
-
/** Factory function to create column family options from state name. */
private final Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory;
@@ -158,9 +155,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
/** Number of bytes required to prefix the key groups. */
private final int keyGroupPrefixBytes;
- /** Thread number used to transfer state files while restoring/snapshotting. */
- private final int numberOfTransferingThreads;
-
/**
* We are not using the default column family for Flink state ops, but we still need to remember this handle so that
* we can close it properly when the backend is closed. Note that the one returned by {@link RocksDB#open(String)}
@@ -201,7 +195,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
private final RocksDbTtlCompactFiltersManager ttlCompactFiltersManager;
public RocksDBKeyedStateBackend(
- String operatorIdentifier,
ClassLoader userCodeClassLoader,
File instanceBasePath,
DBOptions dbOptions,
@@ -211,7 +204,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
ExecutionConfig executionConfig,
- int numberOfTransferingThreads,
TtlTimeProvider ttlTimeProvider,
RocksDB db,
LinkedHashMap<String, RocksDbKvStateInfo> kvStateInformation,
@@ -233,10 +225,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
this.ttlCompactFiltersManager = ttlCompactFiltersManager;
- this.operatorIdentifier = Preconditions.checkNotNull(operatorIdentifier);
-
- this.numberOfTransferingThreads = numberOfTransferingThreads;
-
// ensure that we use the right merge operator, because other code relies on this
this.columnFamilyOptionsFactory = Preconditions.checkNotNull(columnFamilyOptionsFactory);
@@ -500,8 +488,8 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
stateSerializer,
StateSnapshotTransformFactory.noTransform());
- newRocksStateInfo = RocksDBOperationUtils.createStateInfo(newMetaInfo, ttlCompactFiltersManager,
- ttlTimeProvider, db, columnFamilyOptionsFactory);
+ newRocksStateInfo = RocksDBOperationUtils.createStateInfo(
+ newMetaInfo, db, columnFamilyOptionsFactory, ttlCompactFiltersManager);
RocksDBOperationUtils.registerKvStateInformation(this.kvStateInformation, this.nativeMetricMonitor,
stateDesc.getName(), newRocksStateInfo);
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index 3f245d0..aa6845e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -242,7 +242,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
RocksDB db = null;
AbstractRocksDBRestoreOperation restoreOperation = null;
RocksDbTtlCompactFiltersManager ttlCompactFiltersManager =
- new RocksDbTtlCompactFiltersManager(enableTtlCompactionFilter);
+ new RocksDbTtlCompactFiltersManager(enableTtlCompactionFilter, ttlTimeProvider);
ResourceGuard rocksDBResourceGuard = new ResourceGuard();
SnapshotStrategy<K> snapshotStrategy;
@@ -324,7 +324,6 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
}
}
return new RocksDBKeyedStateBackend<>(
- this.operatorIdentifier,
this.userCodeClassLoader,
this.instanceBasePath,
this.dbOptions,
@@ -334,7 +333,6 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
this.numberOfKeyGroups,
this.keyGroupRange,
this.executionConfig,
- this.numberOfTransferingThreads,
this.ttlTimeProvider,
db,
kvStateInformation,
@@ -374,8 +372,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
KeyedStateHandle firstStateHandle = restoreStateHandles.iterator().next();
if (firstStateHandle instanceof IncrementalKeyedStateHandle) {
@@ -395,8 +392,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
} else {
return new RocksDBFullRestoreOperation<>(
keyGroupRange,
@@ -413,8 +409,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
index b183426..1e3fb8f 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java
@@ -20,7 +20,6 @@ package org.apache.flink.contrib.streaming.state;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.contrib.streaming.state.ttl.RocksDbTtlCompactFiltersManager;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
@@ -32,6 +31,8 @@ import org.rocksdb.DBOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -79,38 +80,11 @@ public class RocksDBOperationUtils {
return dbRef;
}
- public static ColumnFamilyDescriptor createColumnFamilyDescriptor(String stateName, ColumnFamilyOptions columnOptions) {
- byte[] nameBytes = stateName.getBytes(ConfigConstants.DEFAULT_CHARSET);
- Preconditions.checkState(!Arrays.equals(RocksDB.DEFAULT_COLUMN_FAMILY, nameBytes),
- "The chosen state name 'default' collides with the name of the default column family!");
-
- return new ColumnFamilyDescriptor(nameBytes, columnOptions);
- }
-
- public static ColumnFamilyHandle createColumnFamily(ColumnFamilyDescriptor columnDescriptor, RocksDB db) {
- try {
- return db.createColumnFamily(columnDescriptor);
- } catch (RocksDBException e) {
- IOUtils.closeQuietly(columnDescriptor.getOptions());
- throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", e);
- }
- }
-
- public static ColumnFamilyHandle createColumnFamily(
- String stateName,
- Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
- RocksDB db) {
- ColumnFamilyOptions options = createColumnFamilyOptions(columnFamilyOptionsFactory, stateName);
- return createColumnFamily(createColumnFamilyDescriptor(stateName, options), db);
- }
-
public static RocksIteratorWrapper getRocksIterator(RocksDB db) {
return new RocksIteratorWrapper(db.newIterator());
}
- public static RocksIteratorWrapper getRocksIterator(
- RocksDB db,
- ColumnFamilyHandle columnFamilyHandle) {
+ public static RocksIteratorWrapper getRocksIterator(RocksDB db, ColumnFamilyHandle columnFamilyHandle) {
return new RocksIteratorWrapper(db.newIterator(columnFamilyHandle));
}
@@ -119,8 +93,8 @@ public class RocksDBOperationUtils {
RocksDBNativeMetricMonitor nativeMetricMonitor,
String columnFamilyName,
RocksDBKeyedStateBackend.RocksDbKvStateInfo registeredColumn) {
- kvStateInformation.put(columnFamilyName, registeredColumn);
+ kvStateInformation.put(columnFamilyName, registeredColumn);
if (nativeMetricMonitor != null) {
nativeMetricMonitor.registerColumnFamily(columnFamilyName, registeredColumn.columnFamilyHandle);
}
@@ -128,26 +102,58 @@ public class RocksDBOperationUtils {
/**
* Creates a state info from a new meta info to use with a k/v state.
+ *
+ * <p>Creates the column family for the state.
+ * Sets TTL compaction filter if {@code ttlCompactFiltersManager} is not {@code null}.
*/
public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
RegisteredStateMetaInfoBase metaInfoBase,
- RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider,
RocksDB db,
- Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory) {
- ColumnFamilyOptions options = createColumnFamilyOptions(columnFamilyOptionsFactory, metaInfoBase.getName());
- ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(ttlTimeProvider, metaInfoBase, options);
- ColumnFamilyDescriptor columnFamilyDescriptor = createColumnFamilyDescriptor(metaInfoBase.getName(), options);
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
+ @Nullable RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
+
+ ColumnFamilyDescriptor columnFamilyDescriptor = createColumnFamilyDescriptor(
+ metaInfoBase, columnFamilyOptionsFactory, ttlCompactFiltersManager);
return new RocksDBKeyedStateBackend.RocksDbKvStateInfo(createColumnFamily(columnFamilyDescriptor, db), metaInfoBase);
}
- public static ColumnFamilyOptions createColumnFamilyOptions(
+ /**
+ * Creates a column descriptor for sate column family.
+ *
+ * <p>Sets TTL compaction filter if {@code ttlCompactFiltersManager} is not {@code null}.
+ */
+ public static ColumnFamilyDescriptor createColumnFamilyDescriptor(
+ RegisteredStateMetaInfoBase metaInfoBase,
Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
- String stateName) {
+ @Nullable RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
+
+ ColumnFamilyOptions options = createColumnFamilyOptions(columnFamilyOptionsFactory, metaInfoBase.getName());
+ if (ttlCompactFiltersManager != null) {
+ ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(metaInfoBase, options);
+ }
+ byte[] nameBytes = metaInfoBase.getName().getBytes(ConfigConstants.DEFAULT_CHARSET);
+ Preconditions.checkState(!Arrays.equals(RocksDB.DEFAULT_COLUMN_FAMILY, nameBytes),
+ "The chosen state name 'default' collides with the name of the default column family!");
+
+ return new ColumnFamilyDescriptor(nameBytes, options);
+ }
+
+ public static ColumnFamilyOptions createColumnFamilyOptions(
+ Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory, String stateName) {
+
// ensure that we use the right merge operator, because other code relies on this
return columnFamilyOptionsFactory.apply(stateName).setMergeOperatorName(MERGE_OPERATOR_NAME);
}
+ private static ColumnFamilyHandle createColumnFamily(ColumnFamilyDescriptor columnDescriptor, RocksDB db) {
+ try {
+ return db.createColumnFamily(columnDescriptor);
+ } catch (RocksDBException e) {
+ IOUtils.closeQuietly(columnDescriptor.getOptions());
+ throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", e);
+ }
+ }
+
public static void addColumnFamilyOptionsToCloseLater(
List<ColumnFamilyOptions> columnFamilyOptions, ColumnFamilyHandle columnFamilyHandle) {
try {
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
index 1eb0f47..1cbeebe 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
@@ -145,12 +145,9 @@ public class RocksDBPriorityQueueSetFactory implements PriorityQueueSetFactory {
if (stateInfo == null) {
// Currently this class is for timer service and TTL feature is not applicable here,
// so no need to register compact filter when creating column family
- final ColumnFamilyHandle columnFamilyHandle =
- RocksDBOperationUtils.createColumnFamily(stateName, columnFamilyOptionsFactory, this.db);
RegisteredPriorityQueueStateBackendMetaInfo<T> metaInfo =
new RegisteredPriorityQueueStateBackendMetaInfo<>(stateName, byteOrderedElementSerializer);
-
- stateInfo = new RocksDBKeyedStateBackend.RocksDbKvStateInfo(columnFamilyHandle, metaInfo);
+ stateInfo = RocksDBOperationUtils.createStateInfo(metaInfo, db, columnFamilyOptionsFactory, null);
RocksDBOperationUtils.registerKvStateInformation(kvStateInformation, nativeMetricMonitor, stateName, stateInfo);
} else {
// TODO we implement the simple way of supporting the current functionality, mimicking keyed state
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
index 9a29b62..064483c 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/AbstractRocksDBRestoreOperation.java
@@ -33,7 +33,6 @@ import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.StateMigrationException;
@@ -42,7 +41,6 @@ import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
import javax.annotation.Nonnull;
@@ -87,7 +85,6 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
// - Full restore
// - data ingestion after db open: #getOrRegisterStateColumnFamilyHandle before creating column family
protected final RocksDbTtlCompactFiltersManager ttlCompactFiltersManager;
- protected final TtlTimeProvider ttlTimeProvider;
protected RocksDB db;
protected ColumnFamilyHandle defaultColumnFamilyHandle;
@@ -109,8 +106,7 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
this.keyGroupRange = keyGroupRange;
this.keyGroupPrefixBytes = keyGroupPrefixBytes;
this.numberOfTransferringThreads = numberOfTransferringThreads;
@@ -127,12 +123,11 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
this.metricGroup = metricGroup;
this.restoreStateHandles = stateHandles;
this.ttlCompactFiltersManager = ttlCompactFiltersManager;
- this.ttlTimeProvider = ttlTimeProvider;
this.columnFamilyHandles = new ArrayList<>(1);
this.columnFamilyDescriptors = Collections.emptyList();
}
- public void openDB() throws IOException {
+ void openDB() throws IOException {
db = RocksDBOperationUtils.openDB(
dbPath,
columnFamilyDescriptors,
@@ -150,9 +145,9 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
return this.db;
}
- protected RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
+ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
ColumnFamilyHandle columnFamilyHandle,
- StateMetaInfoSnapshot stateMetaInfoSnapshot) throws RocksDBException {
+ StateMetaInfoSnapshot stateMetaInfoSnapshot) {
RocksDbKvStateInfo registeredStateMetaInfoEntry =
kvStateInformation.get(stateMetaInfoSnapshot.getName());
@@ -163,8 +158,8 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
RegisteredStateMetaInfoBase stateMetaInfo =
RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
if (columnFamilyHandle == null) {
- registeredStateMetaInfoEntry = RocksDBOperationUtils.createStateInfo(stateMetaInfo,
- ttlCompactFiltersManager, ttlTimeProvider, db, columnFamilyOptionsFactory);
+ registeredStateMetaInfoEntry = RocksDBOperationUtils.createStateInfo(
+ stateMetaInfo, db, columnFamilyOptionsFactory, ttlCompactFiltersManager);
} else {
registeredStateMetaInfoEntry = new RocksDbKvStateInfo(columnFamilyHandle, stateMetaInfo);
}
@@ -181,7 +176,7 @@ public abstract class AbstractRocksDBRestoreOperation<K> implements RocksDBResto
return registeredStateMetaInfoEntry;
}
- protected KeyedBackendSerializationProxy<K> readMetaData(DataInputView dataInputView)
+ KeyedBackendSerializationProxy<K> readMetaData(DataInputView dataInputView)
throws IOException, StateMigrationException {
// isSerializerPresenceRequired flag is set to false, since for the RocksDB state backend,
// deserialization of state happens lazily during runtime; we depend on the fact
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
index e9c8fe9..289ebdc 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBFullRestoreOperation.java
@@ -38,7 +38,6 @@ import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.StreamCompressionDecorator;
import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StateMigrationException;
@@ -103,8 +102,7 @@ public class RocksDBFullRestoreOperation<K> extends AbstractRocksDBRestoreOperat
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
super(
keyGroupRange,
keyGroupPrefixBytes,
@@ -120,8 +118,7 @@ public class RocksDBFullRestoreOperation<K> extends AbstractRocksDBRestoreOperat
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
/**
@@ -169,7 +166,7 @@ public class RocksDBFullRestoreOperation<K> extends AbstractRocksDBRestoreOperat
/**
* Restore the KV-state / ColumnFamily meta data for all key-groups referenced by the current state handle.
*/
- private void restoreKVStateMetaData() throws IOException, StateMigrationException, RocksDBException {
+ private void restoreKVStateMetaData() throws IOException, StateMigrationException {
KeyedBackendSerializationProxy<K> serializationProxy = readMetaData(currentStateHandleInView);
this.keygroupStreamCompressionDecorator = serializationProxy.isUsingKeyGroupCompression() ?
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
index 3371f4f..9088006 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java
@@ -18,7 +18,6 @@
package org.apache.flink.contrib.streaming.state.restore;
-import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.contrib.streaming.state.RocksDBIncrementalCheckpointUtils;
import org.apache.flink.contrib.streaming.state.RocksDBKeySerializationUtils;
import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.RocksDbKvStateInfo;
@@ -44,11 +43,11 @@ import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedBackendSerializationProxy;
import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
import org.apache.flink.runtime.state.StateHandleID;
import org.apache.flink.runtime.state.StateSerializerProvider;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.IOUtils;
import org.rocksdb.ColumnFamilyDescriptor;
@@ -106,8 +105,7 @@ public class RocksDBIncrementalRestoreOperation<K> extends AbstractRocksDBRestor
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider) {
+ @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
super(keyGroupRange,
keyGroupPrefixBytes,
numberOfTransferringThreads,
@@ -122,8 +120,7 @@ public class RocksDBIncrementalRestoreOperation<K> extends AbstractRocksDBRestor
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
this.operatorIdentifier = operatorIdentifier;
this.restoredSstFiles = new TreeMap<>();
this.lastCompletedCheckpointId = -1L;
@@ -252,19 +249,10 @@ public class RocksDBIncrementalRestoreOperation<K> extends AbstractRocksDBRestor
}
}
- private void registerColumnFamilyHandles(List<StateMetaInfoSnapshot> metaInfoSnapshots)
- throws BackendBuildingException {
+ private void registerColumnFamilyHandles(List<StateMetaInfoSnapshot> metaInfoSnapshots) {
// Register CF handlers
for (int i = 0; i < metaInfoSnapshots.size(); ++i) {
- try {
- getOrRegisterStateColumnFamilyHandle(
- columnFamilyHandles.get(i),
- metaInfoSnapshots.get(i));
- } catch (RocksDBException e) {
- String errMsg = "Failed to register CF handle.";
- LOG.error(errMsg, e);
- throw new BackendBuildingException(errMsg, e);
- }
+ getOrRegisterStateColumnFamilyHandle(columnFamilyHandles.get(i), metaInfoSnapshots.get(i));
}
}
@@ -452,16 +440,10 @@ public class RocksDBIncrementalRestoreOperation<K> extends AbstractRocksDBRestor
new ArrayList<>(stateMetaInfoSnapshots.size());
for (StateMetaInfoSnapshot stateMetaInfoSnapshot : stateMetaInfoSnapshots) {
- ColumnFamilyOptions options = RocksDBOperationUtils.createColumnFamilyOptions(
- columnFamilyOptionsFactory, stateMetaInfoSnapshot.getName());
- if (registerTtlCompactFilter) {
- ttlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(ttlTimeProvider,
- stateMetaInfoSnapshot, options);
- }
- ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(
- stateMetaInfoSnapshot.getName().getBytes(ConfigConstants.DEFAULT_CHARSET),
- options);
-
+ RegisteredStateMetaInfoBase metaInfoBase =
+ RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
+ ColumnFamilyDescriptor columnFamilyDescriptor = RocksDBOperationUtils.createColumnFamilyDescriptor(
+ metaInfoBase, columnFamilyOptionsFactory, registerTtlCompactFilter ? ttlCompactFiltersManager : null);
columnFamilyDescriptors.add(columnFamilyDescriptor);
}
return columnFamilyDescriptors;
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
index 873132a..bfc25d6 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBNoneRestoreOperation.java
@@ -26,7 +26,6 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.StateSerializerProvider;
-import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
@@ -57,8 +56,7 @@ public class RocksDBNoneRestoreOperation<K> extends AbstractRocksDBRestoreOperat
RocksDBNativeMetricOptions nativeMetricOptions,
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> restoreStateHandles,
- @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- TtlTimeProvider ttlTimeProvider
+ @Nonnull RocksDbTtlCompactFiltersManager ttlCompactFiltersManager
) {
super(keyGroupRange,
keyGroupPrefixBytes,
@@ -74,8 +72,7 @@ public class RocksDBNoneRestoreOperation<K> extends AbstractRocksDBRestoreOperat
nativeMetricOptions,
metricGroup,
restoreStateHandles,
- ttlCompactFiltersManager,
- ttlTimeProvider);
+ ttlCompactFiltersManager);
}
@Override
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
index ccdb092..affea5b 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
@@ -23,12 +23,10 @@ import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo;
import org.apache.flink.runtime.state.RegisteredStateMetaInfoBase;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.runtime.state.ttl.TtlStateFactory;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.runtime.state.ttl.TtlUtils;
@@ -57,47 +55,30 @@ public class RocksDbTtlCompactFiltersManager {
/** Enables RocksDb compaction filter for State with TTL. */
private final boolean enableTtlCompactionFilter;
+ private final TtlTimeProvider ttlTimeProvider;
+
/** Registered compaction filter factories. */
private final LinkedHashMap<String, FlinkCompactionFilterFactory> compactionFilterFactories;
- public RocksDbTtlCompactFiltersManager(boolean enableTtlCompactionFilter) {
+ public RocksDbTtlCompactFiltersManager(boolean enableTtlCompactionFilter, TtlTimeProvider ttlTimeProvider) {
this.enableTtlCompactionFilter = enableTtlCompactionFilter;
+ this.ttlTimeProvider = ttlTimeProvider;
this.compactionFilterFactories = new LinkedHashMap<>();
}
public void setAndRegisterCompactFilterIfStateTtl(
- TtlTimeProvider ttlTimeProvider,
- @Nonnull StateMetaInfoSnapshot stateMetaInfoSnapshot,
- @Nonnull ColumnFamilyOptions options) {
-
- boolean keyValueState = stateMetaInfoSnapshot.getBackendStateType() == StateMetaInfoSnapshot.BackendStateType.KEY_VALUE;
- if (enableTtlCompactionFilter && keyValueState) {
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<?> stateSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<?>) stateMetaInfoSnapshot.getTypeSerializerSnapshot(
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
- TypeSerializer<?> serializer = stateSerializerSnapshot.restoreSerializer();
- if (TtlStateFactory.TtlSerializer.isTtlStateSerializer(serializer)) {
- createAndSetCompactFilterFactory(stateMetaInfoSnapshot.getName(), ttlTimeProvider, options);
- }
- }
- }
-
- public void setAndRegisterCompactFilterIfStateTtl(
- TtlTimeProvider ttlTimeProvider,
@Nonnull RegisteredStateMetaInfoBase metaInfoBase,
@Nonnull ColumnFamilyOptions options) {
if (enableTtlCompactionFilter && metaInfoBase instanceof RegisteredKeyValueStateBackendMetaInfo) {
RegisteredKeyValueStateBackendMetaInfo kvMetaInfoBase = (RegisteredKeyValueStateBackendMetaInfo) metaInfoBase;
if (TtlStateFactory.TtlSerializer.isTtlStateSerializer(kvMetaInfoBase.getStateSerializer())) {
- createAndSetCompactFilterFactory(metaInfoBase.getName(), ttlTimeProvider, options);
+ createAndSetCompactFilterFactory(metaInfoBase.getName(), options);
}
}
}
- private void createAndSetCompactFilterFactory(
- String stateName, TtlTimeProvider ttlTimeProvider, @Nonnull ColumnFamilyOptions options) {
+ private void createAndSetCompactFilterFactory(String stateName, @Nonnull ColumnFamilyOptions options) {
FlinkCompactionFilterFactory compactionFilterFactory =
new FlinkCompactionFilterFactory(new TimeProviderWrapper(ttlTimeProvider), createRocksDbNativeLogger());