You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2024/06/14 07:35:17 UTC
(ignite-3) branch main updated: IGNITE-22470 Switch IndexMetaStorage from Vault to MetaStorage (#3917)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new d132a15afe IGNITE-22470 Switch IndexMetaStorage from Vault to MetaStorage (#3917)
d132a15afe is described below
commit d132a15afec48769c9016d5849c4435be715e728
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Fri Jun 14 10:35:11 2024 +0300
IGNITE-22470 Switch IndexMetaStorage from Vault to MetaStorage (#3917)
---
.../ignite/internal/event/EventListener.java | 15 ++
.../table/distributed/index/IndexMeta.java | 54 +++--
.../table/distributed/index/IndexMetaStorage.java | 237 +++++++++++++--------
.../index/BaseIndexMetaStorageTest.java | 65 ++++--
.../index/IndexMetaStorageRecoveryTest.java | 104 +++++----
.../distributed/index/IndexMetaStorageTest.java | 112 ++++++----
6 files changed, 376 insertions(+), 211 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java b/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
index 496ce91a81..4b7575ce04 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
@@ -22,6 +22,7 @@ import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedF
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
+import java.util.function.Function;
/** A listener that handles events from an event producer. */
@FunctionalInterface
@@ -52,4 +53,18 @@ public interface EventListener<P extends EventParameters> {
return falseCompletedFuture();
};
}
+
+ /**
+ * Creates an adapter using the passed function; if an error occurs when calling the function, the failed future will be returned with
+ * this error.
+ */
+ static <P extends EventParameters> EventListener<P> fromFunction(Function<P, CompletableFuture<Boolean>> callback) {
+ return parameters -> {
+ try {
+ return callback.apply(parameters);
+ } catch (Throwable e) {
+ return failedFuture(e);
+ }
+ };
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMeta.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMeta.java
index ca4c7c8a20..a45395571e 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMeta.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMeta.java
@@ -31,6 +31,8 @@ import org.apache.ignite.internal.tostring.S;
public class IndexMeta implements Serializable {
private static final long serialVersionUID = 1044129530453957897L;
+ private final int catalogVersion;
+
private final int indexId;
private final int tableId;
@@ -42,23 +44,10 @@ public class IndexMeta implements Serializable {
@IgniteToStringInclude
private final Map<MetaIndexStatus, MetaIndexStatusChange> statusChanges;
- /** Constructor. */
- IndexMeta(CatalogIndexDescriptor catalogIndexDescriptor, Catalog catalog) {
- this(
- catalogIndexDescriptor.id(),
- catalogIndexDescriptor.tableId(),
- catalogIndexDescriptor.name(),
- MetaIndexStatus.convert(catalogIndexDescriptor.status()),
- Map.of(
- MetaIndexStatus.convert(catalogIndexDescriptor.status()),
- new MetaIndexStatusChange(catalog.version(), catalog.time())
- )
- );
- }
-
/**
* Constructor.
*
+ * @param catalogVersion Catalog version in which the current meta was created.
* @param indexId Index ID.
* @param tableId Table ID to which the index belongs.
* @param indexName Index name.
@@ -66,12 +55,14 @@ public class IndexMeta implements Serializable {
* @param statusChanges <b>Immutable</b> map of index statuses with change info (for example catalog version) in which they appeared.
*/
private IndexMeta(
+ int catalogVersion,
int indexId,
int tableId,
String indexName,
MetaIndexStatus currentStatus,
Map<MetaIndexStatus, MetaIndexStatusChange> statusChanges
) {
+ this.catalogVersion = catalogVersion;
this.indexId = indexId;
this.tableId = tableId;
this.indexName = indexName;
@@ -79,6 +70,34 @@ public class IndexMeta implements Serializable {
this.statusChanges = unmodifiableMap(statusChanges);
}
+ /**
+ * Creates a index meta instance.
+ *
+ * @param catalogIndexDescriptor Catalog index descriptor to create meta from.
+ * @param catalog Catalog version from which the {@code catalogIndexDescriptor} was taken.
+ */
+ static IndexMeta of(CatalogIndexDescriptor catalogIndexDescriptor, Catalog catalog) {
+ assert catalog.index(catalogIndexDescriptor.id()) != null :
+ "indexId=" + catalogIndexDescriptor.id() + ", catalogVersion=" + catalog.version();
+
+ return new IndexMeta(
+ catalog.version(),
+ catalogIndexDescriptor.id(),
+ catalogIndexDescriptor.tableId(),
+ catalogIndexDescriptor.name(),
+ MetaIndexStatus.convert(catalogIndexDescriptor.status()),
+ Map.of(
+ MetaIndexStatus.convert(catalogIndexDescriptor.status()),
+ new MetaIndexStatusChange(catalog.version(), catalog.time())
+ )
+ );
+ }
+
+ /** Returns catalog version in which the current meta was created. */
+ int catalogVersion() {
+ return catalogVersion;
+ }
+
/** Returns index ID. */
public int indexId() {
return indexId;
@@ -97,11 +116,12 @@ public class IndexMeta implements Serializable {
/**
* Changes the index name.
*
+ * @param catalogVersion Catalog version in which the index name has changed.
* @param newIndexName New index name.
* @return New instance of the index meta with only a new index name.
*/
- IndexMeta indexName(String newIndexName) {
- return new IndexMeta(indexId, tableId, newIndexName, currentStatus, new EnumMap<>(statusChanges));
+ IndexMeta indexName(int catalogVersion, String newIndexName) {
+ return new IndexMeta(catalogVersion, indexId, tableId, newIndexName, currentStatus, new EnumMap<>(statusChanges));
}
/** Returns the current status of the index. */
@@ -124,7 +144,7 @@ public class IndexMeta implements Serializable {
var newStatuses = new EnumMap<>(statusChanges);
newStatuses.put(newStatus, new MetaIndexStatusChange(catalogVersion, activationTs));
- return new IndexMeta(indexId, tableId, indexName, newStatus, newStatuses);
+ return new IndexMeta(catalogVersion, indexId, tableId, indexName, newStatus, newStatuses);
}
/** Returns a map of index statuses with change info (for example catalog version) in which they appeared. */
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorage.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorage.java
index 0025fc8a7f..dc00d5c9ad 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorage.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorage.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.table.distributed.index;
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.stream.Collectors.toMap;
import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_AVAILABLE;
import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_BUILDING;
@@ -24,19 +26,28 @@ import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_CREAT
import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_REMOVED;
import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_STOPPING;
import static org.apache.ignite.internal.catalog.events.CatalogEvent.TABLE_ALTER;
-import static org.apache.ignite.internal.event.EventListener.fromConsumer;
+import static org.apache.ignite.internal.event.EventListener.fromFunction;
import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestampToLong;
import static org.apache.ignite.internal.lowwatermark.event.LowWatermarkEvent.LOW_WATERMARK_CHANGED;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.exists;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.value;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.noop;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.READ_ONLY;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.REMOVED;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.statusOnRemoveIndex;
+import static org.apache.ignite.internal.util.ByteUtils.intToBytes;
+import static org.apache.ignite.internal.util.ByteUtils.toBytes;
+import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.apache.ignite.internal.util.StringUtils.incrementLastChar;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
@@ -60,16 +71,16 @@ import org.apache.ignite.internal.lowwatermark.event.ChangeLowWatermarkEventPara
import org.apache.ignite.internal.lowwatermark.event.LowWatermarkEvent;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.vault.VaultEntry;
-import org.apache.ignite.internal.vault.VaultManager;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
/**
- * Local storage of {@link IndexMeta index metadata}, based on a {@link CatalogIndexDescriptor} and stored in vault, is responsible for
- * their life cycle, as well as recovery when the node is restarted. Any change to index metadata is saved in vault. The main difference
+ * Local storage of {@link IndexMeta index metadata}, based on a {@link CatalogIndexDescriptor} and stored in metastore, is responsible for
+ * their life cycle, as well as recovery when the node is restarted. Any change to index metadata is saved in metastore. The main difference
* from the {@link CatalogIndexDescriptor} is that index metadata is independent of catalog compaction and stores all the necessary
* information for the user when working with indexes.
*
@@ -94,13 +105,15 @@ import org.jetbrains.annotations.TestOnly;
* </ul>
*/
public class IndexMetaStorage implements IgniteComponent {
- private static final String INDEX_META_KEY_PREFIX = "index.meta.";
+ private static final String INDEX_META_VERSION_KEY_PREFIX = "index.meta.version.";
+
+ private static final String INDEX_META_VALUE_KEY_PREFIX = "index.meta.value.";
private final CatalogService catalogService;
private final LowWatermark lowWatermark;
- private final VaultManager vaultManager;
+ private final MetaStorageManager metaStorageManager;
private final Map<Integer, IndexMeta> indexMetaByIndexId = new ConcurrentHashMap<>();
@@ -108,28 +121,30 @@ public class IndexMetaStorage implements IgniteComponent {
public IndexMetaStorage(
CatalogService catalogService,
LowWatermark lowWatermark,
- VaultManager vaultManager
+ MetaStorageManager metaStorageManager
) {
this.catalogService = catalogService;
this.lowWatermark = lowWatermark;
- this.vaultManager = vaultManager;
+ this.metaStorageManager = metaStorageManager;
}
@Override
public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
- catalogService.listen(INDEX_CREATE, fromConsumer(this::onCatalogIndexCreateEvent));
- catalogService.listen(INDEX_REMOVED, fromConsumer(this::onCatalogIndexRemovedEvent));
- catalogService.listen(INDEX_BUILDING, fromConsumer(this::onCatalogIndexBuildingEvent));
- catalogService.listen(INDEX_AVAILABLE, fromConsumer(this::onCatalogIndexAvailableEvent));
- catalogService.listen(INDEX_STOPPING, fromConsumer(this::onCatalogIndexStoppingEvent));
-
- catalogService.listen(TABLE_ALTER, fromConsumer(this::onCatalogTableAlterEvent));
+ try {
+ catalogService.listen(INDEX_CREATE, fromFunction(this::onCatalogIndexCreateEvent));
+ catalogService.listen(INDEX_REMOVED, fromFunction(this::onCatalogIndexRemovedEvent));
+ catalogService.listen(INDEX_BUILDING, fromFunction(this::onCatalogIndexBuildingEvent));
+ catalogService.listen(INDEX_AVAILABLE, fromFunction(this::onCatalogIndexAvailableEvent));
+ catalogService.listen(INDEX_STOPPING, fromFunction(this::onCatalogIndexStoppingEvent));
- lowWatermark.listen(LOW_WATERMARK_CHANGED, fromConsumer(this::onLwmChanged));
+ catalogService.listen(TABLE_ALTER, fromFunction(this::onCatalogTableAlterEvent));
- recoverIndexMetas();
+ lowWatermark.listen(LOW_WATERMARK_CHANGED, fromFunction(this::onLwmChanged));
- return nullCompletedFuture();
+ return recoverIndexMetas();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ }
}
@Override
@@ -154,25 +169,27 @@ public class IndexMetaStorage implements IgniteComponent {
return List.copyOf(indexMetaByIndexId.values());
}
- private void onCatalogIndexCreateEvent(CreateIndexEventParameters parameters) {
+ private CompletableFuture<Boolean> onCatalogIndexCreateEvent(CreateIndexEventParameters parameters) {
CatalogIndexDescriptor catalogIndexDescriptor = parameters.indexDescriptor();
Catalog catalog = catalog(parameters.catalogVersion());
- updateAndSaveIndexMetaToVault(catalogIndexDescriptor.id(), indexMeta -> {
- assert indexMeta == null : catalogIndexDescriptor.id();
+ return updateAndSaveIndexMetaToMetastore(catalogIndexDescriptor.id(), indexMeta -> {
+ assert indexMeta == null : "indexId=" + catalogIndexDescriptor.id() + "catalogVersion=" + catalog.version();
- return new IndexMeta(catalogIndexDescriptor, catalog);
- });
+ return IndexMeta.of(catalogIndexDescriptor, catalog);
+ }).thenApply(unused -> false);
}
- private void onCatalogIndexRemovedEvent(RemoveIndexEventParameters parameters) {
+ private CompletableFuture<Boolean> onCatalogIndexRemovedEvent(RemoveIndexEventParameters parameters) {
int indexId = parameters.indexId();
int eventCatalogVersion = parameters.catalogVersion();
Catalog catalog = catalog(eventCatalogVersion);
+ CompletableFuture<?>[] resultFuture = new CompletableFuture<?>[1];
+
lowWatermark.getLowWatermarkSafe(lwm -> {
int lwmCatalogVersion = catalogService.activeCatalogVersion(hybridTimestampToLong(lwm));
@@ -180,24 +197,28 @@ public class IndexMetaStorage implements IgniteComponent {
// There is no need to add a read-only index, since the index should be destroyed under the updated low watermark.
IndexMeta removed = indexMetaByIndexId.remove(indexId);
- assert removed != null : indexId;
+ assert removed != null : "indexId=" + indexId + ", catalogVersion=" + catalog.version();
- removeFromVault(removed);
+ resultFuture[0] = removeFromMetastore(removed);
} else {
- updateAndSaveIndexMetaToVault(indexId, indexMeta -> {
- assert indexMeta != null : indexId;
+ resultFuture[0] = updateAndSaveIndexMetaToMetastore(indexId, indexMeta -> {
+ assert indexMeta != null : "indexId=" + indexId + ", catalogVersion=" + catalog.version();
return setNewStatus(indexMeta, statusOnRemoveIndex(indexMeta.status()), catalog);
});
}
});
+
+ assert resultFuture[0] != null : "indexId=" + indexId + ", catalogVersion=" + catalog.version();
+
+ return resultFuture[0].thenApply(unused -> false);
}
- private void onCatalogIndexBuildingEvent(StartBuildingIndexEventParameters parameters) {
- updateIndexStatus(parameters);
+ private CompletableFuture<Boolean> onCatalogIndexBuildingEvent(StartBuildingIndexEventParameters parameters) {
+ return updateIndexStatus(parameters);
}
- private void onCatalogIndexAvailableEvent(MakeIndexAvailableEventParameters parameters) {
+ private CompletableFuture<Boolean> onCatalogIndexAvailableEvent(MakeIndexAvailableEventParameters parameters) {
int catalogVersion = parameters.catalogVersion();
int indexId = parameters.indexId();
@@ -205,27 +226,29 @@ public class IndexMetaStorage implements IgniteComponent {
CatalogTableDescriptor catalogTableDescriptor = catalogTableDescriptor(catalogIndexDescriptor.tableId(), catalogVersion);
if (indexId != catalogTableDescriptor.primaryKeyIndexId()) {
- updateIndexStatus(parameters);
+ return updateIndexStatus(parameters);
}
+
+ return falseCompletedFuture();
}
- private void onCatalogIndexStoppingEvent(StoppingIndexEventParameters parameters) {
- updateIndexStatus(parameters);
+ private CompletableFuture<Boolean> onCatalogIndexStoppingEvent(StoppingIndexEventParameters parameters) {
+ return updateIndexStatus(parameters);
}
- private void updateIndexStatus(IndexEventParameters parameters) {
+ private CompletableFuture<Boolean> updateIndexStatus(IndexEventParameters parameters) {
CatalogIndexDescriptor catalogIndexDescriptor = catalogIndexDescriptor(parameters.indexId(), parameters.catalogVersion());
Catalog catalog = catalog(parameters.catalogVersion());
- updateAndSaveIndexMetaToVault(catalogIndexDescriptor.id(), indexMeta -> {
- assert indexMeta != null : catalogIndexDescriptor.id();
+ return updateAndSaveIndexMetaToMetastore(catalogIndexDescriptor.id(), indexMeta -> {
+ assert indexMeta != null : "indexId=" + catalogIndexDescriptor.id() + ", catalogVersion=" + catalog.version();
return setNewStatus(indexMeta, MetaIndexStatus.convert(catalogIndexDescriptor.status()), catalog);
- });
+ }).thenApply(unused -> false);
}
- private void onCatalogTableAlterEvent(TableEventParameters parameters) {
+ private CompletableFuture<Boolean> onCatalogTableAlterEvent(TableEventParameters parameters) {
if (parameters instanceof RenameTableEventParameters) {
int catalogVersion = parameters.catalogVersion();
@@ -235,63 +258,74 @@ public class IndexMetaStorage implements IgniteComponent {
CatalogIndexDescriptor catalogIndexDescriptor = catalogIndexDescriptor(indexId, catalogVersion);
- updateAndSaveIndexMetaToVault(indexId, indexMeta -> {
- assert indexMeta != null : indexId;
+ return updateAndSaveIndexMetaToMetastore(indexId, indexMeta -> {
+ assert indexMeta != null : "indexId=" + indexId + ", catalogVersion=" + catalogVersion;
- return indexMeta.indexName(catalogIndexDescriptor.name());
- });
+ return indexMeta.indexName(parameters.catalogVersion(), catalogIndexDescriptor.name());
+ }).thenApply(unused -> false);
}
+
+ return falseCompletedFuture();
}
- private void onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ private CompletableFuture<Boolean> onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
int lwmCatalogVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
Iterator<IndexMeta> it = indexMetaByIndexId.values().iterator();
+ var futures = new ArrayList<CompletableFuture<?>>();
+
while (it.hasNext()) {
IndexMeta indexMeta = it.next();
if (shouldBeRemoved(indexMeta, lwmCatalogVersion)) {
it.remove();
- removeFromVault(indexMeta);
+ futures.add(removeFromMetastore(indexMeta));
}
}
+
+ return futures.isEmpty() ? falseCompletedFuture() : allOf(futures.toArray(CompletableFuture[]::new)).thenApply(unused -> false);
}
// TODO: IGNITE-22442 Improve recovery
- private void recoverIndexMetas() {
+ private CompletableFuture<Void> recoverIndexMetas() {
Catalog catalog = catalog(catalogService.latestCatalogVersion());
- Map<Integer, IndexMeta> fromVault = readAllFromVault();
- Map<Integer, CatalogIndexDescriptor> fromCatalogLatest = readAllFromCatalog(catalog.version());
+ Map<Integer, IndexMeta> fromMetastore = readAllFromMetastoreOnRecovery();
+ Map<Integer, CatalogIndexDescriptor> fromCatalogLatest = readAllFromCatalogOnRecovery(catalog.version());
+
+ var futures = new ArrayList<CompletableFuture<?>>();
for (CatalogIndexDescriptor catalogIndexDescriptor : fromCatalogLatest.values()) {
int indexId = catalogIndexDescriptor.id();
- IndexMeta indexMetaFromVault = fromVault.get(indexId);
+ IndexMeta indexMetaFromMetastore = fromMetastore.get(indexId);
- if (indexMetaFromVault == null) {
+ if (indexMetaFromMetastore == null) {
// We did not have time to save at the index creation event.
- updateAndSaveIndexMetaToVault(indexId, indexMeta -> new IndexMeta(catalogIndexDescriptor, catalog));
- } else if (!catalogIndexDescriptor.name().equals(indexMetaFromVault.indexName())) {
+ futures.add(updateAndSaveIndexMetaToMetastore(indexId, indexMeta -> IndexMeta.of(catalogIndexDescriptor, catalog)));
+ } else if (!catalogIndexDescriptor.name().equals(indexMetaFromMetastore.indexName())) {
// We did not have time to process the index renaming event.
- updateAndSaveIndexMetaToVault(indexId, indexMeta -> indexMetaFromVault.indexName(catalogIndexDescriptor.name()));
- } else if (MetaIndexStatus.convert(catalogIndexDescriptor.status()) != indexMetaFromVault.status()) {
+ futures.add(updateAndSaveIndexMetaToMetastore(
+ indexId,
+ indexMeta -> indexMetaFromMetastore.indexName(catalog.version(), catalogIndexDescriptor.name())
+ ));
+ } else if (MetaIndexStatus.convert(catalogIndexDescriptor.status()) != indexMetaFromMetastore.status()) {
// We did not have time to process the index status change event.
- updateAndSaveIndexMetaToVault(
+ futures.add(updateAndSaveIndexMetaToMetastore(
indexId,
- indexMeta -> setNewStatus(indexMetaFromVault, MetaIndexStatus.convert(catalogIndexDescriptor.status()), catalog)
- );
+ indexMeta -> setNewStatus(indexMetaFromMetastore, MetaIndexStatus.convert(catalogIndexDescriptor.status()), catalog)
+ ));
} else {
- indexMetaByIndexId.put(indexId, indexMetaFromVault);
+ indexMetaByIndexId.put(indexId, indexMetaFromMetastore);
}
}
// Let’s deal with read-only indexes.
int lwmCatalogVersion = catalogService.activeCatalogVersion(hybridTimestampToLong(lowWatermark.getLowWatermark()));
- for (IndexMeta indexMeta : fromVault.values()) {
+ for (IndexMeta indexMeta : fromMetastore.values()) {
int indexId = indexMeta.indexId();
if (indexMetaByIndexId.containsKey(indexId)) {
@@ -301,7 +335,7 @@ public class IndexMetaStorage implements IgniteComponent {
if (indexMeta.status() == READ_ONLY || indexMeta.status() == REMOVED) {
if (shouldBeRemoved(indexMeta, lwmCatalogVersion)) {
// We did not have time to process the lwm change event.
- removeFromVault(indexMeta);
+ futures.add(removeFromMetastore(indexMeta));
} else {
indexMetaByIndexId.put(indexId, indexMeta);
}
@@ -309,15 +343,17 @@ public class IndexMetaStorage implements IgniteComponent {
// We did not have time to process the index removal event.
if (catalog.version() <= lwmCatalogVersion) {
// There is no need to add a read-only indexes, since the index should be destroyed under the updated low watermark.
- removeFromVault(indexMeta);
+ futures.add(removeFromMetastore(indexMeta));
} else {
- updateAndSaveIndexMetaToVault(
+ futures.add(updateAndSaveIndexMetaToMetastore(
indexId,
meta -> setNewStatus(indexMeta, statusOnRemoveIndex(indexMeta.status()), catalog)
- );
+ ));
}
}
}
+
+ return futures.isEmpty() ? nullCompletedFuture() : allOf(futures.toArray(CompletableFuture[]::new));
}
private CatalogIndexDescriptor catalogIndexDescriptor(int indexId, int catalogVersion) {
@@ -344,32 +380,23 @@ public class IndexMetaStorage implements IgniteComponent {
return catalog;
}
- private void updateAndSaveIndexMetaToVault(int indexId, Function<@Nullable IndexMeta, IndexMeta> updateFunction) {
- IndexMeta newMeta = indexMetaByIndexId.compute(indexId, (id, indexMeta) -> updateFunction.apply(indexMeta));
-
- saveToVault(newMeta);
- }
-
- private void saveToVault(IndexMeta indexMeta) {
- vaultManager.put(indexMetaVaultKey(indexMeta), ByteUtils.toBytes(indexMeta));
- }
+ private Map<Integer, IndexMeta> readAllFromMetastoreOnRecovery() {
+ CompletableFuture<Long> recoveryFinishedFuture = metaStorageManager.recoveryFinishedFuture();
- private void removeFromVault(IndexMeta indexMeta) {
- vaultManager.remove(indexMetaVaultKey(indexMeta));
- }
+ assert recoveryFinishedFuture.isDone();
- private Map<Integer, IndexMeta> readAllFromVault() {
- ByteArray fromKey = ByteArray.fromString(INDEX_META_KEY_PREFIX);
- ByteArray toKey = ByteArray.fromString(incrementLastChar(INDEX_META_KEY_PREFIX));
+ long recoveryRevision = recoveryFinishedFuture.join();
- try (Cursor<VaultEntry> cursor = vaultManager.range(fromKey, toKey)) {
+ try (Cursor<Entry> cursor = metaStorageManager.prefixLocally(ByteArray.fromString(INDEX_META_VALUE_KEY_PREFIX), recoveryRevision)) {
return cursor.stream()
- .map(vaultEntry -> (IndexMeta) ByteUtils.fromBytes(vaultEntry.value()))
+ .map(Entry::value)
+ .filter(Objects::nonNull)
+ .map(entryBytes -> (IndexMeta) ByteUtils.fromBytes(entryBytes))
.collect(toMap(IndexMeta::indexId, Function.identity()));
}
}
- private Map<Integer, CatalogIndexDescriptor> readAllFromCatalog(int catalogVersion) {
+ private Map<Integer, CatalogIndexDescriptor> readAllFromCatalogOnRecovery(int catalogVersion) {
return catalogService.indexes(catalogVersion).stream()
.collect(toMap(CatalogObjectDescriptor::id, Function.identity()));
}
@@ -378,13 +405,49 @@ public class IndexMetaStorage implements IgniteComponent {
return indexMeta.status(newStatus, catalog.version(), catalog.time());
}
- private static ByteArray indexMetaVaultKey(IndexMeta indexMeta) {
- return ByteArray.fromString(INDEX_META_KEY_PREFIX + indexMeta.indexId());
- }
-
private static boolean shouldBeRemoved(IndexMeta indexMeta, int lwmCatalogVersion) {
MetaIndexStatus status = indexMeta.status();
return (status == READ_ONLY || status == REMOVED) && indexMeta.statusChanges().get(status).catalogVersion() <= lwmCatalogVersion;
}
+
+ private static ByteArray indexMetaVersionKey(IndexMeta indexMeta) {
+ return ByteArray.fromString(INDEX_META_VERSION_KEY_PREFIX + indexMeta.indexId());
+ }
+
+ private static ByteArray indexMetaValueKey(IndexMeta indexMeta) {
+ return ByteArray.fromString(INDEX_META_VALUE_KEY_PREFIX + indexMeta.indexId());
+ }
+
+ private CompletableFuture<?> saveToMetastore(IndexMeta newMeta) {
+ ByteArray versionKey = indexMetaVersionKey(newMeta);
+
+ return metaStorageManager.invoke(
+ value(versionKey).lt(intToBytes(newMeta.catalogVersion())),
+ List.of(
+ put(versionKey, intToBytes(newMeta.catalogVersion())),
+ put(indexMetaValueKey(newMeta), toBytes(newMeta))
+ ),
+ List.of(noop())
+ );
+ }
+
+ private CompletableFuture<?> removeFromMetastore(IndexMeta indexMeta) {
+ ByteArray versionKey = indexMetaVersionKey(indexMeta);
+
+ return metaStorageManager.invoke(
+ exists(versionKey),
+ List.of(remove(versionKey), remove(indexMetaValueKey(indexMeta))),
+ List.of(noop())
+ );
+ }
+
+ private CompletableFuture<?> updateAndSaveIndexMetaToMetastore(
+ int indexId,
+ Function<@Nullable IndexMeta, IndexMeta> updateFunction
+ ) {
+ IndexMeta newMeta = indexMetaByIndexId.compute(indexId, (id, indexMeta) -> updateFunction.apply(indexMeta));
+
+ return saveToMetastore(newMeta);
+ }
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/BaseIndexMetaStorageTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/BaseIndexMetaStorageTest.java
index 4f95f9bf2f..66341ae565 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/BaseIndexMetaStorageTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/BaseIndexMetaStorageTest.java
@@ -29,9 +29,11 @@ import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.hlc.HybridClock;
@@ -40,13 +42,11 @@ import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.lowwatermark.TestLowWatermark;
import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.util.IgniteUtils;
-import org.apache.ignite.internal.vault.VaultEntry;
-import org.apache.ignite.internal.vault.VaultManager;
-import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -63,9 +63,7 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
final HybridClock clock = new HybridClockImpl();
- final TestLowWatermark lowWatermark = new TestLowWatermark();
-
- final VaultManager vaultManager = new VaultManager(new InMemoryVaultService());
+ private final TestLowWatermark lowWatermark = new TestLowWatermark();
MetaStorageManager metastore;
@@ -83,7 +81,7 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
var componentContext = new ComponentContext();
- assertThat(startAsync(componentContext, vaultManager, metastore, catalogManager, indexMetaStorage), willCompleteSuccessfully());
+ assertThat(startAsync(componentContext, metastore, catalogManager, indexMetaStorage), willCompleteSuccessfully());
assertThat(metastore.deployWatches(), willCompleteSuccessfully());
@@ -100,9 +98,8 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
indexMetaStorage == null ? null : indexMetaStorage::beforeNodeStop,
catalogManager == null ? null : catalogManager::beforeNodeStop,
metastore == null ? null : metastore::beforeNodeStop,
- vaultManager::beforeNodeStop,
() -> assertThat(
- stopAsync(componentContext, indexMetaStorage, catalogManager, metastore, vaultManager),
+ stopAsync(componentContext, indexMetaStorage, catalogManager, metastore),
willCompleteSuccessfully()
)
);
@@ -113,7 +110,7 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
catalogManager = createCatalogManager();
- indexMetaStorage = new IndexMetaStorage(catalogManager, lowWatermark, vaultManager);
+ indexMetaStorage = new IndexMetaStorage(catalogManager, lowWatermark, metastore);
}
int indexId(String indexName) {
@@ -124,10 +121,24 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
return getTableIdStrict(catalogManager, tableName, clock.nowLong());
}
- @Nullable IndexMeta fromVault(int indexId) {
- VaultEntry vaultEntry = vaultManager.get(ByteArray.fromString("index.meta." + indexId));
+ @Nullable IndexMeta fromMetastore(int indexId) {
+ byte[] versionBytes = getFromMetastore(ByteArray.fromString("index.meta.version." + indexId)).value();
+ byte[] valueBytes = getFromMetastore(ByteArray.fromString("index.meta.value." + indexId)).value();
+
+ if (valueBytes == null) {
+ assertNull(versionBytes, "indexId=" + indexId);
+
+ return null;
+ }
+
+ assertNotNull(versionBytes, "indexId=" + indexId);
- return vaultEntry == null ? null : ByteUtils.fromBytes(vaultEntry.value());
+ int catalogVersion = ByteUtils.bytesToInt(versionBytes);
+ IndexMeta indexMeta = ByteUtils.fromBytes(valueBytes);
+
+ assertEquals(indexMeta.catalogVersion(), catalogVersion, "indexId=" + indexId);
+
+ return indexMeta;
}
MetaIndexStatusChange toChangeInfo(int catalogVersion) {
@@ -147,7 +158,7 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
int executeCatalogUpdate(Runnable task) {
task.run();
- return catalogManager.latestCatalogVersion();
+ return latestCatalogVersion();
}
void updateLwm(HybridTimestamp newLwm) {
@@ -160,14 +171,28 @@ abstract class BaseIndexMetaStorageTest extends BaseIgniteAbstractTest {
int expTableId,
String expIndexName,
MetaIndexStatus expStatus,
- Map<MetaIndexStatus, MetaIndexStatusChange> expStatuses
+ Map<MetaIndexStatus, MetaIndexStatusChange> expStatuses,
+ int expCatalogVersion
) {
- assertNotNull(indexMeta);
+ assertNotNull(indexMeta, "indexId=" + expIndexId);
assertEquals(expIndexId, indexMeta.indexId());
- assertEquals(expTableId, indexMeta.tableId());
- assertEquals(expIndexName, indexMeta.indexName());
- assertEquals(expStatus, indexMeta.status());
- assertEquals(expStatuses, indexMeta.statusChanges());
+ assertEquals(expTableId, indexMeta.tableId(), "indexId=" + expIndexId);
+ assertEquals(expIndexName, indexMeta.indexName(), "indexId=" + expIndexId);
+ assertEquals(expStatus, indexMeta.status(), "indexId=" + expIndexId);
+ assertEquals(expStatuses, indexMeta.statusChanges(), "indexId=" + expIndexId);
+ assertEquals(expCatalogVersion, indexMeta.catalogVersion(), "indexId=" + expIndexId);
+ }
+
+ private Entry getFromMetastore(ByteArray key) {
+ CompletableFuture<Entry> future = metastore.get(key);
+
+ assertThat(future, willCompleteSuccessfully());
+
+ return future.join();
+ }
+
+ int latestCatalogVersion() {
+ return catalogManager.latestCatalogVersion();
}
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageRecoveryTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageRecoveryTest.java
index 1c51b8fe84..c7f2ebde69 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageRecoveryTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageRecoveryTest.java
@@ -89,20 +89,22 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
int indexId = indexId(INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int catalogVersion = catalogManager.latestCatalogVersion();
+ int catalogVersion = latestCatalogVersion();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REGISTERED, Map.of(REGISTERED, toChangeInfo(catalogVersion)));
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REGISTERED, Map.of(REGISTERED, toChangeInfo(catalogVersion)));
+ Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(REGISTERED, toChangeInfo(catalogVersion));
+
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REGISTERED, expectedStatuses, catalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REGISTERED, expectedStatuses, catalogVersion);
}
@Test
void testMissingRenameTable() throws Exception {
int indexId = indexId(PK_INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int catalogVersion = catalogManager.latestCatalogVersion();
+ int catalogVersion = latestCatalogVersion();
executeCatalogUpdateWithDropEvents(() -> renameSimpleTable(catalogManager, TABLE_NAME, NEW_TABLE_NAME));
@@ -111,10 +113,14 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(AVAILABLE, toChangeInfo(catalogVersion));
+
+ int latestCatalogVersion = latestCatalogVersion();
- checkFields(indexMeta, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
- checkFields(fromVault, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
+ checkFields(indexMeta, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -131,15 +137,17 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
- BUILDING, toChangeInfo(catalogManager.latestCatalogVersion())
+ BUILDING, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -158,16 +166,18 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
- AVAILABLE, toChangeInfo(catalogManager.latestCatalogVersion())
+ AVAILABLE, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -187,17 +197,19 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
AVAILABLE, toChangeInfo(availableIndexCatalogVersion),
- STOPPING, toChangeInfo(catalogManager.latestCatalogVersion())
+ STOPPING, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -218,25 +230,27 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
AVAILABLE, toChangeInfo(availableIndexCatalogVersion),
STOPPING, toChangeInfo(stoppingIndexCatalogVersion),
- READ_ONLY, toChangeInfo(catalogManager.latestCatalogVersion())
+ READ_ONLY, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
}
@Test
void testMissingDropTable() throws Exception {
int indexId = indexId(PK_INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int availableIndexCatalogVersion = catalogManager.latestCatalogVersion();
+ int availableIndexCatalogVersion = latestCatalogVersion();
executeCatalogUpdateWithDropEvents(() -> dropSimpleTable(catalogManager, TABLE_NAME));
@@ -245,15 +259,17 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
AVAILABLE, toChangeInfo(availableIndexCatalogVersion),
- READ_ONLY, toChangeInfo(catalogManager.latestCatalogVersion())
+ READ_ONLY, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses);
- checkFields(fromVault, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -270,15 +286,17 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
- REMOVED, toChangeInfo(catalogManager.latestCatalogVersion())
+ REMOVED, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -297,16 +315,18 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
- REMOVED, toChangeInfo(catalogManager.latestCatalogVersion())
+ REMOVED, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -322,7 +342,7 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@Test
@@ -340,7 +360,7 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@Test
@@ -359,7 +379,7 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
assertNotNull(indexMetaStorage.indexMeta(indexId));
- assertNotNull(fromVault(indexId));
+ assertNotNull(fromMetastore(indexId));
}
@Test
@@ -379,7 +399,7 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@Test
@@ -393,7 +413,7 @@ public class IndexMetaStorageRecoveryTest extends BaseIndexMetaStorageTest {
restartComponents();
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
private void executeCatalogUpdateWithDropEvents(RunnableX task) {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageTest.java
index 3ddb4e59a7..2fa941d66e 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexMetaStorageTest.java
@@ -67,28 +67,34 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
void testPkIndex() {
int indexId = indexId(PK_INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int catalogVersion = catalogManager.latestCatalogVersion();
+ int catalogVersion = latestCatalogVersion();
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
- checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
- checkFields(fromVault, indexId, tableId, PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
+ Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(AVAILABLE, toChangeInfo(catalogVersion));
+
+ checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, AVAILABLE, expectedStatuses, catalogVersion);
+ checkFields(fromMetastore, indexId, tableId, PK_INDEX_NAME, AVAILABLE, expectedStatuses, catalogVersion);
}
@Test
void testRenameTable() {
int indexId = indexId(PK_INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int catalogVersion = catalogManager.latestCatalogVersion();
+ int catalogVersion = latestCatalogVersion();
renameSimpleTable(catalogManager, TABLE_NAME, NEW_TABLE_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(AVAILABLE, toChangeInfo(catalogVersion));
+
+ int latestCatalogVersion = latestCatalogVersion();
- checkFields(indexMeta, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
- checkFields(fromVault, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, Map.of(AVAILABLE, toChangeInfo(catalogVersion)));
+ checkFields(indexMeta, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, NEW_PK_INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -99,10 +105,12 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
int tableId = tableId(TABLE_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REGISTERED, Map.of(REGISTERED, toChangeInfo(catalogVersion)));
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REGISTERED, Map.of(REGISTERED, toChangeInfo(catalogVersion)));
+ Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(REGISTERED, toChangeInfo(catalogVersion));
+
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REGISTERED, expectedStatuses, catalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REGISTERED, expectedStatuses, catalogVersion);
}
@Test
@@ -115,15 +123,17 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
startBuildingIndex(catalogManager, indexId);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
- BUILDING, toChangeInfo(catalogManager.latestCatalogVersion())
+ BUILDING, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, BUILDING, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -138,16 +148,18 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
makeIndexAvailable(catalogManager, indexId);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
- AVAILABLE, toChangeInfo(catalogManager.latestCatalogVersion())
+ AVAILABLE, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, AVAILABLE, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -163,17 +175,19 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
dropSimpleIndex(catalogManager, INDEX_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
AVAILABLE, toChangeInfo(availableIndexCatalogVersion),
- STOPPING, toChangeInfo(catalogManager.latestCatalogVersion())
+ STOPPING, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, STOPPING, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -186,15 +200,17 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
dropSimpleIndex(catalogManager, INDEX_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
- REMOVED, toChangeInfo(catalogManager.latestCatalogVersion())
+ REMOVED, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -209,16 +225,18 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
dropSimpleIndex(catalogManager, INDEX_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
- REMOVED, toChangeInfo(catalogManager.latestCatalogVersion())
+ REMOVED, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, REMOVED, expectedStatuses, latestCatalogVersion);
}
@Test
@@ -235,38 +253,42 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
removeIndex(catalogManager, indexId);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
REGISTERED, toChangeInfo(registeredIndexCatalogVersion),
BUILDING, toChangeInfo(buildingIndexCatalogVersion),
AVAILABLE, toChangeInfo(availableIndexCatalogVersion),
STOPPING, toChangeInfo(stoppingIndexCatalogVersion),
- READ_ONLY, toChangeInfo(catalogManager.latestCatalogVersion())
+ READ_ONLY, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses);
- checkFields(fromVault, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
}
@Test
void testRemoveTable() {
int indexId = indexId(PK_INDEX_NAME);
int tableId = tableId(TABLE_NAME);
- int afterCreateTableCatalogVersion = catalogManager.latestCatalogVersion();
+ int afterCreateTableCatalogVersion = latestCatalogVersion();
dropSimpleTable(catalogManager, TABLE_NAME);
IndexMeta indexMeta = indexMetaStorage.indexMeta(indexId);
- IndexMeta fromVault = fromVault(indexId);
+ IndexMeta fromMetastore = fromMetastore(indexId);
+
+ int latestCatalogVersion = latestCatalogVersion();
Map<MetaIndexStatus, MetaIndexStatusChange> expectedStatuses = Map.of(
AVAILABLE, toChangeInfo(afterCreateTableCatalogVersion),
- READ_ONLY, toChangeInfo(catalogManager.latestCatalogVersion())
+ READ_ONLY, toChangeInfo(latestCatalogVersion)
);
- checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses);
- checkFields(fromVault, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses);
+ checkFields(indexMeta, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
+ checkFields(fromMetastore, indexId, tableId, PK_INDEX_NAME, READ_ONLY, expectedStatuses, latestCatalogVersion);
}
@ParameterizedTest
@@ -287,7 +309,7 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
}
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@ParameterizedTest
@@ -310,7 +332,7 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
}
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@ParameterizedTest
@@ -334,7 +356,7 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
}
assertNotNull(indexMetaStorage.indexMeta(indexId));
- assertNotNull(fromVault(indexId));
+ assertNotNull(fromMetastore(indexId));
}
@ParameterizedTest
@@ -359,7 +381,7 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
}
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
@ParameterizedTest
@@ -378,6 +400,6 @@ public class IndexMetaStorageTest extends BaseIndexMetaStorageTest {
}
assertNull(indexMetaStorage.indexMeta(indexId));
- assertNull(fromVault(indexId));
+ assertNull(fromMetastore(indexId));
}
}