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 2022/08/26 14:03:15 UTC
[ignite-3] branch ignite-17431 created (now b543cd9c7a)
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a change to branch ignite-17431
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at b543cd9c7a Fix DdlCommandHandler, TableManager, IndexManager async methods usage.
This branch includes the following new commits:
new a47ffbe3f9 Make calcite schema objects immutable.
new b543cd9c7a Fix DdlCommandHandler, TableManager, IndexManager async methods usage.
The 2 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/02: Make calcite schema objects immutable.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-17431
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a47ffbe3f9da945e60810283991b6890f6ae2e42
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Aug 25 14:04:57 2022 +0300
Make calcite schema objects immutable.
---
.../internal/sql/engine/schema/IgniteSchema.java | 24 +---
.../sql/engine/schema/IgniteTableImpl.java | 18 ++-
.../sql/engine/schema/SqlSchemaManagerImpl.java | 148 ++++++++++++++-------
.../engine/exec/schema/SqlSchemaManagerTest.java | 71 +++++++++-
4 files changed, 190 insertions(+), 71 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
index f983e3c5f1..243fa682d3 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
@@ -54,6 +54,10 @@ public class IgniteSchema extends AbstractSchema {
this(schemaName, null, null);
}
+ public static IgniteSchema copy(IgniteSchema old) {
+ return new IgniteSchema(old.schemaName, old.tblMap, old.idxMap);
+ }
+
/**
* Get schema name.
*
@@ -99,10 +103,6 @@ public class IgniteSchema extends AbstractSchema {
* @param index Index.
*/
public void addIndex(UUID indexId, IgniteIndex index) {
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17474
- // Decouple tables and indices.
- index.table().addIndex(index);
-
idxMap.put(indexId, index);
}
@@ -110,20 +110,10 @@ public class IgniteSchema extends AbstractSchema {
* Remove index.
*
* @param indexId Index id.
- * @return {@code True} if index was removed or {@code false} otherwise.
+ * @return Removed index.
*/
- public boolean removeIndex(UUID indexId) {
- IgniteIndex rmv = idxMap.remove(indexId);
-
- if (rmv == null) {
- return false;
- }
-
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17474
- // Decouple tables and indices.
- rmv.table().removeIndex(rmv.name());
-
- return true;
+ public IgniteIndex removeIndex(UUID indexId) {
+ return idxMap.remove(indexId);
}
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
index cd9258cad5..5529b215a7 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
@@ -22,6 +22,7 @@ import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -81,7 +82,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
private final Statistic statistic;
- private final Map<String, IgniteIndex> indexes = new ConcurrentHashMap<>();
+ private Map<String, IgniteIndex> indexes = new HashMap<>();
private final List<ColumnDescriptor> columnsOrderedByPhysSchema;
@@ -115,6 +116,21 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
statistic = new StatisticsImpl();
}
+ private IgniteTableImpl(IgniteTableImpl t) {
+ this.desc = t.desc;
+ this.ver = t.ver;
+ this.table = t.table;
+ this.schemaRegistry = t.schemaRegistry;
+ this.schemaDescriptor = t.schemaDescriptor;
+ this.statistic = t.statistic;
+ this.columnsOrderedByPhysSchema = t.columnsOrderedByPhysSchema;
+ this.indexes.putAll(t.indexes);
+ }
+
+ public static IgniteTableImpl copyOf(IgniteTableImpl v) {
+ return new IgniteTableImpl(v);
+ }
+
/** {@inheritDoc} */
@Override
public UUID id() {
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 861500261a..72aefa5be9 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
@@ -26,6 +26,7 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
@@ -66,7 +67,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
private final VersionedValue<Map<String, IgniteSchema>> schemasVv;
- private final VersionedValue<Map<UUID, IgniteTable>> tablesVv;
+ private final VersionedValue<Map<UUID, InternalIgniteTable>> tablesVv;
private final VersionedValue<Map<UUID, IgniteIndex>> indicesVv;
@@ -213,7 +214,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
CompletableFuture<IgniteTableImpl> igniteTableFuture = convert(causalityToken, table);
@@ -223,7 +225,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return failedFuture(ex);
}
- Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
return igniteTableFuture
.thenApply(igniteTable -> inBusyLock(busyLock, () -> {
@@ -281,7 +283,8 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
String calciteTableName = objectSimpleName(schemaName, tableName);
@@ -295,7 +298,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
return failedFuture(ex);
}
- Map<UUID, IgniteTable> resTbls = new HashMap<>(tables);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
resTbls.remove(table.id());
@@ -363,15 +366,9 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
);
}
- private CompletableFuture<IgniteIndex> convert(long causalityToken, Index<?> index) {
- return schemaManager.schemaRegistry(causalityToken, index.tableId())
- .thenApply(schemaRegistry -> inBusyLock(busyLock, () -> convert(index, schemaRegistry)));
- }
-
- private IgniteIndex convert(Index<?> index, SchemaRegistry schemaRegistry) {
+ private IgniteIndex convert(Index<?> index, SchemaRegistry schemaRegistry, IgniteTable table) {
IndexDescriptor desc = index.descriptor();
SchemaDescriptor schema = schemaRegistry.schema();
- IgniteTable table = tablesVv.latest().get(index.tableId());
assert table != null;
@@ -430,36 +427,54 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
-
- CompletableFuture<IgniteIndex> igniteIndexFuture = convert(causalityToken, index);
-
- return indicesVv
- .update(
- causalityToken,
- (indices, ex) -> inBusyLock(busyLock, () -> {
- if (ex != null) {
- return failedFuture(ex);
- }
-
- Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
-
- return igniteIndexFuture
- .thenApply(igniteIndex -> {
- resIdxs.put(index.id(), igniteIndex);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
- return resIdxs;
- });
- })
- )
- .thenCombine(
- igniteIndexFuture,
- (v, igniteIndex) -> inBusyLock(busyLock, () -> {
- schema.addIndex(index.id(), igniteIndex);
+ return tablesVv.update(causalityToken, (tables, tblEx) -> inBusyLock(busyLock, () -> {
+ if (tblEx != null) {
+ return failedFuture(tblEx);
+ }
- return null;
- })
- )
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
+
+ InternalIgniteTable table = resTbls.compute(index.tableId(),
+ (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
+
+ CompletableFuture<IgniteIndex> igniteIndexFuture = schemaManager.schemaRegistry(causalityToken, index.tableId())
+ .thenApply(reg -> convert(index, reg, table));
+
+ return indicesVv
+ .update(
+ causalityToken,
+ (indices, idxEx) -> inBusyLock(busyLock, () -> {
+ if (idxEx != null) {
+ return failedFuture(idxEx);
+ }
+
+ Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+
+ return igniteIndexFuture
+ .thenApply(igniteIndex -> {
+ resIdxs.put(index.id(), igniteIndex);
+
+ return resIdxs;
+ });
+ })
+ )
+ .thenCombine(
+ igniteIndexFuture,
+ (v, igniteIndex) -> inBusyLock(busyLock, () -> {
+ String tblName = tableNameById(schema, index.tableId());
+
+ table.addIndex(igniteIndex);
+ schema.addTable(tblName, (InternalIgniteTable) table);
+ schema.addIndex(index.id(), igniteIndex);
+
+ return v;
+ })
+ )
+ .thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(resTbls)));
+ }))
.thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
}));
@@ -469,6 +484,18 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
}
}
+ @NotNull
+ private static String tableNameById(IgniteSchema schema, UUID tableId) {
+ String tblName = schema.getTableMap()
+ .entrySet()
+ .stream()
+ .filter(entry -> tableId
+ .equals(((InternalIgniteTable) entry.getValue()).id()))
+ .map(Entry::getKey)
+ .findFirst().get();
+ return tblName;
+ }
+
/**
* Index dropped callback method deregisters index from Calcite schema.
*
@@ -489,21 +516,42 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager {
Map<String, IgniteSchema> res = new HashMap<>(schemas);
- IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new);
+ IgniteSchema schema = res.compute(schemaName,
+ (k, v) -> v == null ? new IgniteSchema(schemaName) : IgniteSchema.copy(v));
- if (schema.removeIndex(indexId)) {
- return indicesVv.update(causalityToken, (indices, ex) -> inBusyLock(busyLock, () -> {
- if (ex != null) {
- return failedFuture(ex);
+ IgniteIndex rmvIndex = schema.removeIndex(indexId);
+ if (rmvIndex != null) {
+ return tablesVv.update(causalityToken, (tables, tlbEx) -> inBusyLock(busyLock, () -> {
+ if (tlbEx != null) {
+ return failedFuture(tlbEx);
}
- Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+ Map<UUID, InternalIgniteTable> resTbls = new HashMap<>(tables);
+
+ InternalIgniteTable table = resTbls.compute(rmvIndex.table().id(),
+ (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
- resIdxs.remove(indexId);
+ table.removeIndex(rmvIndex.name());
- return completedFuture(resIdxs);
- }
- )).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
+ return indicesVv.update(causalityToken, (indices, idxEx) -> inBusyLock(busyLock, () -> {
+ if (idxEx != null) {
+ return failedFuture(idxEx);
+ }
+
+ Map<UUID, IgniteIndex> resIdxs = new HashMap<>(indices);
+
+ IgniteIndex rmvIdx = resIdxs.remove(indexId);
+
+ assert table.id().equals(rmvIdx.table().id());
+
+ String tblName = tableNameById(schema, rmvIdx.table().id());
+ schema.addTable(tblName, table);
+
+ return completedFuture(resIdxs);
+ }
+ )).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(resTbls)));
+ })
+ ).thenCompose(v -> inBusyLock(busyLock, () -> completedFuture(res)));
}
return completedFuture(res);
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 b9d2f64351..84bf63c766 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
@@ -24,6 +24,7 @@ import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
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.assertThrows;
@@ -42,6 +43,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
+import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
import org.apache.ignite.internal.index.Index;
import org.apache.ignite.internal.index.IndexDescriptor;
@@ -54,6 +56,7 @@ import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.IgniteTableImpl;
+import org.apache.ignite.internal.sql.engine.schema.InternalIgniteTable;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableImpl;
@@ -83,7 +86,7 @@ public class SqlSchemaManagerTest {
tableVer,
new Column[]{new Column(0, "ID", NativeTypes.INT64, false)},
new Column[]{new Column(1, "VAL", NativeTypes.INT64, false)}
- );
+ );
@Mock
private TableManager tableManager;
@@ -312,6 +315,68 @@ public class SqlSchemaManagerTest {
verifyNoMoreInteractions(tableManager);
}
+
+ @Test
+ public void testIndexEventsProcessed() {
+ InternalTable mock = mock(InternalTable.class);
+ when(mock.tableId()).thenReturn(tableId);
+
+ when(table.name()).thenReturn("TEST_SCHEMA.T");
+ when(table.internalTable()).thenReturn(mock);
+ when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
+ when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
+ when(schemaManager.schemaRegistry(anyLong(), any())).thenReturn(completedFuture(schemaRegistry));
+
+ sqlSchemaManager.onTableCreated("TEST_SCHEMA", table, testRevisionRegister.actualToken() + 1);
+ testRevisionRegister.moveForward();
+
+ IndexDescriptor descMock = mock(IndexDescriptor.class);
+ when(descMock.columns()).thenReturn(List.of());
+
+ when(index.name()).thenReturn("I");
+ when(index.id()).thenReturn(indexId);
+ when(index.tableId()).thenReturn(tableId);
+ when(index.descriptor()).thenReturn(descMock);
+
+ {
+ SchemaPlus schema1 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ sqlSchemaManager.onIndexCreated("TEST_SCHEMA", index, testRevisionRegister.actualToken() + 1);
+ testRevisionRegister.moveForward();
+
+ SchemaPlus schema2 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ // Validate schema snapshot.
+ assertNotSame(schema1, schema2);
+ assertNotSame(schema1.getTable("T"), schema2.getTable("T"));
+
+ assertNull(schema1.unwrap(IgniteSchema.class).index(indexId));
+ assertNotNull(schema2.unwrap(IgniteSchema.class).index(indexId));
+
+ assertNull(((InternalIgniteTable)schema1.getTable("T")).getIndex("I"));
+ assertNotNull(((InternalIgniteTable)schema2.getTable("T")).getIndex("I"));
+ }
+ {
+ sqlSchemaManager.onIndexDropped("TEST_SCHEMA", indexId, testRevisionRegister.actualToken() + 1);
+ SchemaPlus schema1 = sqlSchemaManager.schema("TEST_SCHEMA");
+ testRevisionRegister.moveForward();
+
+ SchemaPlus schema2 = sqlSchemaManager.schema("TEST_SCHEMA");
+
+ // Validate schema snapshot.
+ assertNotSame(schema1, schema2);
+ assertNotSame(schema1.getTable("T"), schema2.getTable("T"));
+
+ assertNotNull(schema1.unwrap(IgniteSchema.class).index(indexId));
+ assertNull(schema2.unwrap(IgniteSchema.class).index(indexId));
+
+ assertNotNull(((InternalIgniteTable)schema1.getTable("T")).getIndex("I"));
+ assertNull(((InternalIgniteTable)schema2.getTable("T")).getIndex("I"));
+ }
+
+ verifyNoMoreInteractions(tableManager);
+ }
+
/**
* Test revision register.
*/
@@ -346,8 +411,8 @@ public class SqlSchemaManagerTest {
Function<Long, CompletableFuture<?>> old = moveRevision;
moveRevision = rev -> allOf(
- old.apply(rev),
- function.apply(rev)
+ old.apply(rev),
+ function.apply(rev)
);
}
}
[ignite-3] 02/02: Fix DdlCommandHandler, TableManager, IndexManager async methods usage.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-17431
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit b543cd9c7a7138c063005451460a3b02d1dacf7c
Author: amashenkov <an...@gmail.com>
AuthorDate: Fri Aug 26 17:02:17 2022 +0300
Fix DdlCommandHandler, TableManager, IndexManager async methods usage.
---
.../apache/ignite/internal/index/IndexManager.java | 38 ----------
.../sql/engine/exec/ExecutionServiceImpl.java | 21 ++++--
.../sql/engine/exec/ddl/DdlCommandHandler.java | 88 +++++++++++-----------
3 files changed, 58 insertions(+), 89 deletions(-)
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
index 322ca04573..8bd34d4f82 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
@@ -122,27 +122,6 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
LOG.info("Index manager stopped");
}
- /**
- * Creates index from provided configuration changer.
- *
- * @param schemaName A name of the schema to create index in.
- * @param indexName A name of the index to create.
- * @param tableName A name of the table to create index for.
- * @param indexChange A consumer that suppose to change the configuration in order to provide description of an index.
- * @param failIfExists Flag indicates whether exception be thrown if index exists or not.
- * @return {@code True} if index was created successfully, {@code false} otherwise.
- * @throws IndexAlreadyExistsException If index already exists and
- */
- public boolean createIndex(
- String schemaName,
- String indexName,
- String tableName,
- boolean failIfExists,
- Consumer<TableIndexChange> indexChange
- ) {
- return join(createIndexAsync(schemaName, indexName, tableName, failIfExists, indexChange)) != null;
- }
-
/**
* Creates index from provided configuration changer.
*
@@ -251,23 +230,6 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
}
}
- /**
- * Drops the index with a given name.
- *
- * @param schemaName A name of the schema the index belong to.
- * @param indexName A name of the index to drop.
- * @param failIfNotExist Flag, which force failure, when {@code trues} if index doen't not exists.
- * @return {@code True} if index was removed, {@code false} otherwise.
- * @throws IndexNotFoundException If index doesn't exist and {@code failIfNotExist} param was {@code true}.
- */
- public boolean dropIndex(
- String schemaName,
- String indexName,
- boolean failIfNotExist
- ) {
- return join(dropIndexAsync(schemaName, indexName, failIfNotExist));
- }
-
/**
* Drops the index with a given name asynchronously.
*
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index c3109c1862..c5c6c9939e 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -22,8 +22,8 @@ import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFI
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -32,6 +32,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
@@ -249,14 +250,18 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}
private AsyncCursor<List<Object>> executeDdl(DdlPlan plan) {
- try {
- boolean ret = ddlCmdHnd.handle(plan.command());
+ CompletableFuture<Iterator<List<Object>>> res = ddlCmdHnd.handle(plan.command())
+ .handle((v, err) -> {
+ if (err == null) {
+ return List.of(List.of((Object) v)).iterator();
+ }
- return new AsyncWrapper<>(Collections.singletonList(Collections.<Object>singletonList(ret)).iterator());
- } catch (IgniteInternalCheckedException e) {
- throw new IgniteInternalException("Failed to execute DDL statement [stmt=" /*+ qry.sql()*/
- + ", err=" + e.getMessage() + ']', e);
- }
+ throw new IgniteInternalException(
+ "Failed to execute DDL statement [stmt=" /*+ qry.sql()*/
+ + ", err=" + err.getMessage() + ']', err);
+ });
+
+ return new AsyncWrapper<>(res, ForkJoinPool.commonPool());
}
private AsyncCursor<List<Object>> executeExplain(ExplainPlan plan) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandler.java
index 1f8cbd52bd..b671771142 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandler.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ddl/DdlCommandHandler.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.sql.engine.exec.ddl;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter.convert;
import static org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter.convertDefaultToConfiguration;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
@@ -25,7 +27,9 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@@ -89,7 +93,7 @@ public class DdlCommandHandler {
}
/** Handles ddl commands. */
- public boolean handle(DdlCommand cmd) throws IgniteInternalCheckedException {
+ public CompletableFuture<Boolean> handle(DdlCommand cmd) {
validateCommand(cmd);
if (cmd instanceof CreateTableCommand) {
@@ -105,9 +109,9 @@ public class DdlCommandHandler {
} else if (cmd instanceof DropIndexCommand) {
return handleDropIndex((DropIndexCommand) cmd);
} else {
- throw new IgniteInternalCheckedException("Unsupported DDL operation ["
+ return failedFuture(new IgniteInternalCheckedException("Unsupported DDL operation ["
+ "cmdName=" + (cmd == null ? null : cmd.getClass().getSimpleName()) + "; "
- + "cmd=\"" + cmd + "\"]");
+ + "cmd=\"" + cmd + "\"]"));
}
}
@@ -123,7 +127,7 @@ public class DdlCommandHandler {
}
/** Handles create table command. */
- private boolean handleCreateTable(CreateTableCommand cmd) {
+ private CompletableFuture<Boolean> handleCreateTable(CreateTableCommand cmd) {
Consumer<TableChange> tblChanger = tableChange -> {
tableChange.changeColumns(columnsChange -> {
for (var col : cmd.columns()) {
@@ -159,41 +163,39 @@ public class DdlCommandHandler {
);
try {
- tableManager.createTable(fullName, tblChanger);
-
- return true;
+ return tableManager.createTableAsync(fullName, tblChanger)
+ .thenApply(t -> Boolean.TRUE);
} catch (TableAlreadyExistsException ex) {
if (!cmd.ifTableExists()) {
- throw ex;
+ return CompletableFuture.failedFuture(ex);
} else {
- return false;
+ return completedFuture(Boolean.FALSE);
}
}
}
/** Handles drop table command. */
- private boolean handleDropTable(DropTableCommand cmd) {
+ private CompletableFuture<Boolean> handleDropTable(DropTableCommand cmd) {
String fullName = TableDefinitionImpl.canonicalName(
IgniteObjectName.quote(cmd.schemaName()),
IgniteObjectName.quote(cmd.tableName())
);
try {
- tableManager.dropTable(fullName);
-
- return true;
+ return tableManager.dropTableAsync(fullName)
+ .thenApply(v -> Boolean.TRUE);
} catch (TableNotFoundException ex) {
if (!cmd.ifTableExists()) {
- throw ex;
+ return CompletableFuture.failedFuture(ex);
} else {
- return false;
+ return completedFuture(Boolean.FALSE);
}
}
}
/** Handles add column command. */
- private boolean handleAlterAddColumn(AlterTableAddCommand cmd) {
+ private CompletableFuture<Boolean> handleAlterAddColumn(AlterTableAddCommand cmd) {
if (nullOrEmpty(cmd.columns())) {
- return false;
+ return completedFuture(Boolean.FALSE);
}
String fullName = TableDefinitionImpl.canonicalName(
@@ -202,20 +204,20 @@ public class DdlCommandHandler {
);
try {
- return addColumnInternal(fullName, cmd.columns(), cmd.ifColumnNotExists());
+ return addColumnInternal(fullName, cmd.columns(), !cmd.ifColumnNotExists());
} catch (TableNotFoundException ex) {
if (!cmd.ifTableExists()) {
- throw ex;
+ return CompletableFuture.failedFuture(ex);
} else {
- return false;
+ return completedFuture(Boolean.FALSE);
}
}
}
/** Handles drop column command. */
- private boolean handleAlterDropColumn(AlterTableDropCommand cmd) {
+ private CompletableFuture<Boolean> handleAlterDropColumn(AlterTableDropCommand cmd) {
if (nullOrEmpty(cmd.columns())) {
- return false;
+ return completedFuture(Boolean.FALSE);
}
String fullName = TableDefinitionImpl.canonicalName(
@@ -227,15 +229,15 @@ public class DdlCommandHandler {
return dropColumnInternal(fullName, cmd.columns(), cmd.ifColumnExists());
} catch (TableNotFoundException ex) {
if (!cmd.ifTableExists()) {
- throw ex;
+ return CompletableFuture.failedFuture(ex);
} else {
- return false;
+ return completedFuture(Boolean.FALSE);
}
}
}
/** Handles create index command. */
- private boolean handleCreateIndex(CreateIndexCommand cmd) {
+ private CompletableFuture<Boolean> handleCreateIndex(CreateIndexCommand cmd) {
// Only sorted idx for now.
//TODO: https://issues.apache.org/jira/browse/IGNITE-17563 Pass null ordering for columns.
SortedIndexDefinitionBuilder idx = SchemaBuilders.sortedIndex(cmd.indexName());
@@ -250,17 +252,18 @@ public class DdlCommandHandler {
idx0.done();
}
- return indexManager.createIndex(
+ return indexManager.createIndexAsync(
cmd.schemaName(),
cmd.indexName(),
cmd.tableName(),
!cmd.ifIndexNotExists(),
- tableIndexChange -> convert(idx.build(), tableIndexChange));
+ tableIndexChange -> convert(idx.build(), tableIndexChange))
+ .thenApply(Objects::nonNull);
}
/** Handles drop index command. */
- private boolean handleDropIndex(DropIndexCommand cmd) {
- return indexManager.dropIndex(cmd.schemaName(), cmd.indexName(), !cmd.ifNotExists());
+ private CompletableFuture<Boolean> handleDropIndex(DropIndexCommand cmd) {
+ return indexManager.dropIndexAsync(cmd.schemaName(), cmd.indexName(), !cmd.ifNotExists());
}
/**
@@ -268,20 +271,21 @@ public class DdlCommandHandler {
*
* @param fullName Table with schema name.
* @param colsDef Columns defenitions.
- * @param colNotExist Flag indicates exceptionally behavior in case of already existing column.
+ * @param failIfExists Flag indicates exceptionally behavior in case of already existing column.
*
* @return {@code true} if the full columns set is applied successfully. Otherwise, returns {@code false}.
*/
- private boolean addColumnInternal(String fullName, List<ColumnDefinition> colsDef, boolean colNotExist) {
+ private CompletableFuture<Boolean> addColumnInternal(String fullName, List<ColumnDefinition> colsDef, boolean failIfExists) {
AtomicBoolean ret = new AtomicBoolean(true);
- tableManager.alterTable(
+
+ return tableManager.alterTableAsync(
fullName,
chng -> chng.changeColumns(cols -> {
Map<String, String> colNamesToOrders = columnOrdersToNames(chng.columns());
List<ColumnDefinition> colsDef0;
- if (!colNotExist) {
+ if (failIfExists) {
colsDef.stream()
.filter(k -> colNamesToOrders.containsKey(k.name()))
.findAny()
@@ -305,9 +309,8 @@ public class DdlCommandHandler {
for (ColumnDefinition col : colsDef0) {
cols.create(col.name(), colChg -> convertColumnDefinition(col, colChg));
}
- }));
-
- return ret.get();
+ }))
+ .thenApply(v -> ret.get());
}
private void convertColumnDefinition(ColumnDefinition definition, ColumnChange columnChange) {
@@ -349,13 +352,13 @@ public class DdlCommandHandler {
*
* @param fullName Table with schema name.
* @param colNames Columns definitions.
- * @param colExist Flag indicates exceptionally behavior in case of already existing column.
+ * @param failIfNotExists Flag indicates exceptionally behavior in case of already existing column.
* @return {@code true} if the full columns set is applied successfully. Otherwise, returns {@code false}.
*/
- private boolean dropColumnInternal(String fullName, Set<String> colNames, boolean colExist) {
+ private CompletableFuture<Boolean> dropColumnInternal(String fullName, Set<String> colNames, boolean failIfNotExists) {
AtomicBoolean ret = new AtomicBoolean(true);
- tableManager.alterTable(
+ return tableManager.alterTableAsync(
fullName,
chng -> chng.changeColumns(cols -> {
PrimaryKeyView priKey = chng.primaryKey();
@@ -370,7 +373,7 @@ public class DdlCommandHandler {
if (!colNamesToOrders.containsKey(colName)) {
ret.set(false);
- if (!colExist) {
+ if (!failIfNotExists) {
throw new ColumnNotFoundException(colName, fullName);
}
} else {
@@ -384,9 +387,8 @@ public class DdlCommandHandler {
}
colNames0.forEach(k -> cols.delete(colNamesToOrders.get(k)));
- }));
-
- return ret.get();
+ }))
+ .thenApply(v -> ret.get());
}
/** Map column name to order. */