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/15 16:27:41 UTC
(ignite-3) branch ignite-21535 created (now e64ca459ab)
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a change to branch ignite-21535
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at e64ca459ab Add test
This branch includes the following new commits:
new 15de215c25 Create catalog events for destroying table and index.
new 1a4ad96cef Add tests
new e64ca459ab Add test
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 03/03: Add test
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-21535
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit e64ca459ab83fe36d99890f072eb581450c9c0a2
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Feb 15 19:27:32 2024 +0300
Add test
---
.../ignite/internal/index/IndexManagerTest.java | 38 ++++++++++++++++++++++
1 file changed, 38 insertions(+)
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
index 6633a11807..d7aeb7ed23 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
@@ -100,6 +100,7 @@ import org.apache.ignite.sql.IgniteSql;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
@@ -219,6 +220,43 @@ public class IndexManagerTest extends BaseIgniteAbstractTest {
verify(tableViewInternal, never()).unregisterIndex(anyInt());
}
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-17626")
+ @Test
+ void testDestroyIndex() throws Exception {
+ createIndex(TABLE_NAME, INDEX_NAME);
+
+ CatalogIndexDescriptor indexDescriptor = catalogManager.index(INDEX_NAME, catalogManager.latestCatalogVersion());
+ int indexId = indexDescriptor.id();
+ int tableId = indexDescriptor.tableId();
+
+ dropIndex(INDEX_NAME);
+ CatalogTestUtils.waitCatalogCompaction(catalogManager, Long.MAX_VALUE);
+
+ long causalityToken = 0L; // Use last token.
+ MvTableStorage mvTableStorage = indexManager.getMvTableStorage(causalityToken, tableId).get();
+
+ verify(mvTableStorage).destroyIndex(indexId);
+ }
+
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-17626")
+ @Test
+ void testIndexDestroyedWithTable() throws Exception {
+ createIndex(TABLE_NAME, INDEX_NAME);
+
+ CatalogIndexDescriptor indexDescriptor = catalogManager.index(INDEX_NAME, catalogManager.latestCatalogVersion());
+ int indexId = indexDescriptor.id();
+ int tableId = indexDescriptor.tableId();
+
+ dropTable(TABLE_NAME);
+ CatalogTestUtils.waitCatalogCompaction(catalogManager, Long.MAX_VALUE);
+
+ long causalityToken = 0L; // Use last token.
+ MvTableStorage mvTableStorage = indexManager.getMvTableStorage(causalityToken, tableId).get();
+
+ verify(mvTableStorage).destroyIndex(indexId);
+ }
+
+
@Test
void testCollectIndexesForRecoveryForCreatedTables() {
createTable(OTHER_TABLE_NAME);
(ignite-3) 01/03: Create catalog events for destroying table and index.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-21535
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 15de215c2567d2702d33ed5565bf5e06bb0bdd75
Author: amashenkov <an...@gmail.com>
AuthorDate: Wed Feb 14 19:18:17 2024 +0300
Create catalog events for destroying table and index.
---
.../internal/catalog/CatalogManagerImpl.java | 61 +++++++++++++++++++---
.../internal/catalog/events/CatalogEvent.java | 6 +++
...EventParameters.java => DestroyIndexEvent.java} | 37 ++++++-------
...eters.java => DestroyIndexEventParameters.java} | 36 ++++++-------
...EventParameters.java => DestroyTableEvent.java} | 30 ++++++-----
...eters.java => DestroyTableEventParameters.java} | 24 +++++----
...ntParameters.java => IndexEventParameters.java} | 13 ++---
.../events/MakeIndexAvailableEventParameters.java | 13 +----
.../catalog/events/RemoveIndexEventParameters.java | 13 +----
.../events/StartBuildingIndexEventParameters.java | 13 +----
.../events/StoppingIndexEventParameters.java | 13 +----
.../ignite/internal/catalog/storage/UpdateLog.java | 4 +-
.../internal/catalog/storage/UpdateLogImpl.java | 6 +--
.../internal/catalog/BaseCatalogManagerTest.java | 2 +
.../ignite/internal/catalog/CatalogTestUtils.java | 31 ++++++++++-
.../internal/table/distributed/TableManager.java | 12 +++--
16 files changed, 180 insertions(+), 134 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
index aa99fcdfff..a0bab25a90 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
@@ -25,12 +25,15 @@ import static org.apache.ignite.internal.catalog.commands.CatalogUtils.fromParam
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.Flow.Publisher;
@@ -45,6 +48,8 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CatalogEventParameters;
+import org.apache.ignite.internal.catalog.events.DestroyIndexEvent;
+import org.apache.ignite.internal.catalog.events.DestroyTableEvent;
import org.apache.ignite.internal.catalog.storage.Fireable;
import org.apache.ignite.internal.catalog.storage.SnapshotEntry;
import org.apache.ignite.internal.catalog.storage.UpdateEntry;
@@ -338,13 +343,13 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
* Cleanup outdated catalog versions, which can't be observed after given timestamp (inclusively), and compact underlying update log.
*
* @param timestamp Earliest observable timestamp.
- * @return Operation future.
+ * @return Operation future, which is completing with {@code true} if a new snapshot has been successfully written, {@code false}
+ * otherwise if a snapshot with the same or greater version already exists.
*/
- public CompletableFuture<Void> compactCatalog(long timestamp) {
+ public CompletableFuture<Boolean> compactCatalog(long timestamp) {
Catalog catalog = catalogAt(timestamp);
- return updateLog.saveSnapshot(new SnapshotEntry(catalog))
- .thenAccept(ignore -> {});
+ return updateLog.saveSnapshot(new SnapshotEntry(catalog));
}
private void registerCatalog(Catalog newCatalog) {
@@ -437,21 +442,57 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
@Override
public CompletableFuture<Void> handle(UpdateLogEvent event, HybridTimestamp metaStorageUpdateTimestamp, long causalityToken) {
if (event instanceof SnapshotEntry) {
- return handle((SnapshotEntry) event);
+ return handle((SnapshotEntry) event, causalityToken);
}
return handle((VersionedUpdate) event, metaStorageUpdateTimestamp, causalityToken);
}
- private CompletableFuture<Void> handle(SnapshotEntry event) {
+ private CompletableFuture<Void> handle(SnapshotEntry event, long causalityToken) {
Catalog catalog = event.snapshot();
+ // Use reverse order to find latest descriptors.
+ Collection<Catalog> droppedCatalogVersions = catalogByVer.headMap(catalog.version(), false).descendingMap().values();
+
+ // Collect destroy events for dropped tables/indexes.
+ IntSet droppedObjects = new IntOpenHashSet();
+ List<Fireable> events = new ArrayList<>();
+
+ droppedCatalogVersions.forEach(oldCatalog -> oldCatalog.indexes().stream()
+ .filter(idx -> catalog.index(idx.id()) == null)
+ .filter(idx -> droppedObjects.add(idx.id()))
+ .forEach(idx -> events.add(
+ new DestroyIndexEvent(idx.id(), idx.tableId(), tableZoneDescriptor(oldCatalog, idx.tableId()).partitions()))
+ ));
+
+ droppedObjects.clear();
+ droppedCatalogVersions.forEach(oldCatalog -> oldCatalog.tables().stream()
+ .filter(tbl -> catalog.table(tbl.id()) == null)
+ .filter(tbl -> droppedObjects.add(tbl.id()))
+ .forEach(tbl -> events.add(new DestroyTableEvent(tbl.id(), tableZoneDescriptor(oldCatalog, tbl.id()).partitions()))));
+
// On recovery phase, we must register catalog from the snapshot.
// In other cases, it is ok to rewrite an existed version, because it's exactly the same.
registerCatalog(catalog);
- truncateUpTo(catalog);
- return nullCompletedFuture();
+ List<CompletableFuture<?>> eventFutures = new ArrayList<>(events.size());
+
+ for (Fireable fireEvent : events) {
+ eventFutures.add(fireEvent(
+ fireEvent.eventType(),
+ fireEvent.createEventParameters(causalityToken, catalog.version())
+ ));
+ }
+
+ return allOf(eventFutures.toArray(CompletableFuture[]::new))
+ .whenComplete((ignore, err) -> {
+ if (err != null) {
+ LOG.warn("Failed to compact catalog.", err);
+ //TODO: IGNITE-14611 Pass exception to an error handler?
+ } else {
+ truncateUpTo(catalog);
+ }
+ });
}
private CompletableFuture<Void> handle(VersionedUpdate update, HybridTimestamp metaStorageUpdateTimestamp, long causalityToken) {
@@ -498,6 +539,10 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
}
}
+ private static CatalogZoneDescriptor tableZoneDescriptor(Catalog oldCatalog, int tableId) {
+ return Objects.requireNonNull(oldCatalog.zone(Objects.requireNonNull(oldCatalog.table(tableId)).zoneId()));
+ }
+
private static Catalog applyUpdateFinal(Catalog catalog, VersionedUpdate update, HybridTimestamp metaStorageUpdateTimestamp) {
long activationTimestamp = metaStorageUpdateTimestamp.addPhysicalTime(update.delayDurationMs()).longValue();
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/CatalogEvent.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/CatalogEvent.java
index 2c65d1f516..ecc5e60c0a 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/CatalogEvent.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/CatalogEvent.java
@@ -33,6 +33,9 @@ public enum CatalogEvent implements Event {
/** This event is fired when a table has been renamed or a column has been modified, added to, or removed from a table. */
TABLE_ALTER,
+ /** This event is fired when a table has been dropped from all catalog versions and can be destroyed. */
+ TABLE_DESTROY,
+
/** This event is fired, when an index was created in Catalog. */
INDEX_CREATE,
@@ -55,6 +58,9 @@ public enum CatalogEvent implements Event {
*/
INDEX_REMOVED,
+ /** This event is fired when an index has been dropped from all catalog versions and can be destroyed. */
+ INDEX_DESTROY,
+
/** This event is fired, when a distribution zone was created in Catalog. */
ZONE_CREATE,
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEvent.java
similarity index 56%
copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEvent.java
index 51e1536a26..6f9b680ee4 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEvent.java
@@ -17,41 +17,34 @@
package org.apache.ignite.internal.catalog.events;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
-
-/**
- * Event parameters for the 'index has moved to the {@link CatalogIndexStatus#STOPPING} that contains an id of the dropped index.
- *
- * @see CatalogEvent#INDEX_STOPPING
- */
-public class StoppingIndexEventParameters extends CatalogEventParameters {
-
- private final int indexId;
+import org.apache.ignite.internal.catalog.storage.Fireable;
+/** {@link CatalogEvent#INDEX_DESTROY} event. */
+public class DestroyIndexEvent implements Fireable {
private final int tableId;
+ private final int indexId;
+ private final int partitions;
/**
* Constructor.
*
- * @param causalityToken Causality token.
- * @param catalogVersion Catalog version.
* @param indexId An id of dropped index.
- * @param tableId Table ID for which the index was removed.
+ * @param tableId An id of table the index belongs to.
+ * @param partitions Table partitions.
*/
- public StoppingIndexEventParameters(long causalityToken, int catalogVersion, int indexId, int tableId) {
- super(causalityToken, catalogVersion);
-
+ public DestroyIndexEvent(int indexId, int tableId, int partitions) {
this.indexId = indexId;
this.tableId = tableId;
+ this.partitions = partitions;
}
- /** Returns an id of dropped index. */
- public int indexId() {
- return indexId;
+ @Override
+ public CatalogEvent eventType() {
+ return CatalogEvent.INDEX_DESTROY;
}
- /** Returns table ID for which the index was removed. */
- public int tableId() {
- return tableId;
+ @Override
+ public CatalogEventParameters createEventParameters(long causalityToken, int catalogVersion) {
+ return new DestroyIndexEventParameters(causalityToken, catalogVersion, indexId, tableId, partitions);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEventParameters.java
similarity index 58%
copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEventParameters.java
index 51e1536a26..d695dcf5c8 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyIndexEventParameters.java
@@ -17,41 +17,35 @@
package org.apache.ignite.internal.catalog.events;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
-
/**
- * Event parameters for the 'index has moved to the {@link CatalogIndexStatus#STOPPING} that contains an id of the dropped index.
- *
- * @see CatalogEvent#INDEX_STOPPING
+ * Destroy index event parameters contains an id of destroying index.
*/
-public class StoppingIndexEventParameters extends CatalogEventParameters {
-
- private final int indexId;
-
+public class DestroyIndexEventParameters extends IndexEventParameters {
private final int tableId;
+ private final int partitions;
/**
* Constructor.
*
* @param causalityToken Causality token.
* @param catalogVersion Catalog version.
- * @param indexId An id of dropped index.
- * @param tableId Table ID for which the index was removed.
+ * @param indexId An Id of destroying index.
+ * @param tableId An Id of table the index belongs to.
+ * @param partitions Number of table partitions.
*/
- public StoppingIndexEventParameters(long causalityToken, int catalogVersion, int indexId, int tableId) {
- super(causalityToken, catalogVersion);
-
- this.indexId = indexId;
+ public DestroyIndexEventParameters(long causalityToken, int catalogVersion, int indexId, int tableId, int partitions) {
+ super(causalityToken, catalogVersion, indexId);
this.tableId = tableId;
+ this.partitions = partitions;
}
- /** Returns an id of dropped index. */
- public int indexId() {
- return indexId;
- }
-
- /** Returns table ID for which the index was removed. */
+ /** Returns an id of table the index belongs to. */
public int tableId() {
return tableId;
}
+
+ /** Returns number of table partitions. */
+ public int partitions() {
+ return partitions;
+ }
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEvent.java
similarity index 54%
copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEvent.java
index 797b75de53..3bdd177588 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEvent.java
@@ -17,25 +17,31 @@
package org.apache.ignite.internal.catalog.events;
-/** {@link CatalogEvent#INDEX_BUILDING} event parameters. */
-public class StartBuildingIndexEventParameters extends CatalogEventParameters {
- private final int indexId;
+import org.apache.ignite.internal.catalog.storage.Fireable;
+
+/** {@link CatalogEvent#TABLE_DESTROY} event. */
+public class DestroyTableEvent implements Fireable {
+ private final int tableId;
+ private final int partitions;
/**
* Constructor.
*
- * @param causalityToken Causality token.
- * @param catalogVersion Catalog version.
- * @param indexId Index ID.
+ * @param tableId An id of dropped table.
+ * @param partitions Table partitions.
*/
- public StartBuildingIndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
- super(causalityToken, catalogVersion);
+ public DestroyTableEvent(int tableId, int partitions) {
+ this.tableId = tableId;
+ this.partitions = partitions;
+ }
- this.indexId = indexId;
+ @Override
+ public CatalogEvent eventType() {
+ return CatalogEvent.TABLE_DESTROY;
}
- /** Returns index ID. */
- public int indexId() {
- return indexId;
+ @Override
+ public CatalogEventParameters createEventParameters(long causalityToken, int catalogVersion) {
+ return new DestroyTableEventParameters(causalityToken, catalogVersion, tableId, partitions);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEventParameters.java
similarity index 62%
copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEventParameters.java
index 797b75de53..5bc3bfd114 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/DestroyTableEventParameters.java
@@ -17,25 +17,27 @@
package org.apache.ignite.internal.catalog.events;
-/** {@link CatalogEvent#INDEX_BUILDING} event parameters. */
-public class StartBuildingIndexEventParameters extends CatalogEventParameters {
- private final int indexId;
+/**
+ * Destroy table event parameters contains an id of destroying table.
+ */
+public class DestroyTableEventParameters extends TableEventParameters {
+ private final int partitions;
/**
* Constructor.
*
* @param causalityToken Causality token.
* @param catalogVersion Catalog version.
- * @param indexId Index ID.
+ * @param tableId An Id of destroying table.
+ * @param partitions Number of table partitions.
*/
- public StartBuildingIndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
- super(causalityToken, catalogVersion);
-
- this.indexId = indexId;
+ public DestroyTableEventParameters(long causalityToken, int catalogVersion, int tableId, int partitions) {
+ super(causalityToken, catalogVersion, tableId);
+ this.partitions = partitions;
}
- /** Returns index ID. */
- public int indexId() {
- return indexId;
+ /** Returns number of table partitions. */
+ public int partitions() {
+ return partitions;
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/IndexEventParameters.java
similarity index 78%
copy from modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
copy to modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/IndexEventParameters.java
index 797b75de53..5c3116ffec 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/IndexEventParameters.java
@@ -17,8 +17,10 @@
package org.apache.ignite.internal.catalog.events;
-/** {@link CatalogEvent#INDEX_BUILDING} event parameters. */
-public class StartBuildingIndexEventParameters extends CatalogEventParameters {
+/**
+ * Event that is related to an index.
+ */
+public abstract class IndexEventParameters extends CatalogEventParameters {
private final int indexId;
/**
@@ -26,15 +28,14 @@ public class StartBuildingIndexEventParameters extends CatalogEventParameters {
*
* @param causalityToken Causality token.
* @param catalogVersion Catalog version.
- * @param indexId Index ID.
+ * @param indexId ID of the index to which the event relates.
*/
- public StartBuildingIndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
+ public IndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
super(causalityToken, catalogVersion);
-
this.indexId = indexId;
}
- /** Returns index ID. */
+ /** Returns an id of a modified index. */
public int indexId() {
return indexId;
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/MakeIndexAvailableEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/MakeIndexAvailableEventParameters.java
index c055ec619e..ab53b12f7f 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/MakeIndexAvailableEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/MakeIndexAvailableEventParameters.java
@@ -18,9 +18,7 @@
package org.apache.ignite.internal.catalog.events;
/** {@link CatalogEvent#INDEX_AVAILABLE} event parameters. */
-public class MakeIndexAvailableEventParameters extends CatalogEventParameters {
- private final int indexId;
-
+public class MakeIndexAvailableEventParameters extends IndexEventParameters {
/**
* Constructor.
*
@@ -29,13 +27,6 @@ public class MakeIndexAvailableEventParameters extends CatalogEventParameters {
* @param indexId Index ID.
*/
public MakeIndexAvailableEventParameters(long causalityToken, int catalogVersion, int indexId) {
- super(causalityToken, catalogVersion);
-
- this.indexId = indexId;
- }
-
- /** Returns index ID. */
- public int indexId() {
- return indexId;
+ super(causalityToken, catalogVersion, indexId);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/RemoveIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/RemoveIndexEventParameters.java
index d299b04e05..4940e86783 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/RemoveIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/RemoveIndexEventParameters.java
@@ -23,9 +23,7 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
* Event parameters for an 'index removed from the Catalog' event (don't confuse it with {@link StoppingIndexEventParameters}
* that is about switching the index to the {@link CatalogIndexStatus#STOPPING} state).
*/
-public class RemoveIndexEventParameters extends CatalogEventParameters {
- private final int indexId;
-
+public class RemoveIndexEventParameters extends IndexEventParameters {
/**
* Constructor.
*
@@ -34,13 +32,6 @@ public class RemoveIndexEventParameters extends CatalogEventParameters {
* @param indexId Index ID.
*/
public RemoveIndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
- super(causalityToken, catalogVersion);
-
- this.indexId = indexId;
- }
-
- /** Returns index ID. */
- public int indexId() {
- return indexId;
+ super(causalityToken, catalogVersion, indexId);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
index 797b75de53..d41e3aeff6 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StartBuildingIndexEventParameters.java
@@ -18,9 +18,7 @@
package org.apache.ignite.internal.catalog.events;
/** {@link CatalogEvent#INDEX_BUILDING} event parameters. */
-public class StartBuildingIndexEventParameters extends CatalogEventParameters {
- private final int indexId;
-
+public class StartBuildingIndexEventParameters extends IndexEventParameters {
/**
* Constructor.
*
@@ -29,13 +27,6 @@ public class StartBuildingIndexEventParameters extends CatalogEventParameters {
* @param indexId Index ID.
*/
public StartBuildingIndexEventParameters(long causalityToken, int catalogVersion, int indexId) {
- super(causalityToken, catalogVersion);
-
- this.indexId = indexId;
- }
-
- /** Returns index ID. */
- public int indexId() {
- return indexId;
+ super(causalityToken, catalogVersion, indexId);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
index 51e1536a26..ac8864b013 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/events/StoppingIndexEventParameters.java
@@ -24,10 +24,7 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
*
* @see CatalogEvent#INDEX_STOPPING
*/
-public class StoppingIndexEventParameters extends CatalogEventParameters {
-
- private final int indexId;
-
+public class StoppingIndexEventParameters extends IndexEventParameters {
private final int tableId;
/**
@@ -39,17 +36,11 @@ public class StoppingIndexEventParameters extends CatalogEventParameters {
* @param tableId Table ID for which the index was removed.
*/
public StoppingIndexEventParameters(long causalityToken, int catalogVersion, int indexId, int tableId) {
- super(causalityToken, catalogVersion);
+ super(causalityToken, catalogVersion, indexId);
- this.indexId = indexId;
this.tableId = tableId;
}
- /** Returns an id of dropped index. */
- public int indexId() {
- return indexId;
- }
-
/** Returns table ID for which the index was removed. */
public int tableId() {
return tableId;
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLog.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLog.java
index ba1c7b495c..3dec66fdcf 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLog.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLog.java
@@ -45,8 +45,8 @@ public interface UpdateLog extends IgniteComponent {
* Saves a snapshot entry and drop updates of previous versions from the log, if supported, otherwise do nothing.
*
* @param snapshotEntry An entry, which represents a result of merging updates of previous versions.
- * @return A {@code true} if snapshot has been successfully appended, {@code false} otherwise
- * if a snapshot with the same or greater version already exists, or snapshots are not supported.
+ * @return A {@code true} if snapshot has been successfully written, {@code false} otherwise
+ * if a snapshot with the same or greater version already exists.
*/
CompletableFuture<Boolean> saveSnapshot(SnapshotEntry snapshotEntry);
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java
index 04faa643d5..b711bfe807 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/UpdateLogImpl.java
@@ -27,6 +27,7 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
import static org.apache.ignite.internal.util.ByteUtils.bytesToInt;
import static org.apache.ignite.internal.util.ByteUtils.intToBytes;
+import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import java.util.ArrayList;
@@ -54,7 +55,6 @@ import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.dsl.Operations;
import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.metastorage.dsl.Update;
-import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.lang.ErrorGroups.Common;
import org.apache.ignite.lang.IgniteException;
@@ -114,7 +114,7 @@ public class UpdateLogImpl implements UpdateLog {
recoveryStateFromMetastore(handler);
- UpdateListener listener = new UpdateListener(onUpdateHandler, marshaller);
+ UpdateListener listener = new UpdateListener(handler, marshaller);
this.listener = listener;
metastore.registerPrefixWatch(CatalogKey.updatePrefix(), listener);
@@ -196,7 +196,7 @@ public class UpdateLogImpl implements UpdateLog {
if (oldSnapshotVersion >= snapshotVersion) {
// Nothing to do.
- return CompletableFutures.trueCompletedFuture();
+ return falseCompletedFuture();
}
Condition versionIsRecent = or(
diff --git a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
index 360e324528..b02ccad23f 100644
--- a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
+++ b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.catalog.commands.CreateHashIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateSortedIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
import org.apache.ignite.internal.catalog.commands.CreateTableCommandBuilder;
+import org.apache.ignite.internal.catalog.commands.DropTableCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
import org.apache.ignite.internal.catalog.storage.UpdateLog;
@@ -61,6 +62,7 @@ public abstract class BaseCatalogManagerTest extends BaseIgniteAbstractTest {
protected static final String TABLE_NAME = "test_table";
protected static final String TABLE_NAME_2 = "test_table_2";
+ protected static final String TABLE_NAME_3 = "test_table_3";
protected static final String INDEX_NAME = "myIndex";
protected static final String INDEX_NAME_2 = "myIndex2";
diff --git a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
index 18146c6433..1b28c39c28 100644
--- a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
+++ b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
@@ -19,9 +19,12 @@ package org.apache.ignite.internal.catalog;
import static java.util.concurrent.CompletableFuture.allOf;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_SCHEMA_NAME;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
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.CompletableFutures.trueCompletedFuture;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.fail;
import java.util.List;
import java.util.Set;
@@ -46,7 +49,6 @@ import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
-import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.lang.ErrorGroups.Common;
import org.apache.ignite.sql.ColumnType;
import org.jetbrains.annotations.Nullable;
@@ -291,6 +293,31 @@ public class CatalogTestUtils {
return AlterZoneCommand.builder().zoneName(zoneName);
}
+ /**
+ * Starts catalog compaction and waits it finished locally.
+ *
+ * @param catalogManager Catalog manager.
+ * @param timestamp Timestamp catalog should be compacted up to.
+ * @return {@code True} if a new snapshot has been successfully written, {@code false} otherwise.
+ */
+ static boolean waitCatalogCompaction(CatalogManager catalogManager, long timestamp) {
+ int version = catalogManager.activeCatalogVersion(timestamp);
+
+ CompletableFuture<Boolean> operationFuture = ((CatalogManagerImpl) catalogManager).compactCatalog(timestamp);
+
+ try {
+ boolean result = operationFuture.get();
+
+ if (result) {
+ waitForCondition(() -> catalogManager.earliestCatalogVersion() == version, 3_000);
+ }
+ } catch (Exception e) {
+ fail(e);
+ }
+
+ return operationFuture.join();
+ }
+
private static class TestUpdateLog implements UpdateLog {
private final HybridClock clock;
@@ -317,7 +344,7 @@ public class CatalogTestUtils {
@Override
public synchronized CompletableFuture<Boolean> saveSnapshot(SnapshotEntry snapshotEntry) {
snapshotVersion = snapshotEntry.version();
- return CompletableFutures.trueCompletedFuture();
+ return trueCompletedFuture();
}
@Override
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 4f440ce765..1f997dfcef 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
@@ -52,6 +52,7 @@ import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermin
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
@@ -2358,13 +2359,18 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
private void startTables(long recoveryRevision) {
sharedTxStateStorage.start();
- int catalogVersion = catalogService.latestCatalogVersion();
+ int earliestCatalogVersion = catalogService.earliestCatalogVersion();
+ int latestCatalogVersion = catalogService.latestCatalogVersion();
+ var startedTables = new IntOpenHashSet();
List<CompletableFuture<?>> startTableFutures = new ArrayList<>();
// TODO: IGNITE-20384 Clean up abandoned resources for dropped zones from volt and metastore
- for (CatalogTableDescriptor tableDescriptor : catalogService.tables(catalogVersion)) {
- startTableFutures.add(createTableLocally(recoveryRevision, catalogVersion, tableDescriptor, true));
+ for (int ver = latestCatalogVersion; ver >= earliestCatalogVersion; ver--) {
+ int ver0 = ver;
+ catalogService.tables(ver).stream()
+ .filter(tbl -> startedTables.add(tbl.id()))
+ .forEach(tableDescriptor -> startTableFutures.add(createTableLocally(recoveryRevision, ver0, tableDescriptor, true)));
}
// Forces you to wait until recovery is complete before the metastore watches is deployed to avoid races with catalog listeners.
(ignite-3) 02/03: Add tests
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-21535
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 1a4ad96cef2b8d89a509d208e3dbc623a902b309
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Feb 15 19:05:03 2024 +0300
Add tests
---
.../catalog/CatalogManagerRecoveryTest.java | 73 +++++++++----
.../internal/catalog/CatalogManagerSelfTest.java | 115 +++++++++++++++++++--
.../internal/catalog/BaseCatalogManagerTest.java | 15 +++
.../ignite/internal/catalog/CatalogTestUtils.java | 91 +++++++++++++++-
.../table/distributed/TableManagerTest.java | 33 ++++++
5 files changed, 293 insertions(+), 34 deletions(-)
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
index e1a53b1f27..953ec87c07 100644
--- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
@@ -22,18 +22,32 @@ import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.INDEX_NA
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.INDEX_NAME_2;
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.TABLE_NAME;
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.TABLE_NAME_2;
+import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.TABLE_NAME_3;
+import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.dropIndexCommand;
+import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.dropTableCommand;
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.simpleIndex;
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.simpleTable;
import static org.apache.ignite.internal.catalog.BaseCatalogManagerTest.startBuildingIndexCommand;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import java.nio.file.Path;
import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.catalog.CatalogTestUtils.TestUpdateHandlerInterceptor;
+import org.apache.ignite.internal.catalog.storage.SnapshotEntry;
+import org.apache.ignite.internal.catalog.storage.VersionedUpdate;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
@@ -42,13 +56,13 @@ import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.TestRocksDbKeyValueStorage;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.util.IgniteUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mockito;
/** For {@link CatalogManager} testing on recovery. */
@ExtendWith(WorkDirectoryExtension.class)
@@ -64,6 +78,8 @@ public class CatalogManagerRecoveryTest extends BaseIgniteAbstractTest {
private CatalogManager catalogManager;
+ private TestUpdateHandlerInterceptor interceptor;
+
@AfterEach
void tearDown() throws Exception {
IgniteUtils.stopAll(catalogManager, metaStorageManager);
@@ -100,6 +116,10 @@ public class CatalogManagerRecoveryTest extends BaseIgniteAbstractTest {
createAndStartComponents();
+ // Check recovery events.
+ verify(interceptor, times(2)).handle(any(VersionedUpdate.class), any(), anyLong());
+ verify(interceptor, Mockito.never()).handle(any(SnapshotEntry.class), any(), anyLong());
+
// Let's check that the versions for the points in time at which they were created are in place.
assertThat(catalogManager.activeCatalogVersion(time0), equalTo(catalogVersion0));
assertThat(catalogManager.activeCatalogVersion(time1), equalTo(catalogVersion1));
@@ -111,40 +131,47 @@ public class CatalogManagerRecoveryTest extends BaseIgniteAbstractTest {
// Let's create a couple of versions of the catalog.
assertThat(catalogManager.execute(simpleTable(TABLE_NAME)), willCompleteSuccessfully());
+ assertThat(catalogManager.execute(simpleIndex(TABLE_NAME, INDEX_NAME)), willCompleteSuccessfully());
assertThat(catalogManager.execute(simpleTable(TABLE_NAME_2)), willCompleteSuccessfully());
+ assertThat(catalogManager.execute(simpleIndex(TABLE_NAME, INDEX_NAME_2)), willCompleteSuccessfully());
+ assertThat(catalogManager.execute(dropTableCommand(TABLE_NAME_2)), willCompleteSuccessfully());
+ assertThat(catalogManager.execute(dropIndexCommand(INDEX_NAME_2)), willCompleteSuccessfully());
- int catalogVersion0 = catalogManager.latestCatalogVersion();
- long time0 = catalogManager.catalog(catalogVersion0).time();
+ // Save version, which will be earliest version after snapshot.
+ int earliestVersion = catalogManager.latestCatalogVersion();
+ long earliestVersionActivationTime = catalogManager.catalog(earliestVersion).time();
+ long snapshotTime = clock.nowLong();
- assertThat(catalogVersion0, greaterThan(catalogManager.earliestCatalogVersion()));
+ assertThat(catalogManager.execute(simpleTable(TABLE_NAME_3)), willCompleteSuccessfully());
- assertThat(catalogManager.execute(simpleIndex(TABLE_NAME, INDEX_NAME)), willCompleteSuccessfully());
- assertThat(catalogManager.execute(simpleIndex(TABLE_NAME_2, INDEX_NAME_2)), willCompleteSuccessfully());
+ int latestVersion = catalogManager.latestCatalogVersion();
+ long latestVersionActivationTime = catalogManager.catalog(latestVersion).time();
- int catalogVersion1 = catalogManager.latestCatalogVersion();
- long time1 = catalogManager.catalog(catalogVersion1).time();
+ // Compact catalog with ignoring snapshot event, then ensure earliest version wasn't changed
+ clearInvocations(interceptor);
+ interceptor.dropSnapshotEvents();
- // Compact catalog.
- assertThat(((CatalogManagerImpl) catalogManager).compactCatalog(time0), willCompleteSuccessfully());
+ assertThat(((CatalogManagerImpl) catalogManager).compactCatalog(snapshotTime), willBe(true));
- IgniteTestUtils.waitForCondition(() -> catalogManager.earliestCatalogVersion() == catalogVersion0, 2_000);
-
- // Let's check outdated versions are not reachable.
- assertThrows(IllegalStateException.class, () -> catalogManager.activeCatalogVersion(0));
- assertThrows(IllegalStateException.class, () -> catalogManager.activeCatalogVersion(time0 - 1));
- assertThat(catalogManager.activeCatalogVersion(time0), equalTo(catalogVersion0));
- assertThat(catalogManager.activeCatalogVersion(time1), equalTo(catalogVersion1));
+ verify(interceptor, timeout(2_000)).handle(any(SnapshotEntry.class), any(), anyLong());
+ assertThat(catalogManager.earliestCatalogVersion(), equalTo(0));
// We will restart and recover the components and also set the clock to the future.
stopComponents();
createAndStartComponents();
- // Let's check that the versions for the points in time at which they were created are in place.
+ // Check recovery events.
+ verify(interceptor).handle(any(SnapshotEntry.class), any(), anyLong());
+ verify(interceptor).handle(any(VersionedUpdate.class), any(), anyLong());
+
+ // Let's check Catalog was recovered from a snapshot and history starts from the expected version.
+ assertThat(catalogManager.earliestCatalogVersion(), equalTo(earliestVersion));
assertThrows(IllegalStateException.class, () -> catalogManager.activeCatalogVersion(0));
- assertThrows(IllegalStateException.class, () -> catalogManager.activeCatalogVersion(time0 - 1));
- assertThat(catalogManager.activeCatalogVersion(time0), equalTo(catalogVersion0));
- assertThat(catalogManager.activeCatalogVersion(time1), equalTo(catalogVersion1));
+ assertThrows(IllegalStateException.class, () -> catalogManager.activeCatalogVersion(earliestVersionActivationTime - 1));
+ assertThat(catalogManager.activeCatalogVersion(earliestVersionActivationTime), equalTo(earliestVersion));
+ assertThat(catalogManager.activeCatalogVersion(snapshotTime), equalTo(earliestVersion));
+ assertThat(catalogManager.activeCatalogVersion(latestVersionActivationTime), equalTo(latestVersion));
}
@Test
@@ -179,7 +206,9 @@ public class CatalogManagerRecoveryTest extends BaseIgniteAbstractTest {
metaStorageManager = StandaloneMetaStorageManager.create(keyValueStorage);
- catalogManager = CatalogTestUtils.createTestCatalogManager(NODE_NAME, clock, metaStorageManager);
+ interceptor = spy(new TestUpdateHandlerInterceptor());
+
+ catalogManager = CatalogTestUtils.createTestCatalogManagerWithInterceptor(NODE_NAME, clock, metaStorageManager, interceptor);
}
private void startComponentsAndDeployWatches() {
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
index 7600345f5c..70e50158ba 100644
--- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
@@ -118,6 +118,7 @@ import org.apache.ignite.internal.catalog.commands.RenameZoneCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor.CatalogIndexDescriptorType;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.descriptors.CatalogObjectDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
@@ -131,6 +132,8 @@ import org.apache.ignite.internal.catalog.events.CatalogEventParameters;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
import org.apache.ignite.internal.catalog.events.CreateTableEventParameters;
import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.catalog.events.DestroyIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DestroyTableEventParameters;
import org.apache.ignite.internal.catalog.events.DropColumnEventParameters;
import org.apache.ignite.internal.catalog.events.DropTableEventParameters;
import org.apache.ignite.internal.catalog.events.DropZoneEventParameters;
@@ -146,7 +149,6 @@ import org.apache.ignite.internal.catalog.storage.VersionedUpdate;
import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalException;
-import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.sql.ColumnType;
import org.hamcrest.TypeSafeMatcher;
import org.jetbrains.annotations.Nullable;
@@ -393,6 +395,31 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertSame(schema, manager.activeSchema(clock.nowLong()));
}
+ @Test
+ void testReCreateTableWithSameName() {
+ assertThat(manager.execute(simpleTable(TABLE_NAME)), willBe(nullValue()));
+
+ int catalogVersion = manager.latestCatalogVersion();
+ CatalogTableDescriptor table1 = manager.table(TABLE_NAME, clock.nowLong());
+ assertNotNull(table1);
+
+ // Drop table.
+ assertThat(manager.execute(dropTableCommand(TABLE_NAME)), willBe(nullValue()));
+ assertNull(manager.table(TABLE_NAME, clock.nowLong()));
+
+ // Re-create table with same name.
+ assertThat(manager.execute(simpleTable(TABLE_NAME)), willBe(nullValue()));
+
+ CatalogTableDescriptor table2 = manager.table(TABLE_NAME, clock.nowLong());
+ assertNotNull(table2);
+
+ // Ensure these are different tables.
+ assertNotEquals(table1.id(), table2.id());
+
+ // Ensure table is available for historical queries.
+ manager.table(table1.id(), catalogVersion);
+ }
+
@Test
public void testAddColumn() {
assertThat(manager.execute(simpleTable(TABLE_NAME)), willBe(nullValue()));
@@ -1146,18 +1173,35 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
manager.listen(CatalogEvent.TABLE_CREATE, eventListener);
manager.listen(CatalogEvent.TABLE_DROP, eventListener);
+ manager.listen(CatalogEvent.TABLE_DESTROY, eventListener);
assertThat(manager.execute(simpleTable(TABLE_NAME)), willBe(nullValue()));
- verify(eventListener).notify(any(CreateTableEventParameters.class), isNull());
+ assertThat(manager.execute(simpleTable(TABLE_NAME_2)), willBe(nullValue()));
+ assertThat(manager.execute(simpleTable(TABLE_NAME_3)), willBe(nullValue()));
+ verify(eventListener, times(3)).notify(any(CreateTableEventParameters.class), isNull());
assertThat(manager.execute(dropTableCommand(TABLE_NAME)), willBe(nullValue()));
- verify(eventListener).notify(any(DropTableEventParameters.class), isNull());
+ assertThat(manager.execute(dropTableCommand(TABLE_NAME_2)), willBe(nullValue()));
+ verify(eventListener, times(2)).notify(any(DropTableEventParameters.class), isNull());
+
+ verifyNoMoreInteractions(eventListener);
+ clearInvocations(eventListener);
+ // Got 'destroy' event only after Catalog compaction.
+ assertThat(CatalogTestUtils.waitCatalogCompaction(manager, clock.nowLong()), equalTo(true));
+ verify(eventListener, times(2)).notify(any(DestroyTableEventParameters.class), isNull());
+
+ verifyNoMoreInteractions(eventListener);
+ clearInvocations();
+
+ // Expect no events if Catalog wasn't compacted.
+ assertThat(CatalogTestUtils.waitCatalogCompaction(manager, clock.nowLong()), equalTo(false));
verifyNoMoreInteractions(eventListener);
}
+ //TODO IGNITE-20121 check correct events fired when dropping index in various states.
@Test
- public void testCreateIndexEvents() {
+ public void testIndexEvents() {
CatalogCommand createIndexCmd = createHashIndexCommand(INDEX_NAME, List.of("ID"));
CatalogCommand dropIndexCmd = DropIndexCommand.builder().schemaName(SCHEMA_NAME).indexName(INDEX_NAME).build();
@@ -1170,6 +1214,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
manager.listen(CatalogEvent.INDEX_AVAILABLE, eventListener);
manager.listen(CatalogEvent.INDEX_STOPPING, eventListener);
manager.listen(CatalogEvent.INDEX_REMOVED, eventListener);
+ manager.listen(CatalogEvent.INDEX_DESTROY, eventListener);
// Try to create index without table.
assertThat(manager.execute(createIndexCmd), willThrow(TableNotFoundValidationException.class));
@@ -1178,7 +1223,9 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
// Create table with PK index.
assertThat(manager.execute(simpleTable(TABLE_NAME)), willCompleteSuccessfully());
verify(eventListener).notify(any(CreateIndexEventParameters.class), isNull());
+ verify(eventListener).notify(any(MakeIndexAvailableEventParameters.class), isNull());
+ verifyNoMoreInteractions(eventListener);
clearInvocations(eventListener);
// Create index.
@@ -1191,6 +1238,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
makeIndexAvailable(indexId(INDEX_NAME));
verify(eventListener).notify(any(MakeIndexAvailableEventParameters.class), isNull());
+ verifyNoMoreInteractions(eventListener);
clearInvocations(eventListener);
// Drop index.
@@ -1201,6 +1249,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
removeIndex(indexId(INDEX_NAME));
verify(eventListener).notify(any(RemoveIndexEventParameters.class), isNull());
+ verifyNoMoreInteractions(eventListener);
clearInvocations(eventListener);
// Drop table with pk index.
@@ -1210,6 +1259,18 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertThat(manager.execute(dropIndexCmd), willThrow(IndexNotFoundValidationException.class));
verify(eventListener).notify(any(RemoveIndexEventParameters.class), isNull());
+ verifyNoMoreInteractions(eventListener);
+ clearInvocations(eventListener);
+
+ // Got 'destroy' event only after Catalog compaction.
+ assertThat(CatalogTestUtils.waitCatalogCompaction(manager, clock.nowLong()), equalTo(true));
+ verify(eventListener, times(2 /* PK + secondary index */)).notify(any(DestroyIndexEventParameters.class), isNull());
+
+ clearInvocations();
+
+ // Expect no events if Catalog wasn't compacted.
+ assertThat(CatalogTestUtils.waitCatalogCompaction(manager, clock.nowLong()), equalTo(false));
+ verifyNoMoreInteractions(eventListener);
}
@Test
@@ -1619,6 +1680,41 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertEquals(pkIndexId, pkRemovedEventParameters.indexId());
}
+ @Test
+ void testReCreateIndexWithSameName() {
+ createSomeTable(TABLE_NAME);
+ createSomeIndex(TABLE_NAME, INDEX_NAME);
+
+ int catalogVersion = manager.latestCatalogVersion();
+ CatalogIndexDescriptor index1 = manager.index(INDEX_NAME, clock.nowLong());
+ assertNotNull(index1);
+
+ int indexId1 = index1.id();
+ startBuildingIndex(indexId1);
+ makeIndexAvailable(indexId1);
+
+ // Drop index.
+ dropIndex(INDEX_NAME);
+ removeIndex(indexId1);
+ assertNull(manager.index(INDEX_NAME, clock.nowLong()));
+
+ // Re-create index with same name.
+ createSomeSortedIndex(TABLE_NAME, INDEX_NAME);
+
+ CatalogIndexDescriptor index2 = manager.index(INDEX_NAME, clock.nowLong());
+ assertNotNull(index2);
+ assertThat(index2.indexType(), equalTo(CatalogIndexDescriptorType.SORTED));
+
+ // Ensure these are different indexes.
+ int indexId2 = index2.id();
+ assertNotEquals(indexId1, indexId2);
+
+ // Ensure dropped index is available for historical queries.
+ assertNotNull(manager.index(indexId1, catalogVersion));
+ assertThat(manager.index(indexId1, catalogVersion).indexType(), equalTo(CatalogIndexDescriptorType.HASH));
+ assertNull(manager.index(indexId2, catalogVersion));
+ }
+
@Test
void testLatestCatalogVersion() {
assertEquals(0, manager.latestCatalogVersion());
@@ -2457,7 +2553,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
}
@Test
- public void testCatalogCompaction() throws InterruptedException {
+ public void testCatalogCompaction() {
assertThat(manager.execute(simpleTable(TABLE_NAME)), willBe(nullValue()));
assertThat(manager.execute(simpleTable(TABLE_NAME_2)), willBe(nullValue()));
@@ -2468,9 +2564,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertThat(manager.execute(simpleIndex(TABLE_NAME, INDEX_NAME)), willBe(nullValue()));
assertThat(manager.execute(simpleIndex(TABLE_NAME, INDEX_NAME_2)), willBe(nullValue()));
- assertThat(manager.compactCatalog(timestamp), willCompleteSuccessfully());
-
- IgniteTestUtils.waitForCondition(() -> manager.earliestCatalogVersion() != 0, 2_000);
+ assertThat(CatalogTestUtils.waitCatalogCompaction(manager, timestamp), equalTo(true));
assertEquals(catalog.version(), manager.earliestCatalogVersion());
@@ -2482,6 +2576,9 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertThrows(IllegalStateException.class, () -> manager.activeCatalogVersion(catalog.time() - 1));
assertSame(catalog.version(), manager.activeCatalogVersion(catalog.time()));
assertSame(catalog.version(), manager.activeCatalogVersion(timestamp));
+
+ assertThat(manager.compactCatalog(timestamp), willBe(false));
+ assertEquals(catalog.version(), manager.earliestCatalogVersion());
}
@Test
@@ -2490,7 +2587,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
long timestamp = clock.nowLong();
- assertThat(manager.compactCatalog(timestamp), willCompleteSuccessfully());
+ assertThat(manager.compactCatalog(timestamp), willBe(false));
assertEquals(0, manager.earliestCatalogVersion());
assertEquals(0, manager.latestCatalogVersion());
diff --git a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
index b02ccad23f..988d101d5e 100644
--- a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
+++ b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/BaseCatalogManagerTest.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.catalog.commands.CreateHashIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateSortedIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
import org.apache.ignite.internal.catalog.commands.CreateTableCommandBuilder;
+import org.apache.ignite.internal.catalog.commands.DropIndexCommand;
import org.apache.ignite.internal.catalog.commands.DropTableCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
@@ -221,4 +222,18 @@ public abstract class BaseCatalogManagerTest extends BaseIgniteAbstractTest {
protected static CatalogCommand startBuildingIndexCommand(int indexId) {
return StartBuildingIndexCommand.builder().indexId(indexId).build();
}
+
+ protected static CatalogCommand dropTableCommand(String tableName) {
+ return DropTableCommand.builder()
+ .schemaName(DEFAULT_SCHEMA_NAME)
+ .tableName(tableName)
+ .build();
+ }
+
+ protected static CatalogCommand dropIndexCommand(String indexName) {
+ return DropIndexCommand.builder()
+ .schemaName(DEFAULT_SCHEMA_NAME)
+ .indexName(indexName)
+ .build();
+ }
}
diff --git a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
index 1b28c39c28..a06467a0de 100644
--- a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
+++ b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
@@ -42,9 +42,12 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.storage.SnapshotEntry;
import org.apache.ignite.internal.catalog.storage.UpdateLog;
+import org.apache.ignite.internal.catalog.storage.UpdateLog.OnUpdateHandler;
+import org.apache.ignite.internal.catalog.storage.UpdateLogEvent;
import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
import org.apache.ignite.internal.catalog.storage.VersionedUpdate;
import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
@@ -113,7 +116,6 @@ public class CatalogTestUtils {
public void beforeNodeStop() {
super.beforeNodeStop();
- clockWaiter.beforeNodeStop();
metastore.beforeNodeStop();
}
@@ -121,7 +123,6 @@ public class CatalogTestUtils {
public void stop() throws Exception {
super.stop();
- clockWaiter.stop();
metastore.stop();
}
};
@@ -161,6 +162,55 @@ public class CatalogTestUtils {
};
}
+ /**
+ * Creates a test implementation of {@link CatalogManager}.
+ *
+ * <p>NOTE: Uses {@link CatalogManagerImpl} under the hood and creates the internals it needs, may change in the future.
+ *
+ * @param nodeName Node name.
+ * @param clock Hybrid clock.
+ * @param metastore Meta storage manager.
+ */
+ public static CatalogManager createTestCatalogManagerWithInterceptor(
+ String nodeName,
+ HybridClock clock,
+ MetaStorageManager metastore,
+ UpdateHandlerInterceptor interceptor
+ ) {
+ var clockWaiter = new ClockWaiter(nodeName, clock);
+
+ UpdateLogImpl updateLog = new UpdateLogImpl(metastore) {
+ @Override
+ public void registerUpdateHandler(OnUpdateHandler handler) {
+ interceptor.registerUpdateHandler(handler);
+
+ super.registerUpdateHandler(interceptor);
+ }
+ };
+
+
+ return new CatalogManagerImpl(updateLog, clockWaiter) {
+ @Override
+ public CompletableFuture<Void> start() {
+ return allOf(clockWaiter.start(), super.start());
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ super.beforeNodeStop();
+
+ clockWaiter.beforeNodeStop();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ super.stop();
+
+ clockWaiter.stop();
+ }
+ };
+ }
+
/**
* Create the same {@link CatalogManager} as for normal operations, but with {@link UpdateLog} that
* simply notifies the manager without storing any updates in metastore.
@@ -300,7 +350,7 @@ public class CatalogTestUtils {
* @param timestamp Timestamp catalog should be compacted up to.
* @return {@code True} if a new snapshot has been successfully written, {@code false} otherwise.
*/
- static boolean waitCatalogCompaction(CatalogManager catalogManager, long timestamp) {
+ public static boolean waitCatalogCompaction(CatalogManager catalogManager, long timestamp) {
int version = catalogManager.activeCatalogVersion(timestamp);
CompletableFuture<Boolean> operationFuture = ((CatalogManagerImpl) catalogManager).compactCatalog(timestamp);
@@ -422,4 +472,39 @@ public class CatalogTestUtils {
.findFirst()
.orElse(null);
}
+
+ /**
+ * Update handler interceptor for test purposes.
+ */
+ public abstract static class UpdateHandlerInterceptor implements OnUpdateHandler {
+ private OnUpdateHandler delegate;
+
+ void registerUpdateHandler(OnUpdateHandler handler) {
+ this.delegate = handler;
+ }
+
+ protected OnUpdateHandler delegate() {
+ return delegate;
+ }
+ }
+
+ /**
+ * An interceptor, which allow dropping events.
+ */
+ public static class TestUpdateHandlerInterceptor extends UpdateHandlerInterceptor {
+ private volatile boolean dropEvents;
+
+ public void dropSnapshotEvents() {
+ this.dropEvents = true;
+ }
+
+ @Override
+ public CompletableFuture<Void> handle(UpdateLogEvent update, HybridTimestamp metaStorageUpdateTimestamp, long causalityToken) {
+ if (dropEvents && update instanceof SnapshotEntry) {
+ return nullCompletedFuture();
+ }
+
+ return delegate().handle(update, metaStorageUpdateTimestamp, causalityToken);
+ }
+ }
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
index 6fa4e74c36..156c020722 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
@@ -35,7 +35,9 @@ import static org.apache.ignite.sql.ColumnType.INT64;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -351,6 +353,37 @@ public class TableManagerTest extends IgniteAbstractTest {
assertEquals(0, tableManager.tables().size());
}
+ /**
+ * Tests create a table through public API right after another table with the same name was dropped.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReCreateTableWithSameName() throws Exception {
+ mockManagersAndCreateTable(DYNAMIC_TABLE_NAME, tblManagerFut);
+
+ TableManager tableManager = tblManagerFut.join();
+
+ TableViewInternal table = (TableViewInternal) tableManager.table(DYNAMIC_TABLE_NAME);
+
+ assertNotNull(table);
+
+ int oldTableId = table.tableId();
+
+ dropTable(DYNAMIC_TABLE_NAME);
+ createTable(DYNAMIC_TABLE_NAME);
+
+ table = tableManager.tableView(DYNAMIC_TABLE_NAME);
+
+ assertNotNull(table);
+ assertNotEquals(oldTableId, table.tableId());
+
+ // TODO IGNITE-20680 ensure old table is available
+ // assertNotNull(tableManager.getTable(oldTableId));
+ assertNotNull(tableManager.getTable(table.tableId()));
+ assertNotSame(tableManager.getTable(oldTableId), tableManager.getTable(table.tableId()));
+ }
+
/**
* Tests a work of the public API for Table manager {@see org.apache.ignite.table.manager.IgniteTables} when the manager is stopping.
*/