You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vp...@apache.org on 2022/07/07 12:43:23 UTC
[ignite-3] branch main updated: IGNITE-17062 Add ability to process local events asynchronously (#842)
This is an automated email from the ASF dual-hosted git repository.
vpyatkov 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 dac5eea8a IGNITE-17062 Add ability to process local events asynchronously (#842)
dac5eea8a is described below
commit dac5eea8a7e068e70759f21dc180b595c9b7038d
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Thu Jul 7 15:43:18 2022 +0300
IGNITE-17062 Add ability to process local events asynchronously (#842)
---
.../ignite/internal/manager/EventListener.java | 7 +-
.../apache/ignite/internal/manager/Producer.java | 68 +++++++--
.../ignite/internal/manager/ProducerTest.java | 164 +++++++++++++++++++++
.../storage/ItRebalanceDistributedTest.java | 10 +-
.../runner/app/ItIgniteNodeRestartTest.java | 25 ++--
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
.../recovery/ConfigurationCatchUpListener.java | 8 +-
.../ignite/internal/schema/SchemaManager.java | 18 ++-
.../internal/sql/engine/SqlQueryProcessor.java | 52 +++----
.../sql/engine/schema/SqlSchemaManagerImpl.java | 72 ++++++---
.../internal/sql/engine/StopCalciteModuleTest.java | 9 ++
.../sql/engine/exec/MockedStructuresTest.java | 12 +-
.../engine/exec/schema/SqlSchemaManagerTest.java | 59 ++++----
.../apache/ignite/internal/table/TableImpl.java | 4 +-
.../internal/table/distributed/TableManager.java | 65 +++++---
.../table/distributed/TableManagerTest.java | 22 +--
16 files changed, 435 insertions(+), 163 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/manager/EventListener.java b/modules/core/src/main/java/org/apache/ignite/internal/manager/EventListener.java
index 0f8fdb474..66f253c3a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/manager/EventListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/manager/EventListener.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.manager;
+import java.util.concurrent.CompletableFuture;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -32,9 +33,11 @@ public interface EventListener<P extends EventParameters> {
*
* @param parameters Parameters provide a properties of the event. This attribute cannot be {@code null}.
* @param exception Exception which is happened during the event produced or {@code null}.
- * @return {@code True} means that the event is handled and a listener will be removed, {@code false} is the listener will stay listen.
+ * @return Completable future, which is completed when event handling is finished. The {@code True} value of the future
+ * means that the event is handled and a listener will be removed, {@code false} is the listener will stay listen.
+ * This future must not be completed with {@code null} value.
*/
- boolean notify(@NotNull P parameters, @Nullable Throwable exception);
+ CompletableFuture<Boolean> notify(@NotNull P parameters, @Nullable Throwable exception);
/**
* Removes an listener from the event producer. When the listener was removed it never receive a notification any more.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/manager/Producer.java b/modules/core/src/main/java/org/apache/ignite/internal/manager/Producer.java
index 665b9d9dc..0292772f6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/manager/Producer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/manager/Producer.java
@@ -17,18 +17,25 @@
package org.apache.ignite.internal.manager;
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.jetbrains.annotations.Nullable;
/**
- * Interface which can produce its events.
+ * A class which can produce its events.
*/
public abstract class Producer<T extends Event, P extends EventParameters> {
/** All listeners. */
- private ConcurrentHashMap<T, ConcurrentLinkedQueue<EventListener<P>>> listeners = new ConcurrentHashMap<>();
+ private ConcurrentHashMap<T, List<EventListener<P>>> listeners = new ConcurrentHashMap<>();
/**
* Registers an event listener. When the event predicate returns true it would never invoke after, otherwise this predicate would
@@ -38,7 +45,7 @@ public abstract class Producer<T extends Event, P extends EventParameters> {
* @param closure Closure.
*/
public void listen(T evt, EventListener<P> closure) {
- listeners.computeIfAbsent(evt, evtKey -> new ConcurrentLinkedQueue<>()).offer(closure);
+ listeners.computeIfAbsent(evt, evtKey -> new CopyOnWriteArrayList<>()).add(closure);
}
/**
@@ -59,7 +66,19 @@ public abstract class Producer<T extends Event, P extends EventParameters> {
* @param cause The exception that was a cause which a listener is removed.
*/
public void removeListener(T evt, EventListener<P> closure, @Nullable IgniteInternalCheckedException cause) {
- if (listeners.computeIfAbsent(evt, evtKey -> new ConcurrentLinkedQueue<>()).remove(closure)) {
+ removeListener(evt, closure, cause, true);
+ }
+
+ /**
+ * Removes a listener associated with the event.
+ *
+ * @param evt Event.
+ * @param closure Closure.
+ * @param cause The exception that was a cause which a listener is removed.
+ * @param callOnRemove Whether to call {@link EventListener#remove(Throwable)} callback on the closure.
+ */
+ private void removeListener(T evt, EventListener<P> closure, @Nullable IgniteInternalCheckedException cause, boolean callOnRemove) {
+ if (listeners.computeIfAbsent(evt, evtKey -> new CopyOnWriteArrayList<>()).remove(closure) && callOnRemove) {
closure.remove(cause == null ? new ListenerRemovedException() : cause.getCause() == null ? cause : cause.getCause());
}
}
@@ -70,24 +89,43 @@ public abstract class Producer<T extends Event, P extends EventParameters> {
* @param evt Event type.
* @param params Event parameters.
* @param err Exception when it was happened, or {@code null} otherwise.
+ * @return Completable future which is completed when event handling is complete.
*/
- protected void fireEvent(T evt, P params, Throwable err) {
- ConcurrentLinkedQueue<EventListener<P>> queue = listeners.get(evt);
+ protected CompletableFuture<?> fireEvent(T evt, P params, Throwable err) {
+ List<EventListener<P>> eventListeners = listeners.get(evt);
- if (queue == null) {
- return;
+ if (eventListeners == null) {
+ return completedFuture(null);
}
- EventListener<P> closure;
+ Collection<CompletableFuture<?>> futures = new ArrayList<>();
- Iterator<EventListener<P>> iter = queue.iterator();
+ Iterator<EventListener<P>> iter = eventListeners.iterator();
while (iter.hasNext()) {
- closure = iter.next();
+ EventListener<P> closure = iter.next();
- if (closure.notify(params, err)) {
- iter.remove();
- }
+ CompletableFuture<?> future = closure.notify(params, err)
+ .thenAccept(b -> {
+ if (b) {
+ removeListener(evt, closure, null, false);
+ }
+ });
+
+ futures.add(future);
}
+
+ return allOf(futures.toArray(new CompletableFuture[0]));
+ }
+
+ /**
+ * Notifies every listener that subscribed before.
+ *
+ * @param evt Event type.
+ * @param params Event parameters.
+ * @return Completable future which is completed when event handling is complete.
+ */
+ protected CompletableFuture<?> fireEvent(T evt, P params) {
+ return fireEvent(evt, params, null);
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/manager/ProducerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/manager/ProducerTest.java
new file mode 100644
index 000000000..52dfb57c5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/manager/ProducerTest.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.manager;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.runAsync;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test of some implementation of {@link Producer} and {@link EventListener}.
+ */
+public class ProducerTest {
+ @Test
+ public void simpleAsyncTest() {
+ Producer<TestEvent, TestEventParameters> producer = new Producer<>() {
+ };
+
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+
+ producer.listen(TestEvent.TEST, (p, e) -> future);
+
+ CompletableFuture<?> eventHandleFuture = producer.fireEvent(TestEvent.TEST, new TestEventParameters(0L));
+
+ assertFalse(eventHandleFuture.isDone());
+
+ future.complete(true);
+
+ assertTrue(eventHandleFuture.isDone());
+ }
+
+ @Test
+ public void stopListenTest() {
+ Producer<TestEvent, TestEventParameters> producer = new Producer<>() {
+ };
+
+ final int stopListenAfterCount = 5;
+ final int fireEventCount = stopListenAfterCount * 2;
+ AtomicInteger listenConter = new AtomicInteger();
+ AtomicInteger removeCounter = new AtomicInteger();
+
+ EventListener<TestEventParameters> listener = createEventListener(
+ (p, e) -> completedFuture(listenConter.incrementAndGet() == stopListenAfterCount),
+ t -> removeCounter.incrementAndGet()
+ );
+
+ producer.listen(TestEvent.TEST, listener);
+
+ for (int i = 0; i < fireEventCount; i++) {
+ producer.fireEvent(TestEvent.TEST, new TestEventParameters(0L));
+ }
+
+ assertEquals(stopListenAfterCount, listenConter.get());
+ assertEquals(0, removeCounter.get());
+ }
+
+ @Test
+ public void parallelTest() {
+ Producer<TestEvent, TestEventParameters> producer = new Producer<>() {
+ };
+
+ final int listenersCount = 10_000;
+ final int listenerIndexToRemove = 10;
+ EventListener<TestEventParameters> listenerToRemove = null;
+
+ CompletableFuture toRemoveFuture = new CompletableFuture();
+
+ for (int i = 0; i < listenersCount; i++) {
+ EventListener<TestEventParameters> listener = i == listenerIndexToRemove
+ ? createEventListener(
+ (p, e) -> {
+ toRemoveFuture.complete(null);
+
+ return completedFuture(false);
+ },
+ t -> {}
+ )
+ : createEventListener((p, e) -> completedFuture(false), t -> {});
+
+ if (i == listenerIndexToRemove) {
+ listenerToRemove = listener;
+ }
+
+ producer.listen(TestEvent.TEST, listener);
+ }
+
+ CompletableFuture fireFuture = runAsync(() -> producer.fireEvent(TestEvent.TEST, new TestEventParameters(0L)).join());
+
+ toRemoveFuture.join();
+
+ producer.removeListener(TestEvent.TEST, listenerToRemove);
+
+ fireFuture.join();
+
+ assertFalse(fireFuture.isCompletedExceptionally());
+ }
+
+ /**
+ * Create event listener.
+ *
+ * @param notify Notify action.
+ * @param remove Remove action.
+ * @return Listener.
+ */
+ private <T extends EventParameters> EventListener<T> createEventListener(
+ BiFunction<T, Throwable, CompletableFuture<Boolean>> notify,
+ Consumer<Throwable> remove
+ ) {
+ return new EventListener<T>() {
+ @Override public CompletableFuture<Boolean> notify(@NotNull T parameters, @Nullable Throwable exception) {
+ return notify.apply(parameters, exception);
+ }
+
+ @Override public void remove(@NotNull Throwable throwable) {
+ remove.accept(throwable);
+ }
+ };
+ }
+
+ /**
+ * Test event.
+ */
+ private enum TestEvent implements Event {
+ TEST
+ }
+
+ /**
+ * Parameters for test event.
+ */
+ private class TestEventParameters extends EventParameters {
+ /**
+ * Constructor.
+ *
+ * @param causalityToken Causality token.
+ */
+ public TestEventParameters(long causalityToken) {
+ super(causalityToken);
+ }
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
index 890d02c3f..5dc972acc 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java
@@ -58,7 +58,6 @@ import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.server.impl.JraftServerImpl;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter;
-import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.storage.DataStorageManager;
import org.apache.ignite.internal.storage.DataStorageModules;
import org.apache.ignite.internal.storage.pagememory.VolatilePageMemoryDataStorageModule;
@@ -376,8 +375,6 @@ public class ItRebalanceDistributedTest {
private final SchemaManager schemaManager;
- private final SqlQueryProcessor sqlQueryProcessor;
-
/**
* Constructor that simply creates a subset of components of this node.
*/
@@ -480,9 +477,6 @@ public class ItRebalanceDistributedTest {
dataStorageMgr,
metaStorageManager,
schemaManager);
-
- //TODO: Get rid of it after IGNITE-17062.
- sqlQueryProcessor = new SqlQueryProcessor(registry, clusterService, tableManager, dataStorageMgr, Map::of);
}
/**
@@ -494,7 +488,7 @@ public class ItRebalanceDistributedTest {
nodeCfgMgr.start();
Stream.of(clusterService, clusterCfgMgr, dataStorageMgr, raftManager, txManager, cmgManager,
- metaStorageManager, baselineMgr, schemaManager, tableManager, sqlQueryProcessor).forEach(IgniteComponent::start);
+ metaStorageManager, baselineMgr, schemaManager, tableManager).forEach(IgniteComponent::start);
CompletableFuture.allOf(
nodeCfgMgr.configurationRegistry().notifyCurrentConfigurationListeners(),
@@ -510,7 +504,7 @@ public class ItRebalanceDistributedTest {
*/
void stop() throws Exception {
var components =
- List.of(sqlQueryProcessor, tableManager, schemaManager, baselineMgr, metaStorageManager, cmgManager, dataStorageMgr,
+ List.of(tableManager, schemaManager, baselineMgr, metaStorageManager, cmgManager, dataStorageMgr,
raftManager, txManager, clusterCfgMgr, clusterService, nodeCfgMgr, vaultManager);
for (IgniteComponent igniteComponent : components) {
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index 5f450028a..89c3e30f5 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -38,7 +38,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.ListIterator;
-import java.util.Map;
import java.util.ServiceLoader;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -72,7 +71,6 @@ import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.recovery.ConfigurationCatchUpListener;
import org.apache.ignite.internal.recovery.RecoveryCompletionFutureFactory;
import org.apache.ignite.internal.schema.SchemaManager;
-import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.storage.DataStorageManager;
import org.apache.ignite.internal.storage.DataStorageModule;
import org.apache.ignite.internal.storage.DataStorageModules;
@@ -191,7 +189,7 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
) throws NodeStoppingException {
Path dir = workDir.resolve(name);
- List<IgniteComponent> res = new ArrayList<>();
+ partialNode = new ArrayList<>();
VaultManager vault = createVault(dir);
@@ -282,13 +280,10 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
schemaManager
);
- //TODO: Get rid of it after IGNITE-17062.
- SqlQueryProcessor queryProcessor = new SqlQueryProcessor(registry, clusterSvc, tableManager, dataStorageManager, Map::of);
-
// Preparing the result map.
- res.add(vault);
- res.add(nodeCfgMgr);
+ partialNode.add(vault);
+ partialNode.add(nodeCfgMgr);
// Start.
@@ -319,14 +314,13 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
clusterCfgMgr,
dataStorageManager,
schemaManager,
- tableManager,
- queryProcessor
+ tableManager
);
for (IgniteComponent component : otherComponents) {
component.start();
- res.add(component);
+ partialNode.add(component);
}
AtomicLong lastRevision = new AtomicLong();
@@ -367,9 +361,7 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
+ ", acceptableDifference: " + IgniteSystemProperties.getInteger(CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY, 100)
);
- partialNode = res;
-
- return res;
+ return partialNode;
}
/**
@@ -875,6 +867,7 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
stopNode(nodes - 1);
createTableWithData(CLUSTER_NODES.get(0), TABLE_NAME_2, nodes);
+ createTableWithData(CLUSTER_NODES.get(0), TABLE_NAME_2 + "0", nodes);
log.info("Starting the node.");
@@ -928,6 +921,8 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
stopNode(0);
+ log.info("Starting METASTORAGE");
+
startNode(testInfo, 0);
log.info("Restarted METASTORAGE");
@@ -1035,7 +1030,7 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest {
Table table = ignite.tables().createTable(
scmTbl1.canonicalName(),
- tbl -> convert(scmTbl1, tbl).changePartitions(10).changeReplicas(replicas).changePartitions(partitions)
+ tbl -> convert(scmTbl1, tbl).changeReplicas(replicas).changePartitions(partitions)
);
for (int i = 0; i < 100; i++) {
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 9657393fa..193786d27 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -325,6 +325,7 @@ public class IgniteImpl implements Ignite {
registry,
clusterSvc,
distributedTblMgr,
+ schemaManager,
dataStorageMgr,
() -> dataStorageModules.collectSchemasFields(modules.distributed().polymorphicSchemaExtensions())
);
@@ -451,7 +452,7 @@ public class IgniteImpl implements Ignite {
LOG.info("Components started, performing recovery");
// Recovery future must be created before configuration listeners are triggered.
- CompletableFuture<Void> recoveryFuture = RecoveryCompletionFutureFactory.create(
+ CompletableFuture<?> recoveryFuture = RecoveryCompletionFutureFactory.create(
clusterCfgMgr,
fut -> new ConfigurationCatchUpListener(cfgStorage, fut, LOG)
);
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/recovery/ConfigurationCatchUpListener.java b/modules/runner/src/main/java/org/apache/ignite/internal/recovery/ConfigurationCatchUpListener.java
index 722861f2c..74405c142 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/recovery/ConfigurationCatchUpListener.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/recovery/ConfigurationCatchUpListener.java
@@ -81,8 +81,8 @@ public class ConfigurationCatchUpListener implements ConfigurationStorageRevisio
*
* @param appliedRevision Applied revision.
*/
- private void checkRevisionUpToDate(long appliedRevision) {
- cfgStorage.lastRevision().thenAccept(rev -> {
+ private CompletableFuture<?> checkRevisionUpToDate(long appliedRevision) {
+ return cfgStorage.lastRevision().thenAccept(rev -> {
synchronized (targetRevisionUpdateMutex) {
assert rev >= appliedRevision : IgniteStringFormatter.format(
"Configuration revision must be greater than local node applied revision [msRev={}, appliedRev={}",
@@ -109,10 +109,10 @@ public class ConfigurationCatchUpListener implements ConfigurationStorageRevisio
if (targetRev >= 0) {
if (isConfigurationUpToDate(targetRev, appliedRevision)) {
- checkRevisionUpToDate(appliedRevision);
+ return checkRevisionUpToDate(appliedRevision);
}
} else {
- checkRevisionUpToDate(appliedRevision);
+ return checkRevisionUpToDate(appliedRevision);
}
return CompletableFuture.completedFuture(null);
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
index c4e7dd04a..b03c77eb2 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java
@@ -120,7 +120,7 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
CompletableFuture<?> createSchemaFut = createSchema(causalityToken, tblId, tableName, schemaDescriptor);
registriesVv.get(causalityToken)
- .thenRun(() -> fireEvent(SchemaEvent.CREATE, new SchemaEventParameters(causalityToken, tblId, schemaDescriptor), null));
+ .thenRun(() -> fireEvent(SchemaEvent.CREATE, new SchemaEventParameters(causalityToken, tblId, schemaDescriptor)));
return createSchemaFut;
}
@@ -240,14 +240,14 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
var clo = new EventListener<SchemaEventParameters>() {
@Override
- public boolean notify(@NotNull SchemaEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(@NotNull SchemaEventParameters parameters, @Nullable Throwable exception) {
if (tblId.equals(parameters.tableId()) && schemaVer <= parameters.schemaDescriptor().version()) {
fut.complete(getSchemaDescriptorLocally(schemaVer, tblCfg));
- return true;
+ return completedFuture(true);
}
- return false;
+ return completedFuture(false);
}
@Override
@@ -346,6 +346,16 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
}
}
+ /**
+ * Returns schema registry by table id.
+ *
+ * @param tableId Table id.
+ * @return Schema registry.
+ */
+ public SchemaRegistry schemaRegistry(UUID tableId) {
+ return registriesVv.latest().get(tableId);
+ }
+
/**
* Drop schema registry for the given table id.
*
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
index b6f17f5bc..424c797a0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
@@ -41,6 +41,7 @@ import org.apache.calcite.util.Pair;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.manager.EventListener;
+import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.sql.engine.exec.ArrayRowHandler;
import org.apache.ignite.internal.sql.engine.exec.ExchangeServiceImpl;
import org.apache.ignite.internal.sql.engine.exec.ExecutionService;
@@ -87,6 +88,8 @@ public class SqlQueryProcessor implements QueryProcessor {
private final TableManager tableManager;
+ private final SchemaManager schemaManager;
+
private final Consumer<Function<Long, CompletableFuture<?>>> registry;
private final DataStorageManager dataStorageManager;
@@ -109,19 +112,21 @@ public class SqlQueryProcessor implements QueryProcessor {
private volatile PrepareService prepareSvc;
- private volatile SqlSchemaManager schemaManager;
+ private volatile SqlSchemaManager sqlSchemaManager;
/** Constructor. */
public SqlQueryProcessor(
Consumer<Function<Long, CompletableFuture<?>>> registry,
ClusterService clusterSrvc,
TableManager tableManager,
+ SchemaManager schemaManager,
DataStorageManager dataStorageManager,
Supplier<Map<String, Map<String, Class<?>>>> dataStorageFieldsSupplier
) {
this.registry = registry;
this.clusterSrvc = clusterSrvc;
this.tableManager = tableManager;
+ this.schemaManager = schemaManager;
this.dataStorageManager = dataStorageManager;
this.dataStorageFieldsSupplier = dataStorageFieldsSupplier;
}
@@ -155,16 +160,16 @@ public class SqlQueryProcessor implements QueryProcessor {
msgSrvc
));
- SqlSchemaManagerImpl schemaManager = new SqlSchemaManagerImpl(tableManager, registry);
+ SqlSchemaManagerImpl sqlSchemaManager = new SqlSchemaManagerImpl(tableManager, schemaManager, registry);
- schemaManager.registerListener(prepareSvc);
+ sqlSchemaManager.registerListener(prepareSvc);
this.prepareSvc = prepareSvc;
var executionSrvc = registerService(ExecutionServiceImpl.create(
clusterSrvc.topologyService(),
msgSrvc,
- schemaManager,
+ sqlSchemaManager,
tableManager,
taskExecutor,
ArrayRowHandler.INSTANCE,
@@ -178,11 +183,11 @@ public class SqlQueryProcessor implements QueryProcessor {
this.executionSrvc = executionSrvc;
- registerTableListener(TableEvent.CREATE, new TableCreatedListener(schemaManager));
- registerTableListener(TableEvent.ALTER, new TableUpdatedListener(schemaManager));
- registerTableListener(TableEvent.DROP, new TableDroppedListener(schemaManager));
+ registerTableListener(TableEvent.CREATE, new TableCreatedListener(sqlSchemaManager));
+ registerTableListener(TableEvent.ALTER, new TableUpdatedListener(sqlSchemaManager));
+ registerTableListener(TableEvent.DROP, new TableDroppedListener(sqlSchemaManager));
- this.schemaManager = schemaManager;
+ this.sqlSchemaManager = sqlSchemaManager;
services.forEach(LifecycleAware::start);
}
@@ -290,7 +295,7 @@ public class SqlQueryProcessor implements QueryProcessor {
var schemaName = session.queryProperties().get(QueryProperty.DEFAULT_SCHEMA);
- SchemaPlus schema = schemaManager.schema(schemaName);
+ SchemaPlus schema = sqlSchemaManager.schema(schemaName);
if (schema == null) {
return CompletableFuture.failedFuture(new IgniteInternalException(format("Schema not found [schemaName={}]", schemaName)));
@@ -378,7 +383,7 @@ public class SqlQueryProcessor implements QueryProcessor {
String sql,
Object... params
) {
- SchemaPlus schema = schemaManager.schema(schemaName);
+ SchemaPlus schema = sqlSchemaManager.schema(schemaName);
if (schema == null) {
throw new IgniteInternalException(format("Schema not found [schemaName={}]", schemaName));
@@ -447,14 +452,13 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public boolean notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- schemaHolder.onTableCreated(
+ public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ return schemaHolder.onTableCreated(
"PUBLIC",
parameters.table(),
parameters.causalityToken()
- );
-
- return false;
+ )
+ .thenApply(v -> false);
}
}
@@ -467,14 +471,13 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public boolean notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- schemaHolder.onTableUpdated(
+ public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ return schemaHolder.onTableUpdated(
"PUBLIC",
parameters.table(),
parameters.causalityToken()
- );
-
- return false;
+ )
+ .thenApply(v -> false);
}
}
@@ -487,14 +490,13 @@ public class SqlQueryProcessor implements QueryProcessor {
/** {@inheritDoc} */
@Override
- public boolean notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- schemaHolder.onTableDropped(
+ public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
+ return schemaHolder.onTableDropped(
"PUBLIC",
parameters.tableName(),
parameters.causalityToken()
- );
-
- return false;
+ )
+ .thenApply(v -> false);
}
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
index b61aac62b..a334b17c1 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
@@ -36,6 +36,8 @@ import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.causality.VersionedValue;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaManager;
+import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.lang.IgniteInternalException;
@@ -56,6 +58,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
private final TableManager tableManager;
+ private final SchemaManager schemaManager;
+
private final VersionedValue<SchemaPlus> calciteSchemaVv;
private final Set<SchemaUpdateListener> listeners = new CopyOnWriteArraySet<>();
@@ -66,9 +70,11 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
*/
public SqlSchemaManagerImpl(
TableManager tableManager,
+ SchemaManager schemaManager,
Consumer<Function<Long, CompletableFuture<?>>> registry
) {
this.tableManager = tableManager;
+ this.schemaManager = schemaManager;
schemasVv = new VersionedValue<>(registry, HashMap::new);
tablesVv = new VersionedValue<>(registry, HashMap::new);
@@ -79,11 +85,20 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
});
schemasVv.whenComplete((token, stringIgniteSchemaMap, throwable) -> {
- rebuild(token, stringIgniteSchemaMap);
+ if (throwable != null) {
+ calciteSchemaVv.completeExceptionally(
+ token,
+ new IgniteInternalException("Couldn't evaluate sql schemas for causality token: " + token, throwable)
+ );
+
+ return;
+ }
+
+ SchemaPlus newCalciteSchema = rebuild(stringIgniteSchemaMap);
listeners.forEach(SchemaUpdateListener::onSchemaUpdated);
- tableManager.onSqlSchemaReady(token);
+ calciteSchemaVv.complete(token, newCalciteSchema);
});
}
@@ -192,7 +207,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
* OnSqlTypeCreated.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public synchronized void onTableCreated(
+ public synchronized CompletableFuture<?> onTableCreated(
String schemaName,
TableImpl table,
long causalityToken
@@ -208,9 +223,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
- IgniteTableImpl igniteTable = convert(table);
-
- schema.addTable(removeSchema(schemaName, table.name()), igniteTable);
+ CompletableFuture<IgniteTableImpl> igniteTableFuture = convert(causalityToken, table);
return tablesVv
.update(
@@ -222,33 +235,46 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
- resTbls.put(igniteTable.id(), igniteTable);
+ return igniteTableFuture
+ .thenApply(igniteTable -> {
+ resTbls.put(igniteTable.id(), igniteTable);
- return completedFuture(resTbls);
+ return resTbls;
+ });
}
)
- .thenCompose(tables -> completedFuture(res));
+ .thenCombine(
+ igniteTableFuture,
+ (v, igniteTable) -> {
+ schema.addTable(removeSchema(schemaName, table.name()), igniteTable);
+
+ return null;
+ }
+ )
+ .thenCompose(v -> completedFuture(res));
}
);
+
+ return calciteSchemaVv.get(causalityToken);
}
/**
* OnSqlTypeUpdated.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public void onTableUpdated(
+ public CompletableFuture<?> onTableUpdated(
String schemaName,
TableImpl table,
long causalityToken
) {
- onTableCreated(schemaName, table, causalityToken);
+ return onTableCreated(schemaName, table, causalityToken);
}
/**
* OnSqlTypeDropped.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public synchronized void onTableDropped(
+ public synchronized CompletableFuture<?> onTableDropped(
String schemaName,
String tableName,
long causalityToken
@@ -290,26 +316,38 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return completedFuture(res);
}
);
+
+ return calciteSchemaVv.get(causalityToken);
}
/**
* Rebuilds Calcite schemas.
*
- * @param causalityToken Causality token.
* @param schemas Ignite schemas.
*/
- private void rebuild(long causalityToken, Map<String, IgniteSchema> schemas) {
+ private SchemaPlus rebuild(Map<String, IgniteSchema> schemas) {
SchemaPlus newCalciteSchema = Frameworks.createRootSchema(false);
newCalciteSchema.add("PUBLIC", new IgniteSchema("PUBLIC"));
schemas.forEach(newCalciteSchema::add);
- calciteSchemaVv.complete(causalityToken, newCalciteSchema);
+ return newCalciteSchema;
+ }
+
+ private CompletableFuture<IgniteTableImpl> convert(long causalityToken, TableImpl table) {
+ return schemaManager.schemaRegistry(causalityToken, table.tableId())
+ .thenApply(schemaRegistry -> convert(table, schemaRegistry));
}
private IgniteTableImpl convert(TableImpl table) {
- SchemaDescriptor descriptor = table.schemaView().schema();
+ SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(table.tableId());
+
+ return convert(table, schemaRegistry);
+ }
+
+ private IgniteTableImpl convert(TableImpl table, SchemaRegistry schemaRegistry) {
+ SchemaDescriptor descriptor = schemaRegistry.schema();
List<ColumnDescriptor> colDescriptors = descriptor.columnNames().stream()
.map(descriptor::column)
@@ -329,7 +367,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return new IgniteTableImpl(
new TableDescriptorImpl(colDescriptors),
table.internalTable(),
- table.schemaView()
+ schemaRegistry
);
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
index 0870c2a7d..617175724 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.sql.engine;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.schema.SchemaManager.INITIAL_SCHEMA_VERSION;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -25,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -56,6 +58,7 @@ import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypes;
import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.registry.SchemaRegistryImpl;
import org.apache.ignite.internal.schema.row.RowAssembler;
@@ -101,6 +104,9 @@ public class StopCalciteModuleTest {
@Mock
TableManager tableManager;
+ @Mock
+ SchemaManager schemaManager;
+
@Mock
DataStorageManager dataStorageManager;
@@ -143,6 +149,8 @@ public class StopCalciteModuleTest {
when(tbl.name()).thenReturn("PUBLIC.TEST");
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaReg));
+
// Mock create table (notify on register listener).
doAnswer(invocation -> {
EventListener<TableEventParameters> clo = (EventListener<TableEventParameters>) invocation.getArguments()[1];
@@ -196,6 +204,7 @@ public class StopCalciteModuleTest {
testRevisionRegister,
clusterSrvc,
tableManager,
+ schemaManager,
dataStorageManager,
Map::of
);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
index d524832a8..abf3f5dfc 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
@@ -182,6 +182,8 @@ public class MockedStructuresTest extends IgniteAbstractTest {
DataStorageManager dataStorageManager;
+ SchemaManager schemaManager;
+
/** Returns current method name. */
private static String getCurrentMethodName() {
return StackWalker.getInstance()
@@ -235,12 +237,17 @@ public class MockedStructuresTest extends IgniteAbstractTest {
dataStorageManager.start();
+ schemaManager = new SchemaManager(revisionUpdater, tblsCfg);
+
+ schemaManager.start();
+
tblManager = mockManagers();
queryProc = new SqlQueryProcessor(
revisionUpdater,
cs,
tblManager,
+ schemaManager,
dataStorageManager,
() -> dataStorageModules.collectSchemasFields(List.of(
RocksDbDataStorageConfigurationSchema.class,
@@ -686,8 +693,6 @@ public class MockedStructuresTest extends IgniteAbstractTest {
}
private TableManager createTableManager() {
- SchemaManager sm = new SchemaManager(revisionUpdater, tblsCfg);
-
TableManager tableManager = new TableManager(
revisionUpdater,
tblsCfg,
@@ -697,10 +702,9 @@ public class MockedStructuresTest extends IgniteAbstractTest {
tm,
dataStorageManager,
msm,
- sm
+ schemaManager
);
- sm.start();
tableManager.start();
return tableManager;
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
index 58f8f20cb..3496b2dff 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.sql.engine.exec.schema;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -25,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -40,6 +42,7 @@ import org.apache.calcite.schema.Table;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypes;
import org.apache.ignite.internal.schema.SchemaDescriptor;
+import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.registry.SchemaRegistryImpl;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.IgniteTableImpl;
@@ -74,13 +77,16 @@ public class SqlSchemaManagerTest {
@Mock
private TableManager tableManager;
+ @Mock
+ private SchemaManager schemaManager;
+
@Mock
private TableImpl table;
@Mock
SchemaRegistryImpl schemaRegistry;
- private SqlSchemaManagerImpl schemaManager;
+ private SqlSchemaManagerImpl sqlSchemaManager;
private TestRevisionRegister testRevisionRegister;
@@ -90,8 +96,9 @@ public class SqlSchemaManagerTest {
testRevisionRegister = new TestRevisionRegister();
- schemaManager = new SqlSchemaManagerImpl(
+ sqlSchemaManager = new SqlSchemaManagerImpl(
tableManager,
+ schemaManager,
testRevisionRegister
);
@@ -102,13 +109,11 @@ public class SqlSchemaManagerTest {
public void testNonExistingTable() throws NodeStoppingException {
UUID tblId = UUID.randomUUID();
- IgniteInternalException ex = assertThrows(IgniteInternalException.class, () -> schemaManager.tableById(tblId, tableVer));
+ IgniteInternalException ex = assertThrows(IgniteInternalException.class, () -> sqlSchemaManager.tableById(tblId, tableVer));
assertThat(ex.getMessage(), containsString("Table not found"));
Mockito.verify(tableManager).table(eq(tblId));
- Mockito.verify(tableManager, times(1)).onSqlSchemaReady(anyLong());
-
Mockito.verifyNoMoreInteractions(tableManager);
}
@@ -124,20 +129,19 @@ public class SqlSchemaManagerTest {
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- IgniteTable actTable = schemaManager.tableById(tableId, tableVer);
+ when(schemaManager.schemaRegistry(any())).thenReturn(schemaRegistry);
+
+ IgniteTable actTable = sqlSchemaManager.tableById(tableId, tableVer);
assertEquals(tableId, actTable.id());
Mockito.verify(tableManager).table(eq(tableId));
- Mockito.verify(tableManager, times(1)).onSqlSchemaReady(anyLong());
-
Mockito.verifyNoMoreInteractions(tableManager);
}
@Test
public void testTableEventIsProcessedRequiredVersionIsSame() {
- when(table.schemaView()).thenReturn(schemaRegistry);
when(table.name()).thenReturn("TEST_SCHEMA.T");
InternalTable mock = mock(InternalTable.class);
@@ -147,21 +151,20 @@ public class SqlSchemaManagerTest {
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- schemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
testRevisionRegister.moveForward();
- IgniteTable actTable = schemaManager.tableById(tableId, tableVer);
+ IgniteTable actTable = sqlSchemaManager.tableById(tableId, tableVer);
assertEquals(tableId, actTable.id());
- Mockito.verify(tableManager, times(2)).onSqlSchemaReady(anyLong());
-
Mockito.verifyNoMoreInteractions(tableManager);
}
@Test
public void testTableEventIsProcessedRequiredVersionIsLess() {
- when(table.schemaView()).thenReturn(schemaRegistry);
when(table.name()).thenReturn("TEST_SCHEMA.T");
InternalTable mock = mock(InternalTable.class);
@@ -171,15 +174,15 @@ public class SqlSchemaManagerTest {
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- schemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
testRevisionRegister.moveForward();
- IgniteTable actTable = schemaManager.tableById(tableId, tableVer - 1);
+ IgniteTable actTable = sqlSchemaManager.tableById(tableId, tableVer - 1);
assertEquals(tableId, actTable.id());
- Mockito.verify(tableManager, times(2)).onSqlSchemaReady(anyLong());
-
Mockito.verifyNoMoreInteractions(tableManager);
}
@@ -194,28 +197,28 @@ public class SqlSchemaManagerTest {
when(table.internalTable()).thenReturn(mock);
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
when(schemaRegistry.lastSchemaVersion()).thenReturn(tableVer - 1);
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+ when(schemaManager.schemaRegistry(any())).thenReturn(schemaRegistry);
- schemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
testRevisionRegister.moveForward();
when(tableManager.table(eq(tableId))).thenReturn(table);
when(schemaRegistry.lastSchemaVersion()).thenReturn(tableVer);
- IgniteTable actTable = schemaManager.tableById(tableId, tableVer);
+ IgniteTable actTable = sqlSchemaManager.tableById(tableId, tableVer);
assertEquals(tableId, actTable.id());
- IgniteInternalException ex = assertThrows(IgniteInternalException.class, () -> schemaManager.tableById(tableId, tableVer + 1));
+ IgniteInternalException ex = assertThrows(IgniteInternalException.class, () -> sqlSchemaManager.tableById(tableId, tableVer + 1));
assertThat(ex.getMessage(), containsString("Table version not found"));
Mockito.verify(tableManager, times(2)).table(eq(tableId));
- Mockito.verify(tableManager, times(2)).onSqlSchemaReady(anyLong());
Mockito.verifyNoMoreInteractions(tableManager);
}
@Test
public void testOnTableDroppedHandler() {
- when(table.schemaView()).thenReturn(schemaRegistry);
when(table.name()).thenReturn("TEST_SCHEMA.T");
InternalTable mock = mock(InternalTable.class);
@@ -225,19 +228,21 @@ public class SqlSchemaManagerTest {
when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- schemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
testRevisionRegister.moveForward();
- Table schemaTable = schemaManager.schema("TEST_SCHEMA").getTable("T");
+ Table schemaTable = sqlSchemaManager.schema("TEST_SCHEMA").getTable("T");
assertNotNull(schemaTable);
IgniteTableImpl igniteTable = assertInstanceOf(IgniteTableImpl.class, schemaTable);
assertEquals(tableId, igniteTable.table().tableId());
- schemaManager.onTableDropped("TEST_SCHEMA", table.name(), testRevisionRegister.actualToken() + 1);
+ sqlSchemaManager.onTableDropped("TEST_SCHEMA", table.name(), testRevisionRegister.actualToken() + 1);
testRevisionRegister.moveForward();
- assertNull(schemaManager.schema("TEST_SCHEMA").getTable("T"));
+ assertNull(sqlSchemaManager.schema("TEST_SCHEMA").getTable("T"));
}
/**
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
index 7728d6f2e..b50c86f47 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
@@ -97,7 +97,9 @@ public class TableImpl implements Table {
* Sets a schema view for the table.
*/
public void schemaView(@NotNull SchemaRegistry schemaReg) {
- assert this.schemaReg == null : "Schema registry is already set [tableName=" + name() + "]";
+ assert this.schemaReg == null : "Schema registry is already set [tableName=" + name() + ']';
+
+ Objects.requireNonNull(schemaReg, "Schema registry must not be null [tableName=" + name() + ']');
this.schemaReg = schemaReg;
}
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 424c9b4d1..056de9a97 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.table.distributed;
import static java.util.Collections.unmodifiableMap;
+import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.getByInternalId;
@@ -39,6 +40,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -121,6 +123,7 @@ import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupListener;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.jraft.util.Utils;
+import org.apache.ignite.raft.jraft.util.concurrent.ConcurrentHashSet;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.manager.IgniteTables;
import org.jetbrains.annotations.NotNull;
@@ -132,6 +135,10 @@ import org.jetbrains.annotations.TestOnly;
*/
public class TableManager extends Producer<TableEvent, TableEventParameters> implements IgniteTables, IgniteTablesInternal,
IgniteComponent {
+ // TODO get rid of this in future? IGNITE-17307
+ /** Timeout to complete the tablesByIdVv on revision update. */
+ private static final long TABLES_COMPLETE_TIMEOUT = 120;
+
/** The logger. */
private static final IgniteLogger LOG = Loggers.forClass(TableManager.class);
@@ -167,6 +174,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/** Versioned store for tables by id. */
private final VersionedValue<Map<UUID, TableImpl>> tablesByIdVv;
+ /** Set of futures that should complete before completion of {@link #tablesByIdVv}, after completion this set is cleared. */
+ private final Set<CompletableFuture<?>> beforeTablesVvComplete = new ConcurrentHashSet<>();
+
/** Resolver that resolves a network address to node id. */
private final Function<NetworkAddress, String> netAddrResolver;
@@ -227,10 +237,27 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
return node.id();
};
+
clusterNodeResolver = topologyService::getByAddress;
tablesByIdVv = new VersionedValue<>(null, HashMap::new);
+ registry.accept(token -> {
+ List<CompletableFuture<?>> futures = new ArrayList<>(beforeTablesVvComplete);
+
+ beforeTablesVvComplete.clear();
+
+ return CompletableFuture.allOf(futures.toArray(new CompletableFuture[] {}))
+ .orTimeout(TABLES_COMPLETE_TIMEOUT, TimeUnit.SECONDS)
+ .whenComplete((v, e) -> {
+ if (e != null) {
+ tablesByIdVv.completeExceptionally(token, e);
+ } else {
+ tablesByIdVv.complete(token);
+ }
+ });
+ });
+
rebalanceScheduler = new ScheduledThreadPoolExecutor(REBALANCE_SCHEDULER_POOL_SIZE,
new NamedThreadFactory("rebalance-scheduler", LOG));
}
@@ -266,7 +293,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
schemaManager.listen(SchemaEvent.CREATE, new EventListener<>() {
/** {@inheritDoc} */
@Override
- public boolean notify(@NotNull SchemaEventParameters parameters, @Nullable Throwable exception) {
+ public CompletableFuture<Boolean> notify(@NotNull SchemaEventParameters parameters, @Nullable Throwable exception) {
if (tablesByIdVv.latest().get(parameters.tableId()) != null) {
fireEvent(
TableEvent.ALTER,
@@ -274,21 +301,11 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
);
}
- return false;
+ return completedFuture(false);
}
});
}
- /**
- * Completes all table futures.
- * TODO: Get rid of it after IGNITE-17062.
- *
- * @param causalityToken Causality token.
- */
- public void onSqlSchemaReady(long causalityToken) {
- tablesByIdVv.complete(causalityToken);
- }
-
/**
* Listener of table create configuration change.
*
@@ -414,7 +431,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
busyLock.leaveBusy();
}
- return CompletableFuture.completedFuture(null);
+ return completedFuture(null);
}
/**
@@ -550,7 +567,6 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
tableStorage.start();
-
InternalTableImpl internalTable = new InternalTableImpl(name, tblId, new Int2ObjectOpenHashMap<>(partitions),
partitions, netAddrResolver, clusterNodeResolver, txManager, tableStorage);
@@ -568,9 +584,11 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
return completedFuture(val);
});
- schemaManager.schemaRegistry(causalityToken, tblId)
+ CompletableFuture<?> schemaFut = schemaManager.schemaRegistry(causalityToken, tblId)
.thenAccept(table::schemaView)
- .thenRun(() -> fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, table), null));
+ .thenCompose(v -> fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, table)));
+
+ beforeTablesVvComplete.add(schemaFut);
// TODO should be reworked in IGNITE-16763
return tablesByIdVv.get(causalityToken).thenRun(() -> completeApiCreateFuture(table));
@@ -622,13 +640,14 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
TableImpl table = tablesByIdVv.latest().get(tblId);
assert table != null : IgniteStringFormatter.format("There is no table with the name specified [name={}, id={}]",
- name, tblId);
+ name, tblId);
table.internalTable().storage().destroy();
- fireEvent(TableEvent.DROP, new TableEventParameters(causalityToken, table), null);
+ CompletableFuture<?> fut = schemaManager.dropRegistry(causalityToken, table.tableId())
+ .thenCompose(v -> fireEvent(TableEvent.DROP, new TableEventParameters(causalityToken, table)));
- schemaManager.dropRegistry(causalityToken, table.tableId());
+ beforeTablesVvComplete.add(fut);
} catch (Exception e) {
fireEvent(TableEvent.DROP, new TableEventParameters(causalityToken, tblId, name), e);
}
@@ -983,7 +1002,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
tableFuts[i++] = tableAsyncInternal(tblId, false);
}
- return CompletableFuture.allOf(tableFuts).thenApply(unused -> {
+ return allOf(tableFuts).thenApply(unused -> {
var tables = new ArrayList<Table>(tableIds.size());
try {
@@ -1142,9 +1161,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
EventListener<TableEventParameters> clo = new EventListener<>() {
@Override
- public boolean notify(@NotNull TableEventParameters parameters, @Nullable Throwable e) {
+ public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable e) {
if (!id.equals(parameters.tableId())) {
- return false;
+ return completedFuture(false);
}
if (e == null) {
@@ -1153,7 +1172,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
getTblFut.completeExceptionally(e);
}
- return true;
+ return completedFuture(true);
}
@Override
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 e5219ed1b..f39c12102 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
@@ -564,7 +564,7 @@ public class TableManagerTest extends IgniteAbstractTest {
when(raftGrpSrvcMock.leader()).thenReturn(new Peer(new NetworkAddress("localhost", 47500)));
- return CompletableFuture.completedFuture(raftGrpSrvcMock);
+ return completedFuture(raftGrpSrvcMock);
});
when(ts.getByAddress(any(NetworkAddress.class))).thenReturn(new ClusterNode(
@@ -601,27 +601,22 @@ public class TableManagerTest extends IgniteAbstractTest {
&& ctx.newValue().get(tableDefinition.canonicalName()) == null;
if (!createTbl && !dropTbl) {
- return CompletableFuture.completedFuture(null);
+ return completedFuture(null);
}
if (phaser != null) {
phaser.arriveAndAwaitAdvance();
}
- return CompletableFuture.completedFuture(null);
+ return completedFuture(null);
});
CountDownLatch createTblLatch = new CountDownLatch(1);
- AtomicLong token = new AtomicLong();
-
tableManager.listen(TableEvent.CREATE, (parameters, exception) -> {
-
createTblLatch.countDown();
- token.set(parameters.causalityToken());
-
- return true;
+ return completedFuture(true);
});
CompletableFuture<Table> tbl2Fut = tableManager.createTableAsync(tableDefinition.canonicalName(),
@@ -636,8 +631,6 @@ public class TableManagerTest extends IgniteAbstractTest {
assertFalse(tbl2Fut.isDone());
- tableManager.onSqlSchemaReady(token.get());
-
TableImpl tbl2 = (TableImpl) tbl2Fut.get();
assertNotNull(tbl2);
@@ -776,13 +769,8 @@ public class TableManagerTest extends IgniteAbstractTest {
sm.start();
- //TODO: Get rid of it after IGNITE-17062.
if (!waitingSqlSchema) {
- tableManager.listen(TableEvent.CREATE, (parameters, exception) -> {
- tableManager.onSqlSchemaReady(parameters.causalityToken());
-
- return false;
- });
+ tableManager.listen(TableEvent.CREATE, (parameters, exception) -> completedFuture(false));
}
tableManager.start();