You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2024/02/28 12:39:40 UTC
(ignite-3) 01/01: wip
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-21609
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 0f11d149b724ddb22889975de2f29fd01c7c046f
Author: amashenkov <an...@gmail.com>
AuthorDate: Wed Feb 28 15:38:34 2024 +0300
wip
---
.../internal/table/distributed/TableManager.java | 119 +++++++++------------
1 file changed, 49 insertions(+), 70 deletions(-)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 8af5a6dd37..07be1795c6 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.table.distributed;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static java.util.Collections.emptyMap;
import static java.util.Collections.emptySet;
import static java.util.Collections.unmodifiableMap;
import static java.util.concurrent.CompletableFuture.allOf;
@@ -51,6 +50,7 @@ import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+import static org.apache.ignite.lang.ErrorGroups.Client.TABLE_ID_NOT_FOUND_ERR;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
@@ -64,6 +64,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
@@ -200,6 +201,7 @@ import org.apache.ignite.internal.utils.RebalanceUtilEx;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.worker.ThreadAssertions;
import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.lang.TableNotFoundException;
import org.apache.ignite.lang.util.IgniteNameUtils;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.TopologyService;
@@ -256,12 +258,12 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
* @see #localPartsByTableIdVv
* @see #assignmentsUpdatedVv
*/
- private final IncrementalVersionedValue<Map<Integer, TableImpl>> tablesByIdVv;
+ private final IncrementalVersionedValue<Void> tablesVv;
/**
* Versioned store for local partition set by table id.
*
- * <p>Completed strictly after {@link #tablesByIdVv} and strictly before {@link #assignmentsUpdatedVv}.
+ * <p>Completed strictly after {@link #tablesVv} and strictly before {@link #assignmentsUpdatedVv}.
*/
private final IncrementalVersionedValue<Map<Integer, PartitionSet>> localPartsByTableIdVv;
@@ -274,11 +276,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*/
private final IncrementalVersionedValue<Void> assignmentsUpdatedVv;
- /**
- * {@link TableImpl} is created during update of tablesByIdVv, we store reference to it in case of updating of tablesByIdVv fails, so we
- * can stop resources associated with the table or to clean up table resources on {@code TableManager#stop()}.
- */
- private final Map<Integer, TableImpl> pendingTables = new ConcurrentHashMap<>();
+ /** Registered tables. */
+ private final Map<Integer, TableImpl> tables = new ConcurrentHashMap<>();
/** Started tables. */
private final Map<Integer, TableImpl> startedTables = new ConcurrentHashMap<>();
@@ -479,9 +478,9 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
schemaVersions = new SchemaVersionsImpl(executorInclinedSchemaSyncService, catalogService, clock);
- tablesByIdVv = new IncrementalVersionedValue<>(registry, HashMap::new);
+ tablesVv = new IncrementalVersionedValue<>(registry);
- localPartsByTableIdVv = new IncrementalVersionedValue<>(dependingOn(tablesByIdVv), HashMap::new);
+ localPartsByTableIdVv = new IncrementalVersionedValue<>(dependingOn(tablesVv), HashMap::new);
assignmentsUpdatedVv = new IncrementalVersionedValue<>(dependingOn(localPartsByTableIdVv));
@@ -723,19 +722,19 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
}
private CompletableFuture<?> onTableRename(RenameTableEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> tablesByIdVv.update(
+ return inBusyLockAsync(busyLock, () -> tablesVv.update(
parameters.causalityToken(),
- (tablesById, e) -> {
+ (ignore, e) -> {
if (e != null) {
return failedFuture(e);
}
- TableImpl table = tablesById.get(parameters.tableId());
+ TableImpl table = tables.get(parameters.tableId());
// TODO: revisit this approach, see https://issues.apache.org/jira/browse/IGNITE-21235.
table.name(parameters.newTableName());
- return completedFuture(tablesById);
+ return nullCompletedFuture();
})
);
}
@@ -1063,12 +1062,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
metaStorageMgr.unregisterWatch(stableAssignmentsRebalanceListener);
metaStorageMgr.unregisterWatch(assignmentsSwitchRebalanceListener);
- var tablesToStop = new HashMap<Integer, TableImpl>();
-
- tablesToStop.putAll(latestTablesById());
- tablesToStop.putAll(pendingTables);
-
- cleanUpTablesResources(tablesToStop);
+ cleanUpTablesResources(tables);
}
@Override
@@ -1239,20 +1233,16 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
var table = new TableImpl(internalTable, lockMgr, schemaVersions, marshallers, sql.get(), tableDescriptor.primaryKeyIndexId());
- tablesByIdVv.update(causalityToken, (previous, e) -> inBusyLock(busyLock, () -> {
+ tablesVv.update(causalityToken, (ignore, e) -> inBusyLock(busyLock, () -> {
if (e != null) {
return failedFuture(e);
}
return schemaManager.schemaRegistry(causalityToken, tableId)
- .thenApply(schema -> {
+ .thenAccept(schema -> {
table.schemaView(schema);
- var val = new HashMap<>(previous);
-
- val.put(tableId, table);
-
- return val;
+ startedTables.put(tableId, table);
});
}));
@@ -1276,8 +1266,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
});
}, ioExecutor)));
- // We bring the future outside to avoid OutdatedTokenException.
- CompletableFuture<Map<Integer, TableImpl>> tablesByIdFuture = tablesByIdVv.get(causalityToken);
+ CompletableFuture<?> tablesByIdFuture = tablesVv.get(causalityToken);
// TODO https://issues.apache.org/jira/browse/IGNITE-19170 Partitions should be started only on the assignments change
// TODO event triggered by zone create or alter.
@@ -1286,20 +1275,14 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return failedFuture(e);
}
- return localPartsUpdateFuture.thenCompose(unused ->
- tablesByIdFuture.thenComposeAsync(tablesById -> inBusyLock(
+ return allOf(localPartsUpdateFuture, tablesByIdFuture)
+ .thenComposeAsync(ignore -> inBusyLock(
busyLock,
() -> startLocalPartitionsAndClients(assignmentsFuture, table)
- ), ioExecutor)
- );
+ ), ioExecutor);
});
- pendingTables.put(tableId, table);
- startedTables.put(tableId, table);
-
- tablesById(causalityToken).thenAccept(ignored -> inBusyLock(busyLock, () -> {
- pendingTables.remove(tableId);
- }));
+ tables.put(tableId, table);
// TODO should be reworked in IGNITE-16763
@@ -1371,7 +1354,12 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*/
private void dropTableLocally(long causalityToken, DestroyTableEventParameters parameters) {
int tableId = parameters.tableId();
- int partitions = parameters.partitions();
+
+ //TODO Drop partitions from parameters and use from storage.
+ int partitions; // = parameters.partitions();
+ TableImpl table0 = tables.get(tableId);
+ assert table0 != null : tableId;
+ partitions = table0.internalTable().partitions();
localPartsByTableIdVv.update(causalityToken, (previousVal, e) -> inBusyLock(busyLock, () -> {
if (e != null) {
@@ -1384,14 +1372,12 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return completedFuture(newMap);
}));
- tablesByIdVv.update(causalityToken, (previousVal, e) -> inBusyLock(busyLock, () -> {
+ tablesVv.update(causalityToken, (ignore, e) -> inBusyLock(busyLock, () -> {
if (e != null) {
return failedFuture(e);
}
- var map = new HashMap<>(previousVal);
-
- TableImpl table = map.remove(tableId);
+ TableImpl table = tables.get(tableId);
assert table != null : tableId;
@@ -1400,6 +1386,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
CompletableFuture<?>[] stopReplicaFutures = new CompletableFuture<?>[partitions];
// TODO https://issues.apache.org/jira/browse/IGNITE-19170 Partitions should be stopped on the assignments change
+ // TODO stop replica asynchronously, out of metastorage event pipeline.
// TODO event triggered by zone drop or alter.
for (int partitionId = 0; partitionId < partitions; partitionId++) {
var replicationGroupId = new TablePartitionId(tableId, partitionId);
@@ -1415,7 +1402,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
runAsync(() -> internalTable.txStateStorage().destroy(), ioExecutor)
),
ioExecutor
- ).thenApply(v -> map);
+ ).thenAccept(ignore0 -> tables.remove(tableId));
}));
startedTables.remove(tableId);
@@ -1488,27 +1475,14 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
* @see #assignmentsUpdatedVv
*/
private CompletableFuture<Map<Integer, TableImpl>> tablesById(long causalityToken) {
- // We bring the future outside to avoid OutdatedTokenException.
- CompletableFuture<Map<Integer, TableImpl>> tablesByIdFuture = tablesByIdVv.get(causalityToken);
-
- return assignmentsUpdatedVv.get(causalityToken).thenCompose(v -> tablesByIdFuture);
+ return assignmentsUpdatedVv.get(causalityToken).thenApply(v -> unmodifiableMap(tables));
}
/**
* Returns the latest tables by ID map, for which all assignment updates have been completed.
*/
private Map<Integer, TableImpl> latestTablesById() {
- // TODO https://issues.apache.org/jira/browse/IGNITE-20915 fix this.
- if (assignmentsUpdatedVv.latestCausalityToken() < 0L) {
- // No tables at all in case of empty causality token.
- return emptyMap();
- } else {
- CompletableFuture<Map<Integer, TableImpl>> tablesByIdFuture = tablesByIdVv.get(tablesByIdVv.latestCausalityToken());
-
- assert tablesByIdFuture.isDone() : "'tablesByIdVv' is always completed strictly before the 'assignmentsUpdatedVv'";
-
- return tablesByIdFuture.join();
- }
+ return unmodifiableMap(tables);
}
/**
@@ -1518,7 +1492,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*/
@TestOnly
public Map<Integer, TableImpl> latestTables() {
- return unmodifiableMap(latestTablesById());
+ return latestTablesById();
}
@Override
@@ -1627,7 +1601,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
}
private CompletableFuture<TableViewInternal> tableAsyncInternalBusy(int tableId) {
- TableImpl tableImpl = latestTablesById().get(tableId);
+ TableImpl tableImpl = startedTables.get(tableId);
if (tableImpl != null) {
return completedFuture(tableImpl);
@@ -1637,16 +1611,21 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
CompletionListener<Void> tablesListener = (token, v, th) -> {
if (th == null) {
- CompletableFuture<Map<Integer, TableImpl>> tablesFuture = tablesByIdVv.get(token);
+ CompletableFuture<?> tablesFuture = tablesVv.get(token);
tablesFuture.whenComplete((tables, e) -> {
if (e != null) {
getLatestTableFuture.completeExceptionally(e);
} else {
- TableImpl table = tables.get(tableId);
+ TableImpl table = startedTables.get(tableId);
if (table != null) {
getLatestTableFuture.complete(table);
+ } else {
+ var err = new TableNotFoundException(UUID.randomUUID(), TABLE_ID_NOT_FOUND_ERR,
+ "Table does not exist or was dropped concurrently: " + tableId, null);
+
+ getLatestTableFuture.completeExceptionally(err);
}
}
});
@@ -1747,8 +1726,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
Assignments pendingAssignments = Assignments.fromBytes(pendingAssignmentsEntry.value());
- return tablesByIdVv.get(revision)
- .thenApply(tables -> {
+ return tablesVv.get(revision)
+ .thenApply(ignore -> {
if (!busyLock.enterBusy()) {
return CompletableFuture.<Void>failedFuture(new NodeStoppingException());
}
@@ -2199,8 +2178,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
}
private CompletableFuture<Void> stopAndDestroyPartition(TablePartitionId tablePartitionId, long causalityToken) {
- return tablesByIdVv.get(causalityToken)
- .thenCompose(tables -> {
+ return tablesVv.get(causalityToken)
+ .thenCompose(ignore -> {
TableImpl table = tables.get(tablePartitionId.tableId());
return stopPartition(tablePartitionId, table)
@@ -2310,7 +2289,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*/
@Override
public @Nullable TableViewInternal cachedTable(int tableId) {
- return startedTables.get(tableId);
+ return tables.get(tableId);
}
/**
@@ -2320,7 +2299,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*/
@TestOnly
public @Nullable TableViewInternal cachedTable(String name) {
- return findTableImplByName(startedTables.values(), name);
+ return findTableImplByName(tables.values(), name);
}
private CatalogTableDescriptor getTableDescriptor(int tableId, int catalogVersion) {