tkalkirill commented on code in PR #6329:
URL: https://github.com/apache/ignite-3/pull/6329#discussion_r2239153102
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -446,6 +448,43 @@ public void destroyRaftNodeStorages(RaftNodeId nodeId,
RaftGroupOptions raftGrou
}
}
+ /**
+ * Destroys Raft group node storages (log storage, metadata storage and
snapshots storage).
+ *
+ * <p>Destruction is durable: that is, if this method returns and after
that the node crashes, after it starts up, the storage
+ * will not be there.
+ *
+ * @param nodeId ID of the Raft node.
+ * @param raftGroupOptions Group options.
+ * @throws NodeStoppingException If the node is already being stopped.
+ */
+ public void destroyRaftNodeStoragesDurably(RaftNodeId nodeId,
RaftGroupOptions raftGroupOptions) throws NodeStoppingException {
Review Comment:
I think it is not necessary to indicate `throws NodeStoppingException`,
documentation is enough.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -446,6 +448,43 @@ public void destroyRaftNodeStorages(RaftNodeId nodeId,
RaftGroupOptions raftGrou
}
}
+ /**
+ * Destroys Raft group node storages (log storage, metadata storage and
snapshots storage).
+ *
+ * <p>Destruction is durable: that is, if this method returns and after
that the node crashes, after it starts up, the storage
+ * will not be there.
+ *
+ * @param nodeId ID of the Raft node.
+ * @param raftGroupOptions Group options.
+ * @throws NodeStoppingException If the node is already being stopped.
+ */
+ public void destroyRaftNodeStoragesDurably(RaftNodeId nodeId,
RaftGroupOptions raftGroupOptions) throws NodeStoppingException {
+ if (!busyLock.enterBusy()) {
Review Comment:
Can use methods `org.apache.ignite.internal.util.IgniteUtils#inBusyLock*`?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -3275,34 +3278,79 @@ public int tableId() {
private void cleanUpResourcesForDroppedTablesOnRecoveryBusy() {
// TODO: IGNITE-20384 Clean up abandoned resources for dropped zones
from vault and metastore
- for (DroppedTableInfo droppedTableInfo : droppedTables(catalogService,
lowWatermark.getLowWatermark())) {
- int catalogVersion = droppedTableInfo.tableRemovalCatalogVersion()
- 1;
- Catalog catalog = catalogService.catalog(catalogVersion);
+ Set<Integer> aliveTableIds = aliveTables(catalogService,
lowWatermark.getLowWatermark());
- CatalogTableDescriptor tableDescriptor =
catalog.table(droppedTableInfo.tableId());
- assert tableDescriptor != null : "tableId=" +
droppedTableInfo.tableId() + ", catalogVersion=" + catalogVersion;
+ destroyMvStoragesForTablesNotIn(aliveTableIds);
- CatalogZoneDescriptor zoneDescriptor =
catalog.zone(tableDescriptor.zoneId());
- assert zoneDescriptor != null : "zoneId=" +
tableDescriptor.zoneId() + ", catalogVersion=" + catalogVersion;
+ if (!nodeProperties.colocationEnabled()) {
+ destroyTxStateStoragesForTablesNotIn(aliveTableIds);
+ destroyReplicationProtocolStoragesForTablesNotIn(aliveTableIds);
+ }
+ }
- destroyTableOnRecoveryBusy(tableDescriptor,
zoneDescriptor.partitions());
+ private void destroyMvStoragesForTablesNotIn(Set<Integer> aliveTableIds) {
+ for (StorageEngine storageEngine : dataStorageMgr.allStorageEngines())
{
+ Set<Integer> tableIdsOnDisk = storageEngine.tableIdsOnDisk();
+
+ for (int tableId : difference(tableIdsOnDisk, aliveTableIds)) {
+ storageEngine.destroyMvTable(tableId);
+ LOG.info("Destroyed table MV storage for table {} in storage
engine '{}'", tableId, storageEngine.name());
Review Comment:
Maybe we should just output a single message saying that we've started
destroying these tables? We have quite a few entries in the log, what do you
think?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -3275,34 +3278,79 @@ public int tableId() {
private void cleanUpResourcesForDroppedTablesOnRecoveryBusy() {
// TODO: IGNITE-20384 Clean up abandoned resources for dropped zones
from vault and metastore
- for (DroppedTableInfo droppedTableInfo : droppedTables(catalogService,
lowWatermark.getLowWatermark())) {
- int catalogVersion = droppedTableInfo.tableRemovalCatalogVersion()
- 1;
- Catalog catalog = catalogService.catalog(catalogVersion);
+ Set<Integer> aliveTableIds = aliveTables(catalogService,
lowWatermark.getLowWatermark());
- CatalogTableDescriptor tableDescriptor =
catalog.table(droppedTableInfo.tableId());
- assert tableDescriptor != null : "tableId=" +
droppedTableInfo.tableId() + ", catalogVersion=" + catalogVersion;
+ destroyMvStoragesForTablesNotIn(aliveTableIds);
- CatalogZoneDescriptor zoneDescriptor =
catalog.zone(tableDescriptor.zoneId());
- assert zoneDescriptor != null : "zoneId=" +
tableDescriptor.zoneId() + ", catalogVersion=" + catalogVersion;
+ if (!nodeProperties.colocationEnabled()) {
+ destroyTxStateStoragesForTablesNotIn(aliveTableIds);
+ destroyReplicationProtocolStoragesForTablesNotIn(aliveTableIds);
+ }
+ }
- destroyTableOnRecoveryBusy(tableDescriptor,
zoneDescriptor.partitions());
+ private void destroyMvStoragesForTablesNotIn(Set<Integer> aliveTableIds) {
+ for (StorageEngine storageEngine : dataStorageMgr.allStorageEngines())
{
+ Set<Integer> tableIdsOnDisk = storageEngine.tableIdsOnDisk();
+
+ for (int tableId : difference(tableIdsOnDisk, aliveTableIds)) {
+ storageEngine.destroyMvTable(tableId);
+ LOG.info("Destroyed table MV storage for table {} in storage
engine '{}'", tableId, storageEngine.name());
+ }
}
}
- private void destroyTableOnRecoveryBusy(CatalogTableDescriptor
tableDescriptor, int partitionCount) {
- StorageEngine engine =
dataStorageMgr.engineByStorageProfile(tableDescriptor.storageProfile());
- assert engine != null : "tableId=" + tableDescriptor.id() + ",
storageProfile=" + tableDescriptor.storageProfile();
+ private void destroyTxStateStoragesForTablesNotIn(Set<Integer>
aliveTableIds) {
+ Set<Integer> tableIdsOnDisk =
sharedTxStateStorage.tableOrZoneIdsOnDisk();
- engine.destroyMvTable(tableDescriptor.id());
+ for (int tableId : difference(tableIdsOnDisk, aliveTableIds)) {
+ sharedTxStateStorage.destroyStorage(tableId);
+ LOG.info("Destroyed table TX state storage for table {}", tableId);
Review Comment:
Same about log message.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorage.java:
##########
@@ -347,4 +349,20 @@ public Set<Integer> tableOrZoneIdsOnDisk() {
return unmodifiableSet(ids);
}
+
+ /**
+ * Flushes the whole storage to disk.
+ */
+ @TestOnly
+ public void flush() {
+ try {
+ awaitFlush(true).get();
Review Comment:
It looks dangerous, let's pass the timeout argument in milliseconds and use,
say, a minute in tests.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1217,16 +1218,54 @@ public Set<ReplicationGroupId> startedGroups() {
/**
* Destroys replication protocol storages for the given group ID.
*
+ * <p>No durability guarantees are provided. If a node crashes, the
storage may come to life.
+ *
+ * @param replicaGrpId Replication group ID.
+ * @throws NodeStoppingException If the node is being stopped.
+ */
+ public void destroyReplicationProtocolStoragesOnStartup(ReplicationGroupId
replicaGrpId)
Review Comment:
Same about `throws NodeStoppingException`.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableUtils.java:
##########
@@ -82,38 +81,28 @@ public static List<Integer>
indexIdsAtRwTxBeginTs(CatalogService catalogService,
}
/**
- * Collects a list of tables that were removed from the catalog and should
have been dropped due to a low watermark (if the catalog
- * version in which the table was removed is less than or equal to the
active catalog version of the low watermark).
+ * Collects all IDs of tables that were not dropped or were dropped, but
should not have been destroyed yet due to a low watermark
+ * (if the catalog version in which the table was removed were less than
or equal to the active catalog version of the low watermark).
*
* @param catalogService Catalog service.
* @param lowWatermark Low watermark, {@code null} if it has never been
updated.
*/
- static List<DroppedTableInfo> droppedTables(CatalogService catalogService,
@Nullable HybridTimestamp lowWatermark) {
- if (lowWatermark == null) {
- return List.of();
+ static Set<Integer> aliveTables(CatalogService catalogService, @Nullable
HybridTimestamp lowWatermark) {
Review Comment:
Will this method be called before watches are deployed?
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
