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/19 12:23:19 UTC
(ignite-3) branch main updated: IGNITE-21535 Create catalog events for destroying table and index (#3221)
This is an automated email from the ASF dual-hosted git repository.
amashenkov 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 5343d1274f IGNITE-21535 Create catalog events for destroying table and index (#3221)
5343d1274f is described below
commit 5343d1274f0f17c63884ecfb19a71cc90ee35d0a
Author: Andrew V. Mashenkov <AM...@users.noreply.github.com>
AuthorDate: Mon Feb 19 15:23:14 2024 +0300
IGNITE-21535 Create catalog events for destroying table and index (#3221)
---
.../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 +-
.../catalog/CatalogManagerRecoveryTest.java | 73 +++++++++----
.../internal/catalog/CatalogManagerSelfTest.java | 115 ++++++++++++++++++--
.../internal/catalog/BaseCatalogManagerTest.java | 17 +++
.../ignite/internal/catalog/CatalogTestUtils.java | 120 ++++++++++++++++++++-
.../ignite/internal/index/IndexManagerTest.java | 38 +++++++
.../table/distributed/TableManagerTest.java | 33 ++++++
19 files changed, 501 insertions(+), 164 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..64e4313429 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 catalog, int tableId) {
+ return Objects.requireNonNull(catalog.zone(Objects.requireNonNull(catalog.table(tableId), "table").zoneId()), "zone");
+ }
+
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/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
index dda29e1b80..86be7eabf2 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 3f296fe792..7917376854 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.
+ assertNotNull(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 360e324528..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,8 @@ 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;
import org.apache.ignite.internal.catalog.storage.UpdateLog;
@@ -61,6 +63,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";
@@ -219,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 07f82632c7..10e6dd46eb 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;
@@ -39,14 +42,16 @@ 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;
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;
@@ -111,7 +116,6 @@ public class CatalogTestUtils {
public void beforeNodeStop() {
super.beforeNodeStop();
- clockWaiter.beforeNodeStop();
metastore.beforeNodeStop();
}
@@ -119,7 +123,6 @@ public class CatalogTestUtils {
public void stop() throws Exception {
super.stop();
- clockWaiter.stop();
metastore.stop();
}
};
@@ -159,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.
@@ -291,6 +343,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.
+ */
+ public 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 +394,7 @@ public class CatalogTestUtils {
@Override
public synchronized CompletableFuture<Boolean> saveSnapshot(SnapshotEntry snapshotEntry) {
snapshotVersion = snapshotEntry.version();
- return CompletableFutures.trueCompletedFuture();
+ return trueCompletedFuture();
}
@Override
@@ -395,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/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);
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.
*/