You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by zs...@apache.org on 2024/03/26 16:57:45 UTC
(ignite-3) branch ignite-21776 created (now 6763140a3c)
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a change to branch ignite-21776
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at 6763140a3c IGNITE-21776 Sql. Concurrent create table command need to wait already processed result
This branch includes the following new commits:
new 6763140a3c IGNITE-21776 Sql. Concurrent create table command need to wait already processed result
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 01/01: IGNITE-21776 Sql. Concurrent create table command need to wait already processed result
Posted by zs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a commit to branch ignite-21776
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 6763140a3c5b884fcefb3164e4a3073e296c707e
Author: zstan <zs...@apache.com>
AuthorDate: Tue Mar 26 18:22:05 2024 +0300
IGNITE-21776 Sql. Concurrent create table command need to wait already processed result
---
.../internal/catalog/CatalogManagerImpl.java | 42 ++++++++++++++++-
.../CatalogVersionAwareValidationException.java | 41 +++++++++++++++++
.../internal/catalog/CatalogManagerSelfTest.java | 52 ++++++++++++++++++++++
3 files changed, 133 insertions(+), 2 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 9a709821ca..35622456d4 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
@@ -374,7 +374,42 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
private CompletableFuture<Integer> saveUpdateAndWaitForActivation(UpdateProducer updateProducer) {
return saveUpdate(updateProducer, 0)
- .thenCompose(newVersion -> {
+ .handle((newVersion, err) -> {
+ if (err != null) {
+ Throwable error;
+ Catalog catalog;
+
+ if (err instanceof CatalogVersionAwareValidationException) {
+ CatalogVersionAwareValidationException err0 = (CatalogVersionAwareValidationException) err;
+ catalog = catalogByVer.get(err0.version());
+ error = err0.initial();
+ } else {
+ catalog = catalogByVer.lastEntry().getValue();
+ error = err;
+ }
+
+ if (catalog.version() == 0) {
+ return failedFuture(error).thenApply(unused -> newVersion);
+ }
+
+ HybridTimestamp tsSafeForRoReadingInPastOptimization =
+ clusterWideEnsuredActivationTsSafeForRoReads(
+ catalog,
+ partitionIdleSafeTimePropagationPeriodMsSupplier,
+ clockService.maxClockSkewMillis()
+ );
+
+ return clockService.waitFor(tsSafeForRoReadingInPastOptimization)
+ .handle((res, ex) -> {
+ if (ex != null) {
+ error.addSuppressed(ex);
+ }
+ return failedFuture(error);
+ })
+ .thenCompose(unused -> failedFuture(error))
+ .thenApply(unused -> newVersion);
+ }
+
Catalog catalog = catalogByVer.get(newVersion);
HybridTimestamp tsSafeForRoReadingInPastOptimization = clusterWideEnsuredActivationTsSafeForRoReads(
@@ -384,7 +419,8 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
);
return clockService.waitFor(tsSafeForRoReadingInPastOptimization).thenApply(unused -> newVersion);
- });
+ })
+ .thenCompose(Function.identity());
}
/**
@@ -410,6 +446,8 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
List<UpdateEntry> updates;
try {
updates = updateProducer.get(catalog);
+ } catch (CatalogValidationException ex) {
+ return failedFuture(new CatalogVersionAwareValidationException(ex, catalog.version()));
} catch (Exception ex) {
return failedFuture(ex);
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogVersionAwareValidationException.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogVersionAwareValidationException.java
new file mode 100644
index 0000000000..4243c7e489
--- /dev/null
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogVersionAwareValidationException.java
@@ -0,0 +1,41 @@
+/*
+ * 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.catalog;
+
+/** Catalog version aware exception wrapper. */
+class CatalogVersionAwareValidationException extends CatalogValidationException {
+ private static final long serialVersionUID = -1482326886088511707L;
+
+ private final int version;
+ private final Throwable initial;
+
+ CatalogVersionAwareValidationException(Throwable exception, int version) {
+ super(exception.getMessage());
+
+ this.version = version;
+ initial = exception;
+ }
+
+ int version() {
+ return version;
+ }
+
+ Throwable initial() {
+ return initial;
+ }
+}
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 b7d5b6d970..d469b4eb2b 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
@@ -82,6 +82,7 @@ import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -1162,6 +1163,57 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
}
}
+ @Test
+ public void createTableIfNotExistWaitsActivationEvenIfTableExists() throws Exception {
+ long delayDuration = TimeUnit.DAYS.toMillis(365);
+
+ int partitionIdleSafeTimePropagationPeriod = 0;
+
+ CatalogManagerImpl manager = new CatalogManagerImpl(updateLog, clockService, delayDuration,
+ partitionIdleSafeTimePropagationPeriod);
+
+ manager.start();
+
+ try {
+ CatalogCommand createTableCommand = spy(simpleTable(TABLE_NAME));
+
+ CompletableFuture<Integer> createTableFuture1 = manager.execute(createTableCommand);
+
+ assertFalse(createTableFuture1.isDone());
+
+ ArgumentCaptor<VersionedUpdate> appendCapture = ArgumentCaptor.forClass(VersionedUpdate.class);
+
+ verify(updateLog).append(appendCapture.capture());
+
+ int catalogVerAfterTableCreate = appendCapture.getValue().version();
+
+ CompletableFuture<Integer> createTableFuture2 = manager.execute(createTableCommand);
+
+ verify(createTableCommand, times(2)).get(any());
+
+ assertFalse(createTableFuture2.isDone());
+
+ verify(clockWaiter, timeout(10_000).times(2)).waitFor(any());
+
+ Catalog catalog0 = manager.catalog(manager.latestCatalogVersion());
+
+ assertNotNull(catalog0);
+
+ HybridTimestamp activationSkew = CatalogUtils.clusterWideEnsuredActivationTsSafeForRoReads(
+ catalog0,
+ () -> partitionIdleSafeTimePropagationPeriod, clockService.maxClockSkewMillis());
+
+ clock.update(activationSkew);
+
+ assertTrue(waitForCondition(createTableFuture1::isDone, 2_000));
+ assertTrue(waitForCondition(createTableFuture2::isDone, 2_000));
+
+ assertSame(manager.schema(catalogVerAfterTableCreate), manager.activeSchema(clock.nowLong()));
+ } finally {
+ manager.stop();
+ }
+ }
+
@Test
public void catalogServiceManagesUpdateLogLifecycle() throws Exception {
UpdateLog updateLogMock = mock(UpdateLog.class);