You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "StefanRRichter (via GitHub)" <gi...@apache.org> on 2024/01/05 14:34:50 UTC
[PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
StefanRRichter opened a new pull request, #24031:
URL: https://github.com/apache/flink/pull/24031
Rebased and slightly refactored version of #23169.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1443451955
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java:
##########
@@ -185,6 +187,33 @@ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
return registeredStateMetaInfoEntry;
}
+ RocksDbKvStateInfo registerStateColumnFamilyHandleWithImport(
+ RegisteredStateMetaInfoBase stateMetaInfo,
+ List<ExportImportFilesMetaData> cfMetaDataList) {
+
+ RocksDbKvStateInfo registeredStateMetaInfoEntry =
+ kvStateInformation.get(stateMetaInfo.getName());
+
+ Preconditions.checkState(registeredStateMetaInfoEntry == null);
+
+ registeredStateMetaInfoEntry =
+ RocksDBOperationUtils.createStateInfo(
+ stateMetaInfo,
+ db,
+ columnFamilyOptionsFactory,
+ ttlCompactFiltersManager,
+ writeBufferManagerCapacity,
+ cfMetaDataList);
+
+ RocksDBOperationUtils.registerKvStateInformation(
+ kvStateInformation,
+ nativeMetricMonitor,
+ stateMetaInfo.getName(),
+ registeredStateMetaInfoEntry);
+
+ return registeredStateMetaInfoEntry;
Review Comment:
The return value isn't used, so return `void` and inline `registeredStateMetaInfoEntry `?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "mayuehappy (via GitHub)" <gi...@apache.org>.
mayuehappy commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481169952
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -539,13 +552,14 @@ private void rescaleClipIngestDB(
}
// If there are temporary instances for which we detected overlap in the buffer or
- // if we have remaining handles to restore, we will insert by copy with from temporary
- // instances to base DB.
+ // if we have remaining unopened handles to restore, we will insert by copy via
+ // temporary instances to base DB.
Review Comment:
In the old code, we called `choseTheBestStateHandleForInitial` to choose the best state handle to init the initial db .Because if we use TheBestHandle as a temporary DB instead of the main DB, we may need to write a lot of data when copying. Can we maintain this logic int the new code ?
When choosing db to export, we prioritize `TheBestHandle` to ensure that we will not copy TheBestHandle during the subsequent copying phase. Can this ensure that there is no regression compared to the old code
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -488,6 +495,8 @@ private void rescaleClipIngestDB(
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
Review Comment:
Here we need to check
Is key in proclaimed range or is there overlap between the checking stateHandles?
For example, the range of proclaimed is [1,5] [6,10] but the actual range is [1,7] [8,9]. Should it be possible to export in this case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481180416
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -539,13 +552,14 @@ private void rescaleClipIngestDB(
}
// If there are temporary instances for which we detected overlap in the buffer or
- // if we have remaining handles to restore, we will insert by copy with from temporary
- // instances to base DB.
+ // if we have remaining unopened handles to restore, we will insert by copy via
+ // temporary instances to base DB.
Review Comment:
Yes, I'm already working on adding that code again here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447406064
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java:
##########
@@ -230,6 +252,19 @@ private static ColumnFamilyHandle createColumnFamily(
}
}
+ private static ColumnFamilyHandle createColumnFamilyWithImport(
+ ColumnFamilyDescriptor columnDescriptor,
+ RocksDB db,
+ List<ExportImportFilesMetaData> metaDataList) {
+ try {
+ return db.createColumnFamilyWithImport(
+ columnDescriptor, new ImportColumnFamilyOptions(), metaDataList);
+ } catch (RocksDBException e) {
+ IOUtils.closeQuietly(columnDescriptor.getOptions());
+ throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", e);
Review Comment:
I modified the code in a different way that makes sure the cleanup happens through the owner of the objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "mayuehappy (via GitHub)" <gi...@apache.org>.
mayuehappy commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481216701
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -310,98 +322,275 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
.map(StateHandleDownloadSpec::createLocalStateHandleForDownloadedState)
.forEach(localKeyedStateHandles::add);
+ // Transfer remaining key-groups from temporary instance into base DB
+ byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+
+ byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+
+ try {
+ // Process all state downloads
+ transferRemoteStateToLocalDirectory(allDownloadSpecs);
+
+ if (useIngestDbRestoreMode) {
+ // Optimized path for merging multiple handles with Ingest/Clip
+ rescaleClipIngestDB(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } else {
+ // Optimized path for single handle and legacy path for merging multiple handles.
+ rescaleCopyFromTemporaryInstance(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ }
+
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
// Choose the best state handle for the initial DB
final IncrementalLocalKeyedStateHandle selectedInitialHandle =
RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
Preconditions.checkNotNull(selectedInitialHandle);
+
// Remove the selected handle from the list so that we don't restore it twice.
localKeyedStateHandles.remove(selectedInitialHandle);
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ try (RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize)) {
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
+ copyTempDbIntoBaseDb(
+ tmpRestoreDBInfo,
+ writeBatchWrapper,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+ }
+ }
+ }
+ }
+
+ private void copyTempDbIntoBaseDb(
+ RestoredDBInstance tmpRestoreDBInfo,
+ RocksDBWriteBatchWrapper writeBatchWrapper,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ logger.info(
+ "Starting copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles = tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null, tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
+ Files.createDirectories(exportCfBasePath);
+
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ exportedColumnFamilyMetaData = new HashMap<>();
+
+ final List<IncrementalLocalKeyedStateHandle> restoreHandles;
+ if (localKeyedStateHandles instanceof ArrayList) {
+ restoreHandles = (List<IncrementalLocalKeyedStateHandle>) localKeyedStateHandles;
+ } else {
+ restoreHandles = new ArrayList<>(localKeyedStateHandles);
+ }
+
+ // This buffer holds open temporary RocksDB instances that we create from individual handles
+ // and acts as a lookahead buffer, so that we don't need to immediately give up on
+ // import/export when we encounter a database that exceeds the proclaimed key-groups range
+ // of its source state handle.
+ final ArrayDeque<RestoredDBInstance> tmpInstanceBuffer =
+ new ArrayDeque<>(incrementalRestoreInstanceBufferSize);
+
try {
- // Process all state downloads
- transferRemoteStateToLocalDirectory(allDownloadSpecs);
+ int restoreIndex = 0;
+ // Initially, we skip the last handle for the case that we don't find any temporary DB
+ // that can be imported and must resort to the copy approach. Then we can open the base
+ // DB from the skipped handle and save one copy.
+ int skipLastHandle = 1;
+ while (restoreIndex < (restoreHandles.size() - skipLastHandle)
+ && tmpInstanceBuffer.size() < incrementalRestoreInstanceBufferSize) {
- // Init the base DB instance with the initial state
- initBaseDBForRescaling(selectedInitialHandle);
+ IncrementalLocalKeyedStateHandle stateHandle = restoreHandles.get(restoreIndex++);
- // Transfer remaining key-groups from temporary instance into base DB
- byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+ RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
- byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+ tmpInstanceBuffer.add(tmpRestoreDBInfo);
- // Insert all remaining state through creating temporary RocksDB instances
- for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
- logger.info(
- "Starting to restore from state handle: {} with rescaling.", stateHandle);
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
+ if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
+ tmpRestoreDBInfo.db, keyGroupPrefixBytes, stateHandle.getKeyGroupRange())) {
- try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromLocalState(stateHandle);
- RocksDBWriteBatchWrapper writeBatchWrapper =
- new RocksDBWriteBatchWrapper(
- this.rocksHandle.getDb(), writeBatchSize)) {
-
- List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
- tmpRestoreDBInfo.columnFamilyDescriptors;
- List<ColumnFamilyHandle> tmpColumnFamilyHandles =
- tmpRestoreDBInfo.columnFamilyHandles;
-
- // iterating only the requested descriptors automatically skips the default
- // column
- // family handle
- for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
- ColumnFamilyHandle tmpColumnFamilyHandle =
- tmpColumnFamilyHandles.get(descIdx);
-
- ColumnFamilyHandle targetColumnFamilyHandle =
- this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
- null,
- tmpRestoreDBInfo.stateMetaInfoSnapshots.get(
- descIdx))
- .columnFamilyHandle;
-
- try (RocksIteratorWrapper iterator =
- RocksDBOperationUtils.getRocksIterator(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandle,
- tmpRestoreDBInfo.readOptions)) {
-
- iterator.seek(startKeyGroupPrefixBytes);
-
- while (iterator.isValid()) {
-
- if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
- iterator.key(), stopKeyGroupPrefixBytes)) {
- writeBatchWrapper.put(
- targetColumnFamilyHandle,
- iterator.key(),
- iterator.value());
- } else {
- // Since the iterator will visit the record according to the
- // sorted
- // order,
- // we can just break here.
- break;
- }
-
- iterator.next();
- }
- } // releases native iterator resources
- }
logger.info(
- "Finished restoring from state handle: {} with rescaling.",
+ "Exporting state handle {} for rescaling with Clip/Ingest DB.",
stateHandle);
+ // Use Range delete to clip the temp db to the target range of the backend
+ RocksDBIncrementalCheckpointUtils.clipDBWithKeyGroupRange(
Review Comment:
Another point is, if these handles do not overlaps, can we first import them before clip them with deleteRange.
Because in current code, first we deleteRange and then exports, each DB will generate a new small file containing the RangeDeletion tombstone during export. So can we deleteRange together after importing, so that we can reduce the number of small files?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447413810
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java:
##########
@@ -305,6 +311,11 @@ private EmbeddedRocksDBStateBackend(
overlapFractionThreshold >= 0 && this.overlapFractionThreshold <= 1,
"Overlap fraction threshold of restoring should be between 0 and 1");
+ useIngestDbRestoreMode =
+ original.useIngestDbRestoreMode == UNDEFINED_USE_INGEST_DB_RESTORE_MODE
+ ? config.get(USE_INGEST_DB_RESTORE_MODE)
+ : original.useIngestDbRestoreMode;
Review Comment:
Yep, I think we better use `TernaryBoolean`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1448593360
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java:
##########
@@ -234,7 +247,8 @@ public void setupRocksKeyedStateBackend() throws Exception {
spy(db),
defaultCFHandle,
optionsContainer.getColumnOptions())
- .setEnableIncrementalCheckpointing(enableIncrementalCheckpointing);
+ .setEnableIncrementalCheckpointing(enableIncrementalCheckpointing)
+ .setUseIngestDbRestoreMode(useIngestDB);
Review Comment:
I see that `testCorrectMergeOperatorSet` sets up the backend independently of this method - we could also add `setUseIngestDbRestoreMode` there.
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java:
##########
@@ -154,6 +164,67 @@ private static void deleteRange(
}
}
+ /**
+ * Clip the entries in the CF according to the range [begin_key, end_key). Any entries outside
+ * this range will be completely deleted (including tombstones).
+ *
+ * @param db the target need to be clipped.
+ * @param columnFamilyHandles the column family need to be clipped.
+ * @param beginKeyBytes the begin key bytes
+ * @param endKeyBytes the end key bytes
+ */
+ public static void clipColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ byte[] beginKeyBytes,
+ byte[] endKeyBytes)
+ throws RocksDBException {
+
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ db.clipColumnFamily(columnFamilyHandle, beginKeyBytes, endKeyBytes);
+ // TODO: temporary fix until https://github.com/facebook/rocksdb/pull/12219
+ // is in the frocksDB release.
+ db.compactRange(columnFamilyHandle);
+ }
+ }
+
+ public static void exportColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ List<StateMetaInfoSnapshot> stateMetaInfoSnapshots,
+ Path exportBasePath,
+ Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>> resultOutput)
+ throws RocksDBException {
+
+ Preconditions.checkArgument(
+ columnFamilyHandles.size() == stateMetaInfoSnapshots.size(),
+ "Lists are aligned by index and must be of the same size!");
+
+ try (final Checkpoint checkpoint = Checkpoint.create(db)) {
+ for (int i = 0; i < columnFamilyHandles.size(); i++) {
+ StateMetaInfoSnapshot metaInfoSnapshot = stateMetaInfoSnapshots.get(i);
+
+ RegisteredStateMetaInfoBase stateMetaInfo =
+ RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(metaInfoSnapshot);
+
+ Path subPath = exportBasePath.resolve(UUID.randomUUID().toString());
+ ExportImportFilesMetaData cfMetaData =
+ checkpoint.exportColumnFamily(
+ columnFamilyHandles.get(i), subPath.toString());
+
+ File[] exportedSstFiles =
+ subPath.toFile()
+ .listFiles((file, name) -> name.toLowerCase().endsWith(".sst"));
+
+ if (exportedSstFiles != null && exportedSstFiles.length > 0) {
+ resultOutput
+ .computeIfAbsent(stateMetaInfo, (key) -> new ArrayList<>())
+ .add(cfMetaData);
+ }
Review Comment:
`else cfMetaData.close();` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24031:
URL: https://github.com/apache/flink/pull/24031#issuecomment-1910661954
One more thing, we should probably update the UI and show the new flag under `Checkpoints / Configuration`. But that can be a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "mayuehappy (via GitHub)" <gi...@apache.org>.
mayuehappy commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481206438
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -488,6 +495,8 @@ private void rescaleClipIngestDB(
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
Review Comment:
haha , that makes sense to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481189095
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -488,6 +495,8 @@ private void rescaleClipIngestDB(
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
Review Comment:
It is possible to import the case in your example, but the cost of detecting the case is that we need to compare all handles against each other which we can only do after all DBs have been opened. I'm excluding this case on purpose to avoid it, because I also assume that this missed opportunities will be very very rare. In particular because subtasks over time will compact their data to the proclaimed range from normal compaction activity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447508385
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksIncrementalCheckpointRescalingTest.java:
##########
@@ -41,14 +41,26 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
/** Tests to guard rescaling from checkpoint. */
+@RunWith(Parameterized.class)
public class RocksIncrementalCheckpointRescalingTest extends TestLogger {
@Rule public TemporaryFolder rootFolder = new TemporaryFolder();
+ @Parameterized.Parameters(name = "useIngestDbRestoreMode: {0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @Parameterized.Parameter public boolean useIngestDbRestoreMode;
Review Comment:
Not the FileStateBackend, because this only affects RocksDB. But there are also more rescaling tests that could cover this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481204289
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -488,6 +495,8 @@ private void rescaleClipIngestDB(
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
Review Comment:
Why I say it's rare because it means that while the handle [6,10] was created by a task with range [6,10] we never wrote a key from that falls into key-group 6 or 7 if the actual range is just [8, 9]. Because of the hash partitioning nature, that is highly unlikely if the task didn't idle the whole time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24031:
URL: https://github.com/apache/flink/pull/24031#issuecomment-1878781250
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "056b9af12d0795dadabde47e50dfb040e14ad71d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "056b9af12d0795dadabde47e50dfb040e14ad71d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 056b9af12d0795dadabde47e50dfb040e14ad71d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter merged PR #24031:
URL: https://github.com/apache/flink/pull/24031
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481219483
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -310,98 +322,275 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
.map(StateHandleDownloadSpec::createLocalStateHandleForDownloadedState)
.forEach(localKeyedStateHandles::add);
+ // Transfer remaining key-groups from temporary instance into base DB
+ byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+
+ byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+
+ try {
+ // Process all state downloads
+ transferRemoteStateToLocalDirectory(allDownloadSpecs);
+
+ if (useIngestDbRestoreMode) {
+ // Optimized path for merging multiple handles with Ingest/Clip
+ rescaleClipIngestDB(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } else {
+ // Optimized path for single handle and legacy path for merging multiple handles.
+ rescaleCopyFromTemporaryInstance(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ }
+
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
// Choose the best state handle for the initial DB
final IncrementalLocalKeyedStateHandle selectedInitialHandle =
RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
Preconditions.checkNotNull(selectedInitialHandle);
+
// Remove the selected handle from the list so that we don't restore it twice.
localKeyedStateHandles.remove(selectedInitialHandle);
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ try (RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize)) {
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
+ copyTempDbIntoBaseDb(
+ tmpRestoreDBInfo,
+ writeBatchWrapper,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+ }
+ }
+ }
+ }
+
+ private void copyTempDbIntoBaseDb(
+ RestoredDBInstance tmpRestoreDBInfo,
+ RocksDBWriteBatchWrapper writeBatchWrapper,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ logger.info(
+ "Starting copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles = tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null, tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
+ Files.createDirectories(exportCfBasePath);
+
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ exportedColumnFamilyMetaData = new HashMap<>();
+
+ final List<IncrementalLocalKeyedStateHandle> restoreHandles;
+ if (localKeyedStateHandles instanceof ArrayList) {
+ restoreHandles = (List<IncrementalLocalKeyedStateHandle>) localKeyedStateHandles;
+ } else {
+ restoreHandles = new ArrayList<>(localKeyedStateHandles);
+ }
+
+ // This buffer holds open temporary RocksDB instances that we create from individual handles
+ // and acts as a lookahead buffer, so that we don't need to immediately give up on
+ // import/export when we encounter a database that exceeds the proclaimed key-groups range
+ // of its source state handle.
+ final ArrayDeque<RestoredDBInstance> tmpInstanceBuffer =
+ new ArrayDeque<>(incrementalRestoreInstanceBufferSize);
+
try {
- // Process all state downloads
- transferRemoteStateToLocalDirectory(allDownloadSpecs);
+ int restoreIndex = 0;
+ // Initially, we skip the last handle for the case that we don't find any temporary DB
+ // that can be imported and must resort to the copy approach. Then we can open the base
+ // DB from the skipped handle and save one copy.
+ int skipLastHandle = 1;
+ while (restoreIndex < (restoreHandles.size() - skipLastHandle)
+ && tmpInstanceBuffer.size() < incrementalRestoreInstanceBufferSize) {
- // Init the base DB instance with the initial state
- initBaseDBForRescaling(selectedInitialHandle);
+ IncrementalLocalKeyedStateHandle stateHandle = restoreHandles.get(restoreIndex++);
- // Transfer remaining key-groups from temporary instance into base DB
- byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+ RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
- byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+ tmpInstanceBuffer.add(tmpRestoreDBInfo);
- // Insert all remaining state through creating temporary RocksDB instances
- for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
- logger.info(
- "Starting to restore from state handle: {} with rescaling.", stateHandle);
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
+ if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
+ tmpRestoreDBInfo.db, keyGroupPrefixBytes, stateHandle.getKeyGroupRange())) {
- try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromLocalState(stateHandle);
- RocksDBWriteBatchWrapper writeBatchWrapper =
- new RocksDBWriteBatchWrapper(
- this.rocksHandle.getDb(), writeBatchSize)) {
-
- List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
- tmpRestoreDBInfo.columnFamilyDescriptors;
- List<ColumnFamilyHandle> tmpColumnFamilyHandles =
- tmpRestoreDBInfo.columnFamilyHandles;
-
- // iterating only the requested descriptors automatically skips the default
- // column
- // family handle
- for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
- ColumnFamilyHandle tmpColumnFamilyHandle =
- tmpColumnFamilyHandles.get(descIdx);
-
- ColumnFamilyHandle targetColumnFamilyHandle =
- this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
- null,
- tmpRestoreDBInfo.stateMetaInfoSnapshots.get(
- descIdx))
- .columnFamilyHandle;
-
- try (RocksIteratorWrapper iterator =
- RocksDBOperationUtils.getRocksIterator(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandle,
- tmpRestoreDBInfo.readOptions)) {
-
- iterator.seek(startKeyGroupPrefixBytes);
-
- while (iterator.isValid()) {
-
- if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
- iterator.key(), stopKeyGroupPrefixBytes)) {
- writeBatchWrapper.put(
- targetColumnFamilyHandle,
- iterator.key(),
- iterator.value());
- } else {
- // Since the iterator will visit the record according to the
- // sorted
- // order,
- // we can just break here.
- break;
- }
-
- iterator.next();
- }
- } // releases native iterator resources
- }
logger.info(
- "Finished restoring from state handle: {} with rescaling.",
+ "Exporting state handle {} for rescaling with Clip/Ingest DB.",
stateHandle);
+ // Use Range delete to clip the temp db to the target range of the backend
+ RocksDBIncrementalCheckpointUtils.clipDBWithKeyGroupRange(
Review Comment:
Makes sense to try that. Your code was doing the delete before they export, so I was also wondering about this but didn't bother to change. Let me try that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1481300226
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -310,98 +322,275 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
.map(StateHandleDownloadSpec::createLocalStateHandleForDownloadedState)
.forEach(localKeyedStateHandles::add);
+ // Transfer remaining key-groups from temporary instance into base DB
+ byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+
+ byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
+ CompositeKeySerializationUtils.serializeKeyGroup(
+ keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+
+ try {
+ // Process all state downloads
+ transferRemoteStateToLocalDirectory(allDownloadSpecs);
+
+ if (useIngestDbRestoreMode) {
+ // Optimized path for merging multiple handles with Ingest/Clip
+ rescaleClipIngestDB(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } else {
+ // Optimized path for single handle and legacy path for merging multiple handles.
+ rescaleCopyFromTemporaryInstance(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ }
+
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
// Choose the best state handle for the initial DB
final IncrementalLocalKeyedStateHandle selectedInitialHandle =
RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
Preconditions.checkNotNull(selectedInitialHandle);
+
// Remove the selected handle from the list so that we don't restore it twice.
localKeyedStateHandles.remove(selectedInitialHandle);
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ try (RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(this.rocksHandle.getDb(), writeBatchSize)) {
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
+ copyTempDbIntoBaseDb(
+ tmpRestoreDBInfo,
+ writeBatchWrapper,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+ }
+ }
+ }
+ }
+
+ private void copyTempDbIntoBaseDb(
+ RestoredDBInstance tmpRestoreDBInfo,
+ RocksDBWriteBatchWrapper writeBatchWrapper,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ logger.info(
+ "Starting copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles = tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null, tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished copy of state handle {} to base DB for rescaling.",
+ tmpRestoreDBInfo.srcStateHandle);
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
+ Files.createDirectories(exportCfBasePath);
+
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ exportedColumnFamilyMetaData = new HashMap<>();
+
+ final List<IncrementalLocalKeyedStateHandle> restoreHandles;
+ if (localKeyedStateHandles instanceof ArrayList) {
+ restoreHandles = (List<IncrementalLocalKeyedStateHandle>) localKeyedStateHandles;
+ } else {
+ restoreHandles = new ArrayList<>(localKeyedStateHandles);
+ }
+
+ // This buffer holds open temporary RocksDB instances that we create from individual handles
+ // and acts as a lookahead buffer, so that we don't need to immediately give up on
+ // import/export when we encounter a database that exceeds the proclaimed key-groups range
+ // of its source state handle.
+ final ArrayDeque<RestoredDBInstance> tmpInstanceBuffer =
+ new ArrayDeque<>(incrementalRestoreInstanceBufferSize);
+
try {
- // Process all state downloads
- transferRemoteStateToLocalDirectory(allDownloadSpecs);
+ int restoreIndex = 0;
+ // Initially, we skip the last handle for the case that we don't find any temporary DB
+ // that can be imported and must resort to the copy approach. Then we can open the base
+ // DB from the skipped handle and save one copy.
+ int skipLastHandle = 1;
+ while (restoreIndex < (restoreHandles.size() - skipLastHandle)
+ && tmpInstanceBuffer.size() < incrementalRestoreInstanceBufferSize) {
- // Init the base DB instance with the initial state
- initBaseDBForRescaling(selectedInitialHandle);
+ IncrementalLocalKeyedStateHandle stateHandle = restoreHandles.get(restoreIndex++);
- // Transfer remaining key-groups from temporary instance into base DB
- byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+ RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
- byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
+ tmpInstanceBuffer.add(tmpRestoreDBInfo);
- // Insert all remaining state through creating temporary RocksDB instances
- for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
- logger.info(
- "Starting to restore from state handle: {} with rescaling.", stateHandle);
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // Check if the data in all SST files referenced in the handle is within the
+ // proclaimed key-groups range of the handle.
+ if (RocksDBIncrementalCheckpointUtils.isSstDataInKeyGroupRange(
+ tmpRestoreDBInfo.db, keyGroupPrefixBytes, stateHandle.getKeyGroupRange())) {
- try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromLocalState(stateHandle);
- RocksDBWriteBatchWrapper writeBatchWrapper =
- new RocksDBWriteBatchWrapper(
- this.rocksHandle.getDb(), writeBatchSize)) {
-
- List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
- tmpRestoreDBInfo.columnFamilyDescriptors;
- List<ColumnFamilyHandle> tmpColumnFamilyHandles =
- tmpRestoreDBInfo.columnFamilyHandles;
-
- // iterating only the requested descriptors automatically skips the default
- // column
- // family handle
- for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
- ColumnFamilyHandle tmpColumnFamilyHandle =
- tmpColumnFamilyHandles.get(descIdx);
-
- ColumnFamilyHandle targetColumnFamilyHandle =
- this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
- null,
- tmpRestoreDBInfo.stateMetaInfoSnapshots.get(
- descIdx))
- .columnFamilyHandle;
-
- try (RocksIteratorWrapper iterator =
- RocksDBOperationUtils.getRocksIterator(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandle,
- tmpRestoreDBInfo.readOptions)) {
-
- iterator.seek(startKeyGroupPrefixBytes);
-
- while (iterator.isValid()) {
-
- if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
- iterator.key(), stopKeyGroupPrefixBytes)) {
- writeBatchWrapper.put(
- targetColumnFamilyHandle,
- iterator.key(),
- iterator.value());
- } else {
- // Since the iterator will visit the record according to the
- // sorted
- // order,
- // we can just break here.
- break;
- }
-
- iterator.next();
- }
- } // releases native iterator resources
- }
logger.info(
- "Finished restoring from state handle: {} with rescaling.",
+ "Exporting state handle {} for rescaling with Clip/Ingest DB.",
stateHandle);
+ // Use Range delete to clip the temp db to the target range of the backend
+ RocksDBIncrementalCheckpointUtils.clipDBWithKeyGroupRange(
Review Comment:
I changed the code accordingly. It didn't work at first because the import code didn't add the ColumnFamilyHandles to the DbHandle. Now it seems to work :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1442957056
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -414,90 +424,70 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
* temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
* are copied into the real restore instance and then the temporary instance is discarded.
*/
- private void restoreWithIngestDbMode(Collection<KeyedStateHandle> restoreStateHandles)
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
throws Exception {
- Preconditions.checkArgument(restoreStateHandles != null && !restoreStateHandles.isEmpty());
-
- Map<StateHandleID, StateHandleDownloadSpec> allDownloadSpecs =
- CollectionUtil.newHashMapWithExpectedSize(restoreStateHandles.size());
-
final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
Files.createDirectories(exportCfBasePath);
- // Open base db as Empty DB
- this.rocksHandle.openDB();
-
- // Prepare and collect all the download request to pull remote state to a local directory
- for (KeyedStateHandle stateHandle : restoreStateHandles) {
- if (!(stateHandle instanceof IncrementalRemoteKeyedStateHandle)) {
- throw unexpectedStateHandleException(
- IncrementalRemoteKeyedStateHandle.class, stateHandle.getClass());
- }
- StateHandleDownloadSpec downloadRequest =
- new StateHandleDownloadSpec(
- (IncrementalRemoteKeyedStateHandle) stateHandle,
- absolutInstanceBasePath.resolve(UUID.randomUUID().toString()));
- allDownloadSpecs.put(stateHandle.getStateHandleId(), downloadRequest);
- }
-
- // Process all state downloads
- transferRemoteStateToLocalDirectory(allDownloadSpecs.values());
-
- // Transfer remaining key-groups from temporary instance into base DB
- byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ columnFamilyMetaDataToImport = new HashMap<>();
- byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
- CompositeKeySerializationUtils.serializeKeyGroup(
- keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
-
- HashMap<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>> cfMetaDataToImport =
- new HashMap();
-
- // Insert all remaining state through creating temporary RocksDB instances
- for (StateHandleDownloadSpec downloadRequest : allDownloadSpecs.values()) {
- logger.info(
- "Starting to restore from state handle: {} with rescaling.",
- downloadRequest.getStateHandle());
+ try {
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ logger.info(
+ "Starting to restore from state handle: {} with rescaling using Clip/Ingest DB.",
+ stateHandle);
- try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromDownloadedState(downloadRequest)) {
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
- List<ColumnFamilyHandle> tmpColumnFamilyHandles =
- tmpRestoreDBInfo.columnFamilyHandles;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
- // Clip all tmp db to Range [startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes)
- RocksDBIncrementalCheckpointUtils.clipColumnFamilies(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandles,
- startKeyGroupPrefixBytes,
- stopKeyGroupPrefixBytes);
-
- // Export all the Column Families
- Map<RegisteredStateMetaInfoBase, ExportImportFilesMetaData> exportedCFAndMetaData =
- RocksDBIncrementalCheckpointUtils.exportColumnFamilies(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandles,
- tmpRestoreDBInfo.stateMetaInfoSnapshots,
- exportCfBasePath);
-
- exportedCFAndMetaData.forEach(
- (stateMeta, cfMetaData) -> {
- if (!cfMetaData.files().isEmpty()) {
- cfMetaDataToImport.putIfAbsent(stateMeta, new ArrayList<>());
- cfMetaDataToImport.get(stateMeta).add(cfMetaData);
- }
- });
- } finally {
- cleanUpPathQuietly(downloadRequest.getDownloadDestination());
+ // Clip all tmp db to Range [startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes)
+ RocksDBIncrementalCheckpointUtils.clipColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+
+ // Export all the Column Families
+ List<Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData>>
+ exportedCFAndMetaData =
+ RocksDBIncrementalCheckpointUtils.exportColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots,
+ exportCfBasePath);
+
+ for (Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData> entry :
+ exportedCFAndMetaData) {
+ ExportImportFilesMetaData cfMetaData = entry.getValue();
+ // TODO: method files() doesn't exist in the RocksDB API
Review Comment:
@mayuehappy Is it correct to remove this code? I could not find a files() method in the RocksDB API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1443459209
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -105,6 +110,10 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private boolean isKeySerializerCompatibilityChecked;
+ private final TriConsumerWithException<
+ Collection<IncrementalLocalKeyedStateHandle>, byte[], byte[], Exception>
+ rescalingRestoreFromLocalStateOperation;
+
Review Comment:
This can be replaced by a boolean flag (`useIngestDbRestoreMode`) and a "normal" `if` and call in `restoreWithRescaling`.
I think it would be more readable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447132521
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -329,74 +332,164 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
CompositeKeySerializationUtils.serializeKeyGroup(
keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
- // Insert all remaining state through creating temporary RocksDB instances
+ rescalingRestoreFromLocalStateOperation.accept(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ // Choose the best state handle for the initial DB
+ final IncrementalLocalKeyedStateHandle selectedInitialHandle =
+ RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
+ localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
+ Preconditions.checkNotNull(selectedInitialHandle);
+
+ // Remove the selected handle from the list so that we don't restore it twice.
+ localKeyedStateHandles.remove(selectedInitialHandle);
+
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ logger.info("Starting to restore from state handle: {} with rescaling.", stateHandle);
+
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
+ RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(
+ this.rocksHandle.getDb(), writeBatchSize)) {
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished restoring from state handle: {} with rescaling.", stateHandle);
+ }
+ }
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
Review Comment:
The folder is temporary and the files will be hard-linked iirc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447413810
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java:
##########
@@ -305,6 +311,11 @@ private EmbeddedRocksDBStateBackend(
overlapFractionThreshold >= 0 && this.overlapFractionThreshold <= 1,
"Overlap fraction threshold of restoring should be between 0 and 1");
+ useIngestDbRestoreMode =
+ original.useIngestDbRestoreMode == UNDEFINED_USE_INGEST_DB_RESTORE_MODE
+ ? config.get(USE_INGEST_DB_RESTORE_MODE)
+ : original.useIngestDbRestoreMode;
Review Comment:
Yep, I think we better use `Boolean`and `null` for undefined.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1447131761
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -329,74 +332,164 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
CompositeKeySerializationUtils.serializeKeyGroup(
keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
- // Insert all remaining state through creating temporary RocksDB instances
+ rescalingRestoreFromLocalStateOperation.accept(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ // Choose the best state handle for the initial DB
+ final IncrementalLocalKeyedStateHandle selectedInitialHandle =
+ RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
+ localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
+ Preconditions.checkNotNull(selectedInitialHandle);
+
+ // Remove the selected handle from the list so that we don't restore it twice.
+ localKeyedStateHandles.remove(selectedInitialHandle);
+
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ logger.info("Starting to restore from state handle: {} with rescaling.", stateHandle);
+
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
+ RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(
+ this.rocksHandle.getDb(), writeBatchSize)) {
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished restoring from state handle: {} with rescaling.", stateHandle);
+ }
+ }
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
+ Files.createDirectories(exportCfBasePath);
+
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ columnFamilyMetaDataToImport = new HashMap<>();
+
+ try {
for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
logger.info(
- "Starting to restore from state handle: {} with rescaling.", stateHandle);
+ "Starting to restore from state handle: {} with rescaling using Clip/Ingest DB.",
+ stateHandle);
try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromLocalState(stateHandle);
- RocksDBWriteBatchWrapper writeBatchWrapper =
- new RocksDBWriteBatchWrapper(
- this.rocksHandle.getDb(), writeBatchSize)) {
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
- List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
- tmpRestoreDBInfo.columnFamilyDescriptors;
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
- // iterating only the requested descriptors automatically skips the default
- // column
- // family handle
- for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
- ColumnFamilyHandle tmpColumnFamilyHandle =
- tmpColumnFamilyHandles.get(descIdx);
-
- ColumnFamilyHandle targetColumnFamilyHandle =
- this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
- null,
- tmpRestoreDBInfo.stateMetaInfoSnapshots.get(
- descIdx))
- .columnFamilyHandle;
-
- try (RocksIteratorWrapper iterator =
- RocksDBOperationUtils.getRocksIterator(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandle,
- tmpRestoreDBInfo.readOptions)) {
-
- iterator.seek(startKeyGroupPrefixBytes);
-
- while (iterator.isValid()) {
-
- if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
- iterator.key(), stopKeyGroupPrefixBytes)) {
- writeBatchWrapper.put(
- targetColumnFamilyHandle,
- iterator.key(),
- iterator.value());
- } else {
- // Since the iterator will visit the record according to the
- // sorted
- // order,
- // we can just break here.
- break;
- }
-
- iterator.next();
- }
- } // releases native iterator resources
+ // Clip all tmp db to Range [startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes)
+ RocksDBIncrementalCheckpointUtils.clipColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+
+ // Export all the Column Families
+ List<Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData>>
+ exportedCFAndMetaData =
+ RocksDBIncrementalCheckpointUtils.exportColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots,
+ exportCfBasePath);
+
+ for (Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData> entry :
+ exportedCFAndMetaData) {
+ ExportImportFilesMetaData cfMetaData = entry.getValue();
+ // TODO: method files() doesn't exist in the RocksDB API
+ // if (cfMetaData.files().isEmpty()) {
+ // continue;
+ // }
+ columnFamilyMetaDataToImport
+ .computeIfAbsent(entry.getKey(), (k) -> new ArrayList<>())
+ .add(cfMetaData);
}
- logger.info(
- "Finished restoring from state handle: {} with rescaling.",
- stateHandle);
}
+ logger.info(
+ "Finished exporting column family from state handle: {} for rescaling.",
+ stateHandle);
}
+
+ // Open the target RocksDB and import the exported column families
+ this.rocksHandle.openDB();
+ columnFamilyMetaDataToImport.forEach(
+ this.rocksHandle::registerStateColumnFamilyHandleWithImport);
+ logger.info(
+ "Finished importing exported column families into target DB for rescaling.");
} finally {
- // Cleanup all download directories
- allDownloadSpecs.stream()
- .map(StateHandleDownloadSpec::getDownloadDestination)
- .forEach(this::cleanUpPathQuietly);
+ cleanUpPathQuietly(exportCfBasePath);
Review Comment:
No, I think the files are hard-linked in the directory of the created instance to which we import.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "StefanRRichter (via GitHub)" <gi...@apache.org>.
StefanRRichter commented on PR #24031:
URL: https://github.com/apache/flink/pull/24031#issuecomment-1885031318
@rkhachatryan Can you take one more look? I have fixed all problems that I encountered in the original PR and also addressed your comments. Using `TestStreamEnvironment.randomizeConfiguration` seems a bit problematic because the RocksDB config keys are not visible in that module. We also still need to change to frocksDB version and the notice file once we have the release.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1443420860
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java:
##########
@@ -305,6 +311,11 @@ private EmbeddedRocksDBStateBackend(
overlapFractionThreshold >= 0 && this.overlapFractionThreshold <= 1,
"Overlap fraction threshold of restoring should be between 0 and 1");
+ useIngestDbRestoreMode =
+ original.useIngestDbRestoreMode == UNDEFINED_USE_INGEST_DB_RESTORE_MODE
+ ? config.get(USE_INGEST_DB_RESTORE_MODE)
+ : original.useIngestDbRestoreMode;
Review Comment:
Won't this ignore `original.useIngestDbRestoreMode` if it's set to `false`?
(I guess this is a copy-paste from other fields, but they aren't boolean)
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -329,74 +332,164 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
CompositeKeySerializationUtils.serializeKeyGroup(
keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
- // Insert all remaining state through creating temporary RocksDB instances
+ rescalingRestoreFromLocalStateOperation.accept(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ // Choose the best state handle for the initial DB
+ final IncrementalLocalKeyedStateHandle selectedInitialHandle =
+ RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
+ localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
+ Preconditions.checkNotNull(selectedInitialHandle);
+
+ // Remove the selected handle from the list so that we don't restore it twice.
+ localKeyedStateHandles.remove(selectedInitialHandle);
+
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ logger.info("Starting to restore from state handle: {} with rescaling.", stateHandle);
+
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
+ RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(
+ this.rocksHandle.getDb(), writeBatchSize)) {
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished restoring from state handle: {} with rescaling.", stateHandle);
+ }
+ }
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
+ Files.createDirectories(exportCfBasePath);
+
+ final Map<RegisteredStateMetaInfoBase, List<ExportImportFilesMetaData>>
+ columnFamilyMetaDataToImport = new HashMap<>();
+
+ try {
for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
logger.info(
- "Starting to restore from state handle: {} with rescaling.", stateHandle);
+ "Starting to restore from state handle: {} with rescaling using Clip/Ingest DB.",
+ stateHandle);
try (RestoredDBInstance tmpRestoreDBInfo =
- restoreTempDBInstanceFromLocalState(stateHandle);
- RocksDBWriteBatchWrapper writeBatchWrapper =
- new RocksDBWriteBatchWrapper(
- this.rocksHandle.getDb(), writeBatchSize)) {
+ restoreTempDBInstanceFromLocalState(stateHandle)) {
- List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
- tmpRestoreDBInfo.columnFamilyDescriptors;
List<ColumnFamilyHandle> tmpColumnFamilyHandles =
tmpRestoreDBInfo.columnFamilyHandles;
- // iterating only the requested descriptors automatically skips the default
- // column
- // family handle
- for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
- ColumnFamilyHandle tmpColumnFamilyHandle =
- tmpColumnFamilyHandles.get(descIdx);
-
- ColumnFamilyHandle targetColumnFamilyHandle =
- this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
- null,
- tmpRestoreDBInfo.stateMetaInfoSnapshots.get(
- descIdx))
- .columnFamilyHandle;
-
- try (RocksIteratorWrapper iterator =
- RocksDBOperationUtils.getRocksIterator(
- tmpRestoreDBInfo.db,
- tmpColumnFamilyHandle,
- tmpRestoreDBInfo.readOptions)) {
-
- iterator.seek(startKeyGroupPrefixBytes);
-
- while (iterator.isValid()) {
-
- if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
- iterator.key(), stopKeyGroupPrefixBytes)) {
- writeBatchWrapper.put(
- targetColumnFamilyHandle,
- iterator.key(),
- iterator.value());
- } else {
- // Since the iterator will visit the record according to the
- // sorted
- // order,
- // we can just break here.
- break;
- }
-
- iterator.next();
- }
- } // releases native iterator resources
+ // Clip all tmp db to Range [startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes)
+ RocksDBIncrementalCheckpointUtils.clipColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ startKeyGroupPrefixBytes,
+ stopKeyGroupPrefixBytes);
+
+ // Export all the Column Families
+ List<Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData>>
+ exportedCFAndMetaData =
+ RocksDBIncrementalCheckpointUtils.exportColumnFamilies(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandles,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots,
+ exportCfBasePath);
+
+ for (Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData> entry :
+ exportedCFAndMetaData) {
+ ExportImportFilesMetaData cfMetaData = entry.getValue();
+ // TODO: method files() doesn't exist in the RocksDB API
+ // if (cfMetaData.files().isEmpty()) {
+ // continue;
+ // }
+ columnFamilyMetaDataToImport
+ .computeIfAbsent(entry.getKey(), (k) -> new ArrayList<>())
+ .add(cfMetaData);
}
- logger.info(
- "Finished restoring from state handle: {} with rescaling.",
- stateHandle);
}
+ logger.info(
+ "Finished exporting column family from state handle: {} for rescaling.",
+ stateHandle);
}
+
+ // Open the target RocksDB and import the exported column families
+ this.rocksHandle.openDB();
+ columnFamilyMetaDataToImport.forEach(
+ this.rocksHandle::registerStateColumnFamilyHandleWithImport);
+ logger.info(
+ "Finished importing exported column families into target DB for rescaling.");
} finally {
- // Cleanup all download directories
- allDownloadSpecs.stream()
- .map(StateHandleDownloadSpec::getDownloadDestination)
- .forEach(this::cleanUpPathQuietly);
+ cleanUpPathQuietly(exportCfBasePath);
Review Comment:
Just to clarify, `cleanUpPathQuietly(exportCfBasePath)` doesn't affect the opened RocksDb instance by deleting the imported files?
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java:
##########
@@ -154,6 +162,54 @@ private static void deleteRange(
}
}
+ /**
+ * Clip the entries in the CF according to the range [begin_key, end_key). Any entries outside
+ * this range will be completely deleted (including tombstones).
+ *
+ * @param db the target need to be clipped.
+ * @param columnFamilyHandles the column family need to be clipped.
+ * @param beginKeyBytes the begin key bytes
+ * @param endKeyBytes the end key bytes
+ */
+ public static void clipColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ byte[] beginKeyBytes,
+ byte[] endKeyBytes)
+ throws RocksDBException {
+
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ db.clipColumnFamily(columnFamilyHandle, beginKeyBytes, endKeyBytes);
+ }
+ }
+
+ public static List<Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData>>
+ exportColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ List<StateMetaInfoSnapshot> stateMetaInfoSnapshots,
Review Comment:
This to lists should match each other, right?
Should we check with `Preconditions.checkState(columnFamilyHandles.size() == stateMetaInfoSnapshots.size());`,
or better pass RocksDB instance here?
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java:
##########
@@ -185,6 +187,33 @@ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
return registeredStateMetaInfoEntry;
}
+ RocksDbKvStateInfo registerStateColumnFamilyHandleWithImport(
+ RegisteredStateMetaInfoBase stateMetaInfo,
+ List<ExportImportFilesMetaData> cfMetaDataList) {
+
+ RocksDbKvStateInfo registeredStateMetaInfoEntry =
+ kvStateInformation.get(stateMetaInfo.getName());
+
+ Preconditions.checkState(registeredStateMetaInfoEntry == null);
Review Comment:
Replace with
```
Preconditions.checkState(!kvStateInformation.containsKey(stateMetaInfo.getName()));
```
?
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksIncrementalCheckpointRescalingTest.java:
##########
@@ -419,7 +431,12 @@ private KeyedOneInputStreamOperatorTestHarness<String, String, Integer> getHarne
}
private StateBackend getStateBackend() throws Exception {
- return new RocksDBStateBackend("file://" + rootFolder.newFolder().getAbsolutePath(), true);
+ RocksDBStateBackend rocksDBStateBackend =
+ new RocksDBStateBackend("file://" + rootFolder.newFolder().getAbsolutePath(), true);
+ Configuration configuration = new Configuration();
Review Comment:
Missing import.
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackend.java:
##########
@@ -179,6 +182,8 @@ public class EmbeddedRocksDBStateBackend extends AbstractManagedMemoryStateBacke
*/
private double overlapFractionThreshold;
+ private boolean useIngestDbRestoreMode;
+
Review Comment:
Can be `final` according to IntelliJ :)
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java:
##########
@@ -123,16 +125,36 @@ public static RocksDBKeyedStateBackend.RocksDbKvStateInfo createStateInfo(
RocksDB db,
Function<String, ColumnFamilyOptions> columnFamilyOptionsFactory,
@Nullable RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
- @Nullable Long writeBufferManagerCapacity) {
+ @Nullable Long writeBufferManagerCapacity,
+ List<ExportImportFilesMetaData> cfMetaDataList) {
Review Comment:
Can't we use an empty list for `cfMetaDataList` instead of `null`?
If not, mark `@Nullable`?
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -329,74 +332,164 @@ private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandl
CompositeKeySerializationUtils.serializeKeyGroup(
keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
- // Insert all remaining state through creating temporary RocksDB instances
+ rescalingRestoreFromLocalStateOperation.accept(
+ localKeyedStateHandles, startKeyGroupPrefixBytes, stopKeyGroupPrefixBytes);
+ } finally {
+ // Cleanup all download directories
+ allDownloadSpecs.stream()
+ .map(StateHandleDownloadSpec::getDownloadDestination)
+ .forEach(this::cleanUpPathQuietly);
+ }
+ }
+
+ private void rescaleCopyFromTemporaryInstance(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ // Choose the best state handle for the initial DB
+ final IncrementalLocalKeyedStateHandle selectedInitialHandle =
+ RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
+ localKeyedStateHandles, keyGroupRange, overlapFractionThreshold);
+
+ Preconditions.checkNotNull(selectedInitialHandle);
+
+ // Remove the selected handle from the list so that we don't restore it twice.
+ localKeyedStateHandles.remove(selectedInitialHandle);
+
+ // Init the base DB instance with the initial state
+ initBaseDBForRescaling(selectedInitialHandle);
+
+ for (IncrementalLocalKeyedStateHandle stateHandle : localKeyedStateHandles) {
+ logger.info("Starting to restore from state handle: {} with rescaling.", stateHandle);
+
+ try (RestoredDBInstance tmpRestoreDBInfo =
+ restoreTempDBInstanceFromLocalState(stateHandle);
+ RocksDBWriteBatchWrapper writeBatchWrapper =
+ new RocksDBWriteBatchWrapper(
+ this.rocksHandle.getDb(), writeBatchSize)) {
+
+ List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors =
+ tmpRestoreDBInfo.columnFamilyDescriptors;
+ List<ColumnFamilyHandle> tmpColumnFamilyHandles =
+ tmpRestoreDBInfo.columnFamilyHandles;
+
+ // iterating only the requested descriptors automatically skips the default
+ // column
+ // family handle
+ for (int descIdx = 0; descIdx < tmpColumnFamilyDescriptors.size(); ++descIdx) {
+ ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(descIdx);
+
+ ColumnFamilyHandle targetColumnFamilyHandle =
+ this.rocksHandle.getOrRegisterStateColumnFamilyHandle(
+ null,
+ tmpRestoreDBInfo.stateMetaInfoSnapshots.get(descIdx))
+ .columnFamilyHandle;
+
+ try (RocksIteratorWrapper iterator =
+ RocksDBOperationUtils.getRocksIterator(
+ tmpRestoreDBInfo.db,
+ tmpColumnFamilyHandle,
+ tmpRestoreDBInfo.readOptions)) {
+
+ iterator.seek(startKeyGroupPrefixBytes);
+
+ while (iterator.isValid()) {
+
+ if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(
+ iterator.key(), stopKeyGroupPrefixBytes)) {
+ writeBatchWrapper.put(
+ targetColumnFamilyHandle, iterator.key(), iterator.value());
+ } else {
+ // Since the iterator will visit the record according to the
+ // sorted
+ // order,
+ // we can just break here.
+ break;
+ }
+
+ iterator.next();
+ }
+ } // releases native iterator resources
+ }
+ logger.info(
+ "Finished restoring from state handle: {} with rescaling.", stateHandle);
+ }
+ }
+ }
+
+ /**
+ * Recovery from multi incremental states with rescaling. For rescaling, this method creates a
+ * temporary RocksDB instance for a key-groups shard. All contents from the temporary instance
+ * are copied into the real restore instance and then the temporary instance is discarded.
+ */
+ private void rescaleClipIngestDB(
+ Collection<IncrementalLocalKeyedStateHandle> localKeyedStateHandles,
+ byte[] startKeyGroupPrefixBytes,
+ byte[] stopKeyGroupPrefixBytes)
+ throws Exception {
+
+ final Path absolutInstanceBasePath = instanceBasePath.getAbsoluteFile().toPath();
+ final Path exportCfBasePath = absolutInstanceBasePath.resolve("export-cfs");
Review Comment:
Just to clarify, this is not a temporary folder so that the files can be efficiently sym-linked?
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksIncrementalCheckpointRescalingTest.java:
##########
@@ -41,14 +41,26 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
/** Tests to guard rescaling from checkpoint. */
+@RunWith(Parameterized.class)
public class RocksIncrementalCheckpointRescalingTest extends TestLogger {
@Rule public TemporaryFolder rootFolder = new TemporaryFolder();
+ @Parameterized.Parameters(name = "useIngestDbRestoreMode: {0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @Parameterized.Parameter public boolean useIngestDbRestoreMode;
Review Comment:
It would be great to parameterise other unit-tests. Maybe `EmbeddedRocksDBStateBackendTest` and `FileStateBackendTest`?
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBHandle.java:
##########
@@ -185,6 +187,33 @@ RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
return registeredStateMetaInfoEntry;
}
+ RocksDbKvStateInfo registerStateColumnFamilyHandleWithImport(
+ RegisteredStateMetaInfoBase stateMetaInfo,
+ List<ExportImportFilesMetaData> cfMetaDataList) {
+
+ RocksDbKvStateInfo registeredStateMetaInfoEntry =
+ kvStateInformation.get(stateMetaInfo.getName());
+
+ Preconditions.checkState(registeredStateMetaInfoEntry == null);
+
+ registeredStateMetaInfoEntry =
+ RocksDBOperationUtils.createStateInfo(
+ stateMetaInfo,
+ db,
+ columnFamilyOptionsFactory,
+ ttlCompactFiltersManager,
+ writeBufferManagerCapacity,
+ cfMetaDataList);
+
+ RocksDBOperationUtils.registerKvStateInformation(
+ kvStateInformation,
+ nativeMetricMonitor,
+ stateMetaInfo.getName(),
+ registeredStateMetaInfoEntry);
+
+ return registeredStateMetaInfoEntry;
Review Comment:
The return value isn't used, so return `void` and inline?
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java:
##########
@@ -105,6 +110,10 @@ public class RocksDBIncrementalRestoreOperation<K> implements RocksDBRestoreOper
private boolean isKeySerializerCompatibilityChecked;
+ private final TriConsumerWithException<
+ Collection<IncrementalLocalKeyedStateHandle>, byte[], byte[], Exception>
+ rescalingRestoreFromLocalStateOperation;
+
Review Comment:
This can be replaced by a boolean flag (`useIngestDbRestoreMode`) and a "normal" if and call in `restoreWithRescaling`.
I think it would be more readable.
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBOperationUtils.java:
##########
@@ -230,6 +252,19 @@ private static ColumnFamilyHandle createColumnFamily(
}
}
+ private static ColumnFamilyHandle createColumnFamilyWithImport(
+ ColumnFamilyDescriptor columnDescriptor,
+ RocksDB db,
+ List<ExportImportFilesMetaData> metaDataList) {
+ try {
+ return db.createColumnFamilyWithImport(
+ columnDescriptor, new ImportColumnFamilyOptions(), metaDataList);
+ } catch (RocksDBException e) {
+ IOUtils.closeQuietly(columnDescriptor.getOptions());
+ throw new FlinkRuntimeException("Error creating ColumnFamilyHandle.", e);
Review Comment:
Missing `metaDataList.forEach(IOUtils::closeQuietly);` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1443421426
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBIncrementalCheckpointUtils.java:
##########
@@ -154,6 +162,54 @@ private static void deleteRange(
}
}
+ /**
+ * Clip the entries in the CF according to the range [begin_key, end_key). Any entries outside
+ * this range will be completely deleted (including tombstones).
+ *
+ * @param db the target need to be clipped.
+ * @param columnFamilyHandles the column family need to be clipped.
+ * @param beginKeyBytes the begin key bytes
+ * @param endKeyBytes the end key bytes
+ */
+ public static void clipColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ byte[] beginKeyBytes,
+ byte[] endKeyBytes)
+ throws RocksDBException {
+
+ for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandles) {
+ db.clipColumnFamily(columnFamilyHandle, beginKeyBytes, endKeyBytes);
+ }
+ }
+
+ public static List<Pair<RegisteredStateMetaInfoBase, ExportImportFilesMetaData>>
+ exportColumnFamilies(
+ RocksDB db,
+ List<ColumnFamilyHandle> columnFamilyHandles,
+ List<StateMetaInfoSnapshot> stateMetaInfoSnapshots,
Review Comment:
These two lists should match each other, right?
Should we check with `Preconditions.checkState(columnFamilyHandles.size() == stateMetaInfoSnapshots.size());`,
or better pass RocksDB instance here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-31238] [WIP] Use IngestDB to speed up Rocksdb rescaling recovery (rebased) [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24031:
URL: https://github.com/apache/flink/pull/24031#discussion_r1443422130
##########
flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksIncrementalCheckpointRescalingTest.java:
##########
@@ -419,7 +431,12 @@ private KeyedOneInputStreamOperatorTestHarness<String, String, Integer> getHarne
}
private StateBackend getStateBackend() throws Exception {
- return new RocksDBStateBackend("file://" + rootFolder.newFolder().getAbsolutePath(), true);
+ RocksDBStateBackend rocksDBStateBackend =
+ new RocksDBStateBackend("file://" + rootFolder.newFolder().getAbsolutePath(), true);
+ Configuration configuration = new Configuration();
Review Comment:
Missing import for `Configuration`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org