[
https://issues.apache.org/jira/browse/FLINK-6364?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15993352#comment-15993352
]
ASF GitHub Bot commented on FLINK-6364:
---------------------------------------
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/3801#discussion_r114288802
--- Diff:
flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateHandle.java
---
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.state.CompositeStateHandle;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateHandle;
+import org.apache.flink.runtime.state.SharedStateHandle;
+import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.StateUtil;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The handle to states in incremental snapshots taken by {@link
RocksDBKeyedStateBackend}
+ */
+public class RocksDBKeyedStateHandle implements KeyedStateHandle,
CompositeStateHandle {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RocksDBKeyedStateHandle.class);
+
+ private static final long serialVersionUID = -8328808513197388231L;
+
+ private final JobID jobId;
+
+ private final String operatorIdentifier;
+
+ private final KeyGroupRange keyGroupRange;
+
+ private final Set<String> newSstFileNames;
+
+ private final Map<String, StreamStateHandle> sstFiles;
+
+ private final Map<String, StreamStateHandle> miscFiles;
+
+ private final StreamStateHandle metaStateHandle;
+
+ private boolean registered;
+
+ RocksDBKeyedStateHandle(
+ JobID jobId,
+ String operatorIdentifier,
+ KeyGroupRange keyGroupRange,
+ Set<String> newSstFileNames,
+ Map<String, StreamStateHandle> sstFiles,
+ Map<String, StreamStateHandle> miscFiles,
+ StreamStateHandle metaStateHandle) {
+
+ this.jobId = jobId;
+ this.operatorIdentifier = operatorIdentifier;
+ this.keyGroupRange = keyGroupRange;
+ this.newSstFileNames = newSstFileNames;
+ this.sstFiles = sstFiles;
+ this.miscFiles = miscFiles;
+ this.metaStateHandle = metaStateHandle;
+ this.registered = false;
+ }
+
+ @Override
+ public KeyGroupRange getKeyGroupRange() {
+ return keyGroupRange;
+ }
+
+ public Map<String, StreamStateHandle> getSstFiles() {
+ return sstFiles;
+ }
+
+ public Map<String, StreamStateHandle> getMiscFiles() {
+ return miscFiles;
+ }
+
+ public StreamStateHandle getMetaStateHandle() {
+ return metaStateHandle;
+ }
+
+ @Override
+ public KeyedStateHandle getIntersection(KeyGroupRange keyGroupRange) {
+ if (this.keyGroupRange.getIntersection(keyGroupRange) !=
KeyGroupRange.EMPTY_KEY_GROUP_RANGE) {
+ return this;
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public void discardState() throws Exception {
+
+ try {
+ metaStateHandle.discardState();
+ } catch (Exception e) {
+ LOG.warn("Could not properly discard meta data.", e);
+ }
+
+ try {
+
StateUtil.bestEffortDiscardAllStateObjects(miscFiles.values());
+ } catch (Exception e) {
+ LOG.warn("Could not properly discard misc file state.",
e);
+ }
+
+ if (!registered) {
+ for (String newSstFileName : newSstFileNames) {
+ StreamStateHandle handle =
sstFiles.get(newSstFileName);
+ try {
+ handle.discardState();
+ } catch (Exception e) {
+ LOG.warn("Could not properly discard
sst file state", e);
+ }
+ }
+ }
+ }
+
+ @Override
+ public long getStateSize() {
+ long size = StateUtil.getStateSize(metaStateHandle);
+
+ for (StreamStateHandle sstFileHandle : sstFiles.values()) {
+ size += sstFileHandle.getStateSize();
+ }
+
+ for (StreamStateHandle miscFileHandle : miscFiles.values()) {
+ size += miscFileHandle.getStateSize();
+ }
+
+ return size;
+ }
+
+ @Override
+ public void registerSharedStates(SharedStateRegistry stateRegistry) {
+ Preconditions.checkState(!registered, "The state handle has
already registered its shared states.");
+
+ for (Map.Entry<String, StreamStateHandle> sstFileEntry :
sstFiles.entrySet()) {
+ SstFileStateHandle stateHandle = new
SstFileStateHandle(sstFileEntry.getKey(), sstFileEntry.getValue());
+
+ int referenceCount =
stateRegistry.register(stateHandle);
+
+ if (newSstFileNames.contains(sstFileEntry.getKey())) {
+ Preconditions.checkState(referenceCount == 1);
+ } else {
+ Preconditions.checkState(referenceCount > 1);
+ }
+ }
+
+ registered = true;
+ }
+
+ @Override
+ public void unregisterSharedStates(SharedStateRegistry stateRegistry) {
+ Preconditions.checkState(registered, "The state handle has not
registered its shared states yet.");
+
+ for (Map.Entry<String, StreamStateHandle> sstFileEntry :
sstFiles.entrySet()) {
+ stateRegistry.unregister(new
SstFileStateHandle(sstFileEntry.getKey(), sstFileEntry.getValue()));
+ }
+
+ registered = false;
+ }
+
+ private class SstFileStateHandle implements SharedStateHandle {
--- End diff --
Looking a bit ahead, the concept of a state handle to shared files might be
something to reuse in other future cases, e.g. imagine incremental snapshots
for the heap based backend. I suggest to move this in full class named
`SharedFileStateHandle`. Conceptionally and implementation-wise this could be
`extends FileStateHandle implements SharedStateHandle`.
> Implement incremental checkpointing in RocksDBStateBackend
> ----------------------------------------------------------
>
> Key: FLINK-6364
> URL: https://issues.apache.org/jira/browse/FLINK-6364
> Project: Flink
> Issue Type: Sub-task
> Components: State Backends, Checkpointing
> Reporter: Xiaogang Shi
> Assignee: Xiaogang Shi
>
> {{RocksDBStateBackend}} is well suited for incremental checkpointing because
> RocksDB is base on LSM trees, which record updates in new sst files and all
> sst files are immutable. By only materializing those new sst files, we can
> significantly improve the performance of checkpointing.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
