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 2023/07/25 10:00:22 UTC
[ignite-3] 02/11: Switch other managers to Catalog events.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19499
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 81c8ef0b97dbf9c17df591dcb7c6f3c516deef93
Author: amashenkov <an...@gmail.com>
AuthorDate: Thu Jul 13 13:23:23 2023 +0300
Switch other managers to Catalog events.
---
.../ignite/internal/catalog/CatalogService.java | 3 +
.../apache/ignite/internal/index/IndexManager.java | 314 +++------------------
.../ignite/internal/index/IndexManagerTest.java | 2 +-
.../storage/ItRebalanceDistributedTest.java | 2 +-
.../runner/app/ItIgniteNodeRestartTest.java | 4 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 6 +-
.../internal/schema/CatalogDescriptorUtils.java | 50 ++++
.../ignite/internal/schema/SchemaManager.java | 158 ++++++-----
.../internal/sql/engine/SqlQueryProcessor.java | 34 +--
.../sql/engine/exec/ExecutionServiceImpl.java | 7 +-
.../sql/engine/schema/CatalogSqlSchemaManager.java | 11 +-
.../internal/sql/engine/util/BaseQueryContext.java | 4 +-
.../sql/engine/exec/MockedStructuresTest.java | 4 +-
.../internal/table/distributed/TableManager.java | 70 ++---
.../table/distributed/TableManagerTest.java | 11 +-
15 files changed, 255 insertions(+), 425 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java
index 82c61d276e..880e2e2767 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogService.java
@@ -42,12 +42,14 @@ public interface CatalogService {
CatalogTableDescriptor table(String tableName, long timestamp);
+ @Deprecated
CatalogTableDescriptor table(int tableId, long timestamp);
CatalogTableDescriptor table(int tableId, int catalogVersion);
CatalogIndexDescriptor index(String indexName, long timestamp);
+ @Deprecated
CatalogIndexDescriptor index(int indexId, long timestamp);
CatalogSchemaDescriptor schema(int version);
@@ -56,6 +58,7 @@ public interface CatalogService {
CatalogZoneDescriptor zone(String zoneName, long timestamp);
+ @Deprecated
CatalogZoneDescriptor zone(int zoneId, long timestamp);
CatalogZoneDescriptor zone(int zoneId, int version);
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 ff34697c24..555e05de18 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
@@ -18,23 +18,15 @@
package org.apache.ignite.internal.index;
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.schema.CatalogDescriptorUtils.toIndexDescriptor;
-import static org.apache.ignite.internal.schema.CatalogDescriptorUtils.toTableDescriptor;
-import static org.apache.ignite.internal.schema.configuration.SchemaConfigurationUtils.findTableView;
import static org.apache.ignite.internal.util.ArrayUtils.STRING_EMPTY_ARRAY;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
import java.util.function.Function;
-import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
-import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.commands.AbstractIndexCommandParams;
import org.apache.ignite.internal.catalog.commands.CreateHashIndexParams;
@@ -46,6 +38,9 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexColumnDescript
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSortedIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
import org.apache.ignite.internal.index.event.IndexEvent;
import org.apache.ignite.internal.index.event.IndexEventParameters;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -59,31 +54,18 @@ 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.schema.configuration.TableView;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
-import org.apache.ignite.internal.schema.configuration.TablesView;
-import org.apache.ignite.internal.schema.configuration.index.HashIndexChange;
-import org.apache.ignite.internal.schema.configuration.index.SortedIndexChange;
-import org.apache.ignite.internal.schema.configuration.index.TableIndexChange;
-import org.apache.ignite.internal.schema.configuration.index.TableIndexConfiguration;
-import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
import org.apache.ignite.internal.storage.index.StorageHashIndexDescriptor;
import org.apache.ignite.internal.storage.index.StorageIndexDescriptor;
import org.apache.ignite.internal.storage.index.StorageSortedIndexDescriptor;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.distributed.PartitionSet;
import org.apache.ignite.internal.table.distributed.TableManager;
-import org.apache.ignite.internal.table.event.TableEvent;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.StringUtils;
import org.apache.ignite.lang.ErrorGroups;
-import org.apache.ignite.lang.ErrorGroups.Common;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.IgniteInternalException;
-import org.apache.ignite.lang.IndexAlreadyExistsException;
-import org.apache.ignite.lang.IndexNotFoundException;
import org.apache.ignite.lang.NodeStoppingException;
-import org.apache.ignite.lang.TableNotFoundException;
/**
* An Ignite component that is responsible for handling index-related commands like CREATE or DROP
@@ -93,9 +75,6 @@ import org.apache.ignite.lang.TableNotFoundException;
public class IndexManager extends Producer<IndexEvent, IndexEventParameters> implements IgniteComponent {
private static final IgniteLogger LOG = Loggers.forClass(IndexManager.class);
- /** Common tables and indexes configuration. */
- private final TablesConfiguration tablesCfg;
-
/** Schema manager. */
private final SchemaManager schemaManager;
@@ -114,18 +93,15 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
/**
* Constructor.
*
- * @param tablesCfg Tables and indexes configuration.
* @param catalogManager Catalog manager.
* @param schemaManager Schema manager.
* @param tableManager Table manager.
*/
public IndexManager(
- TablesConfiguration tablesCfg,
CatalogManager catalogManager,
SchemaManager schemaManager,
TableManager tableManager
) {
- this.tablesCfg = Objects.requireNonNull(tablesCfg, "tablesCfg");
this.schemaManager = Objects.requireNonNull(schemaManager, "schemaManager");
this.tableManager = tableManager;
this.catalogManager = catalogManager;
@@ -136,40 +112,10 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
public void start() {
LOG.debug("Index manager is about to start");
- tablesCfg.indexes().listenElements(new ConfigurationListener());
-
- tableManager.listen(TableEvent.CREATE, (param, ex) -> {
- if (ex != null) {
- return completedFuture(false);
- }
-
- // We can't return this future as the listener's result, because a deadlock can happen in the configuration component:
- // this listener is called inside a configuration notification thread and all notifications are required to finish before
- // new configuration modifications can occur (i.e. we are creating an index below). Therefore we create the index fully
- // asynchronously and rely on the underlying components to handle PK index synchronisation.
- tableManager.tableAsync(param.causalityToken(), param.tableId())
- .thenCompose(table -> {
- String[] pkColumns = Arrays.stream(table.schemaView().schema().keyColumns().columns())
- .map(Column::name)
- .toArray(String[]::new);
-
- String pkName = table.name() + "_PK";
- int pkId = table.tableId() + 1;
-
- // Update config bypassing the Catalog, because PK has just been created in Catalog with the table.
- return createIndexInternal(pkId, "PUBLIC", pkName, table.name(), false,
- change -> change.changeUniq(true).convert(HashIndexChange.class)
- .changeColumnNames(pkColumns)
- );
- })
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Error when creating index: " + e);
- }
- });
-
- return completedFuture(false);
- });
+ catalogManager.listen(CatalogEvent.INDEX_CREATE,
+ (params, ex) -> onIndexCreate((CreateIndexEventParameters) params).thenApply(ignore -> false));
+ catalogManager.listen(CatalogEvent.INDEX_DROP,
+ (params, ex) -> onIndexDrop((DropIndexEventParameters) params).thenApply(ignore -> false));
LOG.info("Index manager started");
}
@@ -199,119 +145,19 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
}
try {
+ if (StringUtils.nullOrEmpty(params.indexName())) {
+ return failedFuture(new IgniteInternalException(
+ ErrorGroups.Index.INVALID_INDEX_DEFINITION_ERR,
+ "Index name should be at least 1 character long"
+ ));
+ }
+
CompletableFuture<Void> indexCreateFuture = (params instanceof CreateSortedIndexParams)
? catalogManager.createIndex((CreateSortedIndexParams) params)
: catalogManager.createIndex((CreateHashIndexParams) params);
return indexCreateFuture.thenApply(ignore -> catalogManager.index(params.indexName(), Long.MAX_VALUE))
- .thenCompose(index -> createIndexInternal(
- index.id(),
- params.schemaName(),
- params.indexName(),
- params.tableName(),
- true,
- createIndexChanger(params)
- ));
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- @Deprecated(forRemoval = true)
- private Consumer<TableIndexChange> createIndexChanger(AbstractIndexCommandParams indexParams) {
- if (indexParams instanceof CreateSortedIndexParams) {
- CreateSortedIndexParams params = (CreateSortedIndexParams) indexParams;
- return tableIndexChange -> tableIndexChange.convert(SortedIndexChange.class).changeColumns(colsInit -> {
- for (int i = 0; i < params.columns().size(); i++) {
- String columnName = params.columns().get(i);
- CatalogColumnCollation collation = params.collations().get(i);
- //TODO: https://issues.apache.org/jira/browse/IGNITE-17563 Pass null ordering for columns.
- colsInit.create(columnName, colInit -> colInit.changeAsc(collation.asc()));
- }
- });
- } else {
- CreateHashIndexParams params = (CreateHashIndexParams) indexParams;
- return tableIndexChange -> tableIndexChange.convert(HashIndexChange.class)
- .changeColumnNames(params.columns().toArray(STRING_EMPTY_ARRAY));
- }
- }
-
- @Deprecated(forRemoval = true)
- private CompletableFuture<Boolean> createIndexInternal(
- int indexId,
- String schemaName,
- String indexName,
- String tableName,
- boolean failIfExists,
- Consumer<TableIndexChange> indexChange
- ) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new NodeStoppingException());
- }
-
- LOG.debug("Going to create index [schema={}, table={}, index={}]", schemaName, tableName, indexName);
-
- try {
- validateName(indexName);
-
- CompletableFuture<Boolean> future = new CompletableFuture<>();
-
- // Check index existence flag, avoid usage of hasCause + IndexAlreadyExistsException.
- AtomicBoolean idxExist = new AtomicBoolean(false);
-
- tablesCfg.change(tablesChange -> tablesChange.changeIndexes(indexListChange -> {
- idxExist.set(false);
-
- if (indexListChange.get(indexName) != null) {
- idxExist.set(true);
-
- throw new IndexAlreadyExistsException(schemaName, indexName);
- }
-
- TableView tableCfg = tablesChange.tables().get(tableName);
-
- if (tableCfg == null) {
- throw new TableNotFoundException(schemaName, tableName);
- }
-
- int tableId = tableCfg.id();
-
- tablesChange.changeGlobalIdCounter(indexId);
-
- Consumer<TableIndexChange> chg = indexChange.andThen(c -> c.changeTableId(tableId).changeId(indexId));
-
- indexListChange.create(indexName, chg);
- })).whenComplete((index, th) -> {
- if (th != null) {
- LOG.debug("Unable to create index [schema={}, table={}, index={}]",
- th, schemaName, tableName, indexName);
-
- if (!failIfExists && idxExist.get()) {
- future.complete(false);
- } else {
- future.completeExceptionally(th);
- }
- } else {
- TableIndexConfiguration idxCfg = tablesCfg.indexes().get(indexName);
-
- if (idxCfg != null && idxCfg.value() != null) {
- LOG.info("Index created [schema={}, table={}, index={}, indexId={}]",
- schemaName, tableName, indexName, idxCfg.id().value());
-
- future.complete(true);
- } else {
- var exception = new IgniteInternalException(
- Common.INTERNAL_ERR, "Looks like the index was concurrently deleted");
-
- LOG.info("Unable to create index [schema={}, table={}, index={}]",
- exception, schemaName, tableName, indexName);
-
- future.completeExceptionally(exception);
- }
- }
- });
-
- return future;
+ .thenApply(ignore -> false);
} finally {
busyLock.leaveBusy();
}
@@ -326,89 +172,22 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
}
try {
return catalogManager.dropIndex(params)
- .thenCompose(ignore -> dropIndexAsyncInternal(params.schemaName(), params.indexName(), true));
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- @Deprecated(forRemoval = true)
- private CompletableFuture<Boolean> dropIndexAsyncInternal(
- String schemaName,
- String indexName,
- boolean failIfNotExists
- ) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new NodeStoppingException());
- }
-
- LOG.debug("Going to drop index [schema={}, index={}]", schemaName, indexName);
-
- try {
- validateName(indexName);
-
- final CompletableFuture<Boolean> future = new CompletableFuture<>();
-
- // Check index existence flag, avoid usage of hasCause + IndexAlreadyExistsException.
- AtomicBoolean idxOrTblNotExist = new AtomicBoolean(false);
-
- tablesCfg.indexes().change(indexListChange -> {
- idxOrTblNotExist.set(false);
-
- TableIndexView idxView = indexListChange.get(indexName);
-
- if (idxView == null) {
- idxOrTblNotExist.set(true);
-
- throw new IndexNotFoundException(schemaName, indexName);
- }
-
- indexListChange.delete(indexName);
- }).whenComplete((ignored, th) -> {
- if (th != null) {
- LOG.info("Unable to drop index [schema={}, index={}]", th, schemaName, indexName);
-
- if (!failIfNotExists && idxOrTblNotExist.get()) {
- future.complete(false);
- } else {
- future.completeExceptionally(th);
- }
- } else {
- LOG.info("Index dropped [schema={}, index={}]", schemaName, indexName);
-
- future.complete(true);
- }
- });
-
- return future;
+ .thenApply(ignore -> false);
} finally {
busyLock.leaveBusy();
}
}
- private void validateName(String indexName) {
- if (StringUtils.nullOrEmpty(indexName)) {
- throw new IgniteInternalException(
- ErrorGroups.Index.INVALID_INDEX_DEFINITION_ERR,
- "Index name should be at least 1 character long"
- );
- }
- }
-
/**
* Callback method is called when index configuration changed and an index was dropped.
*
* @param evt Index configuration event.
* @return A future.
*/
- private CompletableFuture<?> onIndexDrop(ConfigurationNotificationEvent<TableIndexView> evt) {
- TableIndexView tableIndexView = evt.oldValue();
-
- int idxId = tableIndexView.id();
-
- int tableId = tableIndexView.tableId();
-
- long causalityToken = evt.storageRevision();
+ private CompletableFuture<?> onIndexDrop(DropIndexEventParameters evt) {
+ int idxId = evt.indexId();
+ int tableId = evt.tableId();
+ long causalityToken = evt.causalityToken();
if (!busyLock.enterBusy()) {
fireEvent(IndexEvent.DROP,
@@ -422,14 +201,20 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
try {
CompletableFuture<?> fireEventFuture = fireEvent(IndexEvent.DROP, new IndexEventParameters(causalityToken, tableId, idxId));
- CompletableFuture<?> dropIndexFuture = tableManager.tableAsync(causalityToken, tableId)
+ catalogManager.table(tableId, evt.catalogVersion());
+ CompletableFuture<?> dropIndexFuture = tableManager.tableAsync(tableId)
.thenAccept(table -> {
if (table != null) { // in case of DROP TABLE the table will be removed first
table.unregisterIndex(idxId);
}
});
+ // TODO: investigate why DropIndexFuture hangs.
return allOf(fireEventFuture, dropIndexFuture);
+ } catch (Throwable th) {
+ LOG.warn("Failed to process drop index event.", th);
+
+ return failedFuture(th);
} finally {
busyLock.leaveBusy();
}
@@ -441,16 +226,12 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
* @param evt Index configuration changed event.
* @return A future.
*/
- private CompletableFuture<?> onIndexCreate(ConfigurationNotificationEvent<TableIndexView> evt) {
- TableIndexView indexConfig = evt.newValue();
-
- int tableId = indexConfig.tableId();
+ private CompletableFuture<?> onIndexCreate(CreateIndexEventParameters evt) {
+ CatalogIndexDescriptor indexDescriptor = evt.indexDescriptor();
if (!busyLock.enterBusy()) {
- int idxId = indexConfig.id();
-
fireEvent(IndexEvent.CREATE,
- new IndexEventParameters(evt.storageRevision(), tableId, idxId),
+ new IndexEventParameters(evt.causalityToken(), indexDescriptor.tableId(), indexDescriptor.id()),
new NodeStoppingException()
);
@@ -458,16 +239,15 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
}
try {
- TablesView tablesView = evt.newValue(TablesView.class);
-
- TableView tableView = findTableView(tablesView.tables(), tableId);
+ CatalogTableDescriptor tableDescriptor = catalogManager.table(indexDescriptor.tableId(), evt.catalogVersion());
- assert tableView != null : "tableId=" + tableId + ", indexId=" + indexConfig.id();
+ assert tableDescriptor != null : "Table not found: tableId=" + indexDescriptor.tableId();
- CatalogTableDescriptor tableDescriptor = toTableDescriptor(tableView);
- CatalogIndexDescriptor indexDescriptor = toIndexDescriptor(indexConfig);
+ return createIndexLocally(evt.causalityToken(), tableDescriptor, indexDescriptor);
+ } catch (Throwable th) {
+ LOG.warn("Failed to process create index event.", th);
- return createIndexLocally(evt.storageRevision(), tableDescriptor, indexDescriptor);
+ return failedFuture(th);
} finally {
busyLock.leaveBusy();
}
@@ -615,28 +395,6 @@ public class IndexManager extends Producer<IndexEvent, IndexEventParameters> imp
}
}
- private class ConfigurationListener implements ConfigurationNamedListListener<TableIndexView> {
- @Override
- public CompletableFuture<?> onCreate(ConfigurationNotificationEvent<TableIndexView> ctx) {
- return onIndexCreate(ctx);
- }
-
- @Override
- public CompletableFuture<?> onRename(ConfigurationNotificationEvent<TableIndexView> ctx) {
- return failedFuture(new UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-16196"));
- }
-
- @Override
- public CompletableFuture<?> onDelete(ConfigurationNotificationEvent<TableIndexView> ctx) {
- return onIndexDrop(ctx);
- }
-
- @Override
- public CompletableFuture<?> onUpdate(ConfigurationNotificationEvent<TableIndexView> ctx) {
- return failedFuture(new IllegalStateException("Should not be called"));
- }
- }
-
/**
* Converts a catalog index descriptor to an event index descriptor.
*
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
index 53b2abf6d2..f0e0cb527e 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
@@ -120,7 +120,7 @@ public class IndexManagerTest {
when(catalogManager.index(anyString(), anyLong())).thenReturn(indexDescriptor);
when(indexDescriptor.id()).thenReturn(1);
- indexManager = new IndexManager(tablesConfig, catalogManager, schManager, tableManagerMock);
+ indexManager = new IndexManager(catalogManager, schManager, tableManagerMock);
indexManager.start();
assertThat(
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 a1ba47148b..0db166eb30 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
@@ -783,7 +783,7 @@ public class ItRebalanceDistributedTest {
clockWaiter
);
- schemaManager = new SchemaManager(registry, tablesCfg, metaStorageManager);
+ schemaManager = new SchemaManager(registry, metaStorageManager, catalogManager);
distributionZoneManager = new DistributionZoneManager(
zonesCfg,
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 10b9f15fb1..1e3141f6c5 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
@@ -345,7 +345,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
clockWaiter
);
- SchemaManager schemaManager = new SchemaManager(registry, tablesConfig, metaStorageMgr);
+ SchemaManager schemaManager = new SchemaManager(registry, metaStorageMgr, catalogManager);
DistributionZoneManager distributionZoneManager = new DistributionZoneManager(
zonesConfig,
@@ -385,7 +385,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
null
);
- var indexManager = new IndexManager(tablesConfig, catalogManager, schemaManager, tableManager);
+ var indexManager = new IndexManager(catalogManager, schemaManager, tableManager);
SqlQueryProcessor qryEngine = new SqlQueryProcessor(
registry,
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 324c7f117c..931dffd433 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
@@ -495,8 +495,6 @@ public class IgniteImpl implements Ignite {
)
);
- schemaManager = new SchemaManager(registry, tablesConfig, metaStorageMgr);
-
SchemaSynchronizationConfiguration schemaSyncConfig = clusterConfigRegistry.getConfiguration(
SchemaSynchronizationConfiguration.KEY
);
@@ -507,6 +505,8 @@ public class IgniteImpl implements Ignite {
() -> schemaSyncConfig.delayDuration().value()
);
+ schemaManager = new SchemaManager(registry, metaStorageMgr, catalogManager);
+
distributionZoneManager = new DistributionZoneManager(
zonesConfiguration,
tablesConfig,
@@ -549,7 +549,7 @@ public class IgniteImpl implements Ignite {
distributionZoneManager
);
- indexManager = new IndexManager(tablesConfig, catalogManager, schemaManager, distributedTblMgr);
+ indexManager = new IndexManager(catalogManager, schemaManager, distributedTblMgr);
qryEngine = new SqlQueryProcessor(
registry,
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/CatalogDescriptorUtils.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/CatalogDescriptorUtils.java
index bcff576238..e400dd1d11 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/CatalogDescriptorUtils.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/CatalogDescriptorUtils.java
@@ -21,6 +21,7 @@ import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation.ASC_NULLS_LAST;
import static org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation.DESC_NULLS_FIRST;
+import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.internal.catalog.commands.CatalogUtils;
import org.apache.ignite.internal.catalog.commands.ColumnParams;
@@ -309,4 +310,53 @@ public class CatalogDescriptorUtils {
throw new IllegalArgumentException("Unknown type " + colType.spec().name());
}
}
+
+ /**
+ * Converts table descriptor to table schema descriptor.
+ */
+ public static SchemaDescriptor convert(int schemaVer, CatalogTableDescriptor tableDescriptor) {
+ List<Column> keyCols = new ArrayList<>(tableDescriptor.primaryKeyColumns().size());
+ List<Column> valCols = new ArrayList<>(tableDescriptor.columns().size() - tableDescriptor.primaryKeyColumns().size());
+
+ int idx = 0;
+
+ for (CatalogTableColumnDescriptor column : tableDescriptor.columns()) {
+ if (tableDescriptor.isPrimaryKeyColumn(column.name())) {
+ keyCols.add(convert(idx, column));
+ } else {
+ valCols.add(convert(idx, column));
+ }
+
+ idx++;
+ }
+
+ return new SchemaDescriptor(
+ schemaVer,
+ keyCols.toArray(Column[]::new),
+ tableDescriptor.colocationColumns().toArray(String[]::new),
+ valCols.toArray(Column[]::new)
+ );
+ }
+
+ private static Column convert(int columnOrder, CatalogTableColumnDescriptor columnDescriptor) {
+ NativeType type = getNativeType(columnDescriptor);
+
+ DefaultValue defaultValue = columnDescriptor.defaultValue();
+
+ DefaultValueProvider defaultValueProvider = toValueProvider(defaultValue);
+
+ return new Column(columnOrder, columnDescriptor.name(), type, columnDescriptor.nullable(), defaultValueProvider);
+ }
+
+ private static DefaultValueProvider toValueProvider(DefaultValue defaultValue) {
+ switch (defaultValue.type()) {
+ case CONSTANT:
+ return DefaultValueProvider.constantProvider(((ConstantValue) defaultValue).value());
+ case FUNCTION_CALL:
+ return DefaultValueProvider.forValueGenerator(
+ DefaultValueGenerator.valueOf(((FunctionCall) defaultValue).functionName()));
+ default:
+ throw new IllegalStateException("Unknown default value type: " + defaultValue.type());
+ }
+ }
}
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 0abadc888d..d61c9389cd 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
@@ -35,8 +35,14 @@ import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.LongFunction;
-import org.apache.ignite.configuration.NamedListView;
-import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.events.AddColumnEventParameters;
+import org.apache.ignite.internal.catalog.events.AlterColumnEventParameters;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CatalogEventParameters;
+import org.apache.ignite.internal.catalog.events.CreateTableEventParameters;
+import org.apache.ignite.internal.catalog.events.DropColumnEventParameters;
import org.apache.ignite.internal.causality.CompletionListener;
import org.apache.ignite.internal.causality.IncrementalVersionedValue;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -45,10 +51,6 @@ import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
-import org.apache.ignite.internal.schema.configuration.ColumnView;
-import org.apache.ignite.internal.schema.configuration.ExtendedTableView;
-import org.apache.ignite.internal.schema.configuration.TableConfiguration;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.schema.event.SchemaEvent;
import org.apache.ignite.internal.schema.event.SchemaEventParameters;
import org.apache.ignite.internal.schema.marshaller.schema.SchemaSerializerImpl;
@@ -79,56 +81,97 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
- /** Tables configuration. */
- private final TablesConfiguration tablesCfg;
-
/** Versioned store for tables by name. */
private final IncrementalVersionedValue<Map<Integer, SchemaRegistryImpl>> registriesVv;
/** Meta storage manager. */
private final MetaStorageManager metastorageMgr;
+ /** Catalog manager. */
+ private final CatalogManager catalogManager;
+
/** Constructor. */
public SchemaManager(
Consumer<LongFunction<CompletableFuture<?>>> registry,
- TablesConfiguration tablesCfg,
- MetaStorageManager metastorageMgr
- ) {
+ MetaStorageManager metastorageMgr,
+ CatalogManager catalogManager) {
this.registriesVv = new IncrementalVersionedValue<>(registry, HashMap::new);
- this.tablesCfg = tablesCfg;
+ this.catalogManager = catalogManager;
this.metastorageMgr = metastorageMgr;
}
/** {@inheritDoc} */
@Override
public void start() {
- tablesCfg.tables().any().columns().listen(this::onSchemaChange);
+ catalogManager.listen(CatalogEvent.TABLE_CREATE,
+ (params, ex) -> onSchemaChange((CreateTableEventParameters) params).thenApply(ignore -> false));
+ catalogManager.listen(CatalogEvent.TABLE_ALTER,
+ (params, ex) -> onSchemaChange(params).thenApply(ignore -> false));
}
/**
* Listener of schema configuration changes.
*
- * @param ctx Configuration context.
+ * @param evt Event parameters.
* @return A future.
*/
- private CompletableFuture<?> onSchemaChange(ConfigurationNotificationEvent<NamedListView<ColumnView>> ctx) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
- }
+ private CompletableFuture<?> onSchemaChange(CreateTableEventParameters evt) {
+ CatalogTableDescriptor tableDescriptor = evt.tableDescriptor();
+
+ int newSchemaVersion = INITIAL_SCHEMA_VERSION;// evt.catalogVersion();
+ int tblId = tableDescriptor.id();
+
+ SchemaDescriptor newSchema = CatalogDescriptorUtils.convert(newSchemaVersion, tableDescriptor);
+
+ return onSchemaChange(tblId, newSchema, evt.causalityToken());
+ }
+ /**
+ * Listener of schema configuration changes.
+ *
+ * @param evt Event parameters.
+ * @return A future.
+ */
+ private CompletableFuture<?> onSchemaChange(CatalogEventParameters evt) {
try {
- ExtendedTableView tblCfg = ctx.newValue(ExtendedTableView.class);
+ CatalogTableDescriptor tableDescriptor;
+
+ if (evt instanceof AddColumnEventParameters) {
+ AddColumnEventParameters params = (AddColumnEventParameters) evt;
- int newSchemaVersion = tblCfg.schemaId();
+ tableDescriptor = catalogManager.table(params.tableId(), params.catalogVersion());
+ } else if (evt instanceof DropColumnEventParameters) {
+ DropColumnEventParameters params = (DropColumnEventParameters) evt;
- int tblId = tblCfg.id();
+ tableDescriptor = catalogManager.table(params.tableId(), params.catalogVersion());
+ } else if (evt instanceof AlterColumnEventParameters) {
+ AlterColumnEventParameters params = (AlterColumnEventParameters) evt;
- if (searchSchemaByVersion(tblId, newSchemaVersion) != null) {
- return completedFuture(null);
+ tableDescriptor = catalogManager.table(params.tableId(), params.catalogVersion());
+ } else {
+ return completedFuture(new UnsupportedOperationException("Unexpected event."));
}
- SchemaDescriptor newSchema = SchemaUtils.prepareSchemaDescriptor(newSchemaVersion, tblCfg);
+ int newSchemaVersion = registriesVv.latest().get(tableDescriptor.id()).lastSchemaVersion() + 1;;
+
+ SchemaDescriptor newSchema = CatalogDescriptorUtils.convert(newSchemaVersion, tableDescriptor);
+
+ return onSchemaChange(tableDescriptor.id(), newSchema, evt.causalityToken());
+ } catch (Throwable th) {
+ th.printStackTrace();
+ return failedFuture(th);
+ }
+ }
+ /**
+ * Update schema.
+ */
+ private CompletableFuture<?> onSchemaChange(int tblId, SchemaDescriptor newSchema, long causalityToken) {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
+ }
+
+ try {
// This is intentionally a blocking call to enforce configuration listener execution order. Unfortunately it is not possible
// to execute this method asynchronously, because the schema descriptor is needed to fire the CREATE event as a synchronous part
// of the configuration listener.
@@ -142,26 +185,15 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
return failedFuture(e);
}
- long causalityToken = ctx.storageRevision();
-
- // Fire event early, because dependent listeners have to register VersionedValues' update futures
- var eventParams = new SchemaEventParameters(causalityToken, tblId, newSchema);
-
- fireEvent(SchemaEvent.CREATE, eventParams)
- .whenComplete((v, e) -> {
+ return registriesVv.update(causalityToken,
+ (registries, e) -> inBusyLock(busyLock, () -> {
if (e != null) {
- LOGGER.warn("Error when processing CREATE event", e);
+ return failedFuture(new IgniteInternalException(IgniteStringFormatter.format(
+ "Cannot create a schema for the table [tblId={}, ver={}]", tblId, newSchema.version()), e)
+ );
}
- });
- return registriesVv.update(causalityToken, (registries, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(new IgniteInternalException(IgniteStringFormatter.format(
- "Cannot create a schema for the table [tblId={}, ver={}]", tblId, newSchemaVersion), e)
- );
- }
-
- return registerSchema(registries, tblId, tblCfg.name(), newSchema);
+ return registerSchema(registries, tblId, newSchema);
}));
} finally {
busyLock.leaveBusy();
@@ -191,7 +223,6 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
*
* @param registries Map of schema registries.
* @param tableId Table id.
- * @param tableName Table name.
* @param schema Schema descriptor.
* @return Future that, when complete, will resolve into an updated map of schema registries
* (to be used in {@link IncrementalVersionedValue#update}).
@@ -199,42 +230,39 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
private CompletableFuture<Map<Integer, SchemaRegistryImpl>> registerSchema(
Map<Integer, SchemaRegistryImpl> registries,
int tableId,
- String tableName,
SchemaDescriptor schema
) {
ByteArray key = schemaWithVerHistKey(tableId, schema.version());
byte[] serializedSchema = SchemaSerializerImpl.INSTANCE.serialize(schema);
- return metastorageMgr.invoke(notExists(key), put(key, serializedSchema), noop())
- .thenApply(t -> {
- SchemaRegistryImpl reg = registries.get(tableId);
+ SchemaRegistryImpl reg = registries.get(tableId);
- if (reg == null) {
- Map<Integer, SchemaRegistryImpl> copy = new HashMap<>(registries);
+ if (reg == null) {
+ Map<Integer, SchemaRegistryImpl> copy = new HashMap<>(registries);
- copy.put(tableId, createSchemaRegistry(tableId, tableName, schema));
+ copy.put(tableId, createSchemaRegistry(tableId, schema));
- return copy;
- } else {
- reg.onSchemaRegistered(schema);
+ registries = copy;
+ } else {
+ reg.onSchemaRegistered(schema);
+ }
- return registries;
- }
- });
+ metastorageMgr.invoke(notExists(key), put(key, serializedSchema), noop());
+
+ return completedFuture(registries);
}
/**
* Create schema registry for the table.
*
* @param tableId Table id.
- * @param tableName Table name.
* @param initialSchema Initial schema for the registry.
* @return Schema registry.
*/
- private SchemaRegistryImpl createSchemaRegistry(int tableId, String tableName, SchemaDescriptor initialSchema) {
+ private SchemaRegistryImpl createSchemaRegistry(int tableId, SchemaDescriptor initialSchema) {
return new SchemaRegistryImpl(
- ver -> inBusyLock(busyLock, () -> tableSchema(tableId, tableName, ver)),
+ ver -> inBusyLock(busyLock, () -> tableSchema(tableId, ver)),
() -> inBusyLock(busyLock, () -> latestSchemaVersion(tableId)),
initialSchema
);
@@ -247,15 +275,13 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
* @param schemaVer Schema version.
* @return Schema descriptor.
*/
- private CompletableFuture<SchemaDescriptor> tableSchema(int tblId, String tableName, int schemaVer) {
- TableConfiguration tblCfg = tablesCfg.tables().get(tableName);
-
+ private CompletableFuture<SchemaDescriptor> tableSchema(int tblId, int schemaVer) {
CompletableFuture<SchemaDescriptor> fut = new CompletableFuture<>();
SchemaRegistry registry = registriesVv.latest().get(tblId);
if (registry.lastSchemaVersion() > schemaVer) {
- return getSchemaDescriptor(schemaVer, tblCfg);
+ return getSchemaDescriptor(schemaVer, tblId);
}
CompletionListener<Map<Integer, SchemaRegistryImpl>> schemaListener = (token, regs, e) -> {
@@ -328,11 +354,11 @@ public class SchemaManager extends Producer<SchemaEvent, SchemaEventParameters>
* Gets a schema descriptor from the configuration storage.
*
* @param schemaVer Schema version.
- * @param tblCfg Table configuration.
+ * @param tblId
* @return Schema descriptor.
*/
- private CompletableFuture<SchemaDescriptor> getSchemaDescriptor(int schemaVer, TableConfiguration tblCfg) {
- CompletableFuture<Entry> ent = metastorageMgr.get(schemaWithVerHistKey(tblCfg.id().value(), schemaVer));
+ private CompletableFuture<SchemaDescriptor> getSchemaDescriptor(int schemaVer, Integer tblId) {
+ CompletableFuture<Entry> ent = metastorageMgr.get(schemaWithVerHistKey(tblId, schemaVer));
return ent.thenApply(e -> SchemaSerializerImpl.INSTANCE.deserialize(e.value()));
}
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 a0b5c9bc24..010e7b1e4d 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
@@ -40,6 +40,7 @@ import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.util.Pair;
import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.index.IndexManager;
@@ -68,6 +69,7 @@ import org.apache.ignite.internal.sql.engine.prepare.PrepareService;
import org.apache.ignite.internal.sql.engine.prepare.PrepareServiceImpl;
import org.apache.ignite.internal.sql.engine.property.PropertiesHelper;
import org.apache.ignite.internal.sql.engine.property.PropertiesHolder;
+import org.apache.ignite.internal.sql.engine.schema.CatalogSqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
import org.apache.ignite.internal.sql.engine.session.Session;
@@ -238,14 +240,6 @@ public class SqlQueryProcessor implements QueryProcessor {
msgSrvc
));
- SqlSchemaManagerImpl sqlSchemaManager = new SqlSchemaManagerImpl(
- tableManager,
- schemaManager,
- registry,
- busyLock
- );
-
- sqlSchemaManager.registerListener(prepareSvc);
this.prepareSvc = prepareSvc;
@@ -261,7 +255,8 @@ public class SqlQueryProcessor implements QueryProcessor {
var dependencyResolver = new ExecutionDependencyResolverImpl(executableTableRegistry);
- sqlSchemaManager.registerListener(executableTableRegistry);
+ CatalogSqlSchemaManager sqlSchemaManager = new CatalogSqlSchemaManager(catalogManager, 1000);
+ this.sqlSchemaManager = sqlSchemaManager;
var executionSrvc = registerService(ExecutionServiceImpl.create(
clusterSrvc.topologyService(),
@@ -272,7 +267,8 @@ public class SqlQueryProcessor implements QueryProcessor {
ArrayRowHandler.INSTANCE,
mailboxRegistry,
exchangeService,
- dependencyResolver
+ dependencyResolver,
+ tableManager
));
clusterSrvc.topologyService().addEventHandler(executionSrvc);
@@ -280,14 +276,6 @@ public class SqlQueryProcessor implements QueryProcessor {
this.executionSrvc = executionSrvc;
- registerTableListener(TableEvent.CREATE, new TableCreatedListener(sqlSchemaManager));
- registerTableListener(TableEvent.ALTER, new TableUpdatedListener(sqlSchemaManager));
- registerTableListener(TableEvent.DROP, new TableDroppedListener(sqlSchemaManager));
-
- registerIndexListener(IndexEvent.CREATE, new IndexCreatedListener(sqlSchemaManager));
- registerIndexListener(IndexEvent.DROP, new IndexDroppedListener(sqlSchemaManager));
-
- this.sqlSchemaManager = sqlSchemaManager;
services.forEach(LifecycleAware::start);
}
@@ -416,7 +404,11 @@ public class SqlQueryProcessor implements QueryProcessor {
AtomicReference<InternalTransaction> tx = new AtomicReference<>();
CompletableFuture<AsyncSqlCursor<List<Object>>> stage = start
- .thenCompose(ignored -> {
+ // TODO: wait for latest catalog.
+ .thenCompose(ignore -> tableManager.tablesAsync())
+ .thenCompose(tables -> CompletableFuture.allOf(tables.stream()
+ .map(t -> tableManager.tableAsync(t.name())).toArray(CompletableFuture[]::new))
+ ).thenCompose(ignored -> {
ParsedResult result = parserService.parse(sql);
validateParsedStatement(context, outerTx, result, params);
@@ -427,7 +419,9 @@ public class SqlQueryProcessor implements QueryProcessor {
tx.set(implicitTxRequired ? txManager.begin(!rwOp, null) : outerTx);
- SchemaPlus schema = sqlSchemaManager.schema(schemaName);
+ int catalogVersion = catalogManager.activeCatalogVersion(tx.get().startTimestamp().longValue());
+
+ SchemaPlus schema = sqlSchemaManager.schema(schemaName, catalogVersion);
if (schema == null) {
return CompletableFuture.failedFuture(new SchemaNotFoundException(schemaName));
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 65e878da53..ea12b0a4e0 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
@@ -235,13 +235,13 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return queryManager.execute(tx, plan);
}
- private BaseQueryContext createQueryContext(UUID queryId, @Nullable String schema, Object[] params) {
+ private BaseQueryContext createQueryContext(UUID queryId, @Nullable String schema, int version, Object[] params) {
return BaseQueryContext.builder()
.queryId(queryId)
.parameters(params)
.frameworkConfig(
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(sqlSchemaManager.schema(schema))
+ .defaultSchema(sqlSchemaManager.schema(schema, version))
.build()
)
.logger(LOG)
@@ -423,7 +423,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private DistributedQueryManager getOrCreateQueryManager(QueryStartRequest msg) {
return queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
- BaseQueryContext ctx = createQueryContext(key, msg.schema(), msg.parameters());
+ //TODO: change type to int.
+ BaseQueryContext ctx = createQueryContext(key, msg.schema(), (int) msg.schemaVersion(), msg.parameters());
return new DistributedQueryManager(ctx);
});
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java
index ed27b83e91..4f75c4e454 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java
@@ -68,10 +68,12 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
}
/** {@inheritDoc} */
+ @Deprecated
@Override
public SchemaPlus schema(@Nullable String schema) {
// Should be removed -schema(name, version) must be used instead
- throw new UnsupportedOperationException();
+ // throw new UnsupportedOperationException();
+ return activeSchema(schema, catalogManager.activeCatalogVersion(Long.MAX_VALUE));
}
/** {@inheritDoc} */
@@ -87,7 +89,12 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
@Override
public IgniteTable tableById(int id) {
// Should be removed - this method is used to obtain native types from a table.
- throw new UnsupportedOperationException();
+ //throw new UnsupportedOperationException();
+ CatalogTableDescriptor table = catalogManager.table(id, Long.MAX_VALUE);
+
+ assert table != null;
+
+ return (IgniteTable) activeSchema(DEFAULT_SCHEMA_NAME, Long.MAX_VALUE).getTable(table.name());
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java
index c6a8cc0743..37cae7bd55 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/BaseQueryContext.java
@@ -52,7 +52,7 @@ import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.metadata.cost.IgniteCostFactory;
import org.apache.ignite.internal.sql.engine.rex.IgniteRexBuilder;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.apache.ignite.internal.util.ArrayUtils;
@@ -222,7 +222,7 @@ public final class BaseQueryContext extends AbstractQueryContext {
}
public long schemaVersion() {
- return Objects.requireNonNull(schema().unwrap(IgniteSchema.class)).schemaVersion();
+ return Objects.requireNonNull(schema().unwrap(IgniteCatalogSchema.class)).version();
}
/**
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 955a4223c2..26a355f767 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
@@ -281,7 +281,7 @@ public class MockedStructuresTest extends IgniteAbstractTest {
dataStorageManager.start();
- schemaManager = new SchemaManager(revisionUpdater, tblsCfg, msm);
+ schemaManager = new SchemaManager(revisionUpdater, msm, catalogManager);
schemaManager.start();
@@ -313,7 +313,7 @@ public class MockedStructuresTest extends IgniteAbstractTest {
tblManager = mockManagers();
- idxManager = new IndexManager(tblsCfg, catalogManager, schemaManager, tblManager);
+ idxManager = new IndexManager(catalogManager, schemaManager, tblManager);
idxManager.start();
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 f7ca64053d..f0b9b39f7b 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
@@ -46,6 +46,7 @@ import static org.apache.ignite.internal.utils.RebalanceUtil.extractTableId;
import static org.apache.ignite.internal.utils.RebalanceUtil.pendingPartAssignmentsKey;
import static org.apache.ignite.internal.utils.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
+import static org.apache.ignite.lang.IgniteStringFormatter.format;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import java.io.IOException;
@@ -100,7 +101,6 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogDataStorageDescript
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
-import org.apache.ignite.internal.catalog.events.CatalogEventParameters;
import org.apache.ignite.internal.catalog.events.CreateTableEventParameters;
import org.apache.ignite.internal.catalog.events.DropTableEventParameters;
import org.apache.ignite.internal.causality.CompletionListener;
@@ -113,7 +113,6 @@ import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
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.manager.IgniteComponent;
import org.apache.ignite.internal.manager.Producer;
import org.apache.ignite.internal.metastorage.Entry;
@@ -151,8 +150,6 @@ import org.apache.ignite.internal.schema.configuration.TablesChange;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
import org.apache.ignite.internal.schema.configuration.storage.DataStorageView;
-import org.apache.ignite.internal.schema.event.SchemaEvent;
-import org.apache.ignite.internal.schema.event.SchemaEventParameters;
import org.apache.ignite.internal.storage.DataStorageManager;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
@@ -536,18 +533,9 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
catalogManager.listen(CatalogEvent.TABLE_CREATE,
(parameters, exception) -> onTableCreate((CreateTableEventParameters) parameters).thenApply(ignore -> false));
- catalogManager.listen(CatalogEvent.TABLE_CREATE,
+ catalogManager.listen(CatalogEvent.TABLE_DROP,
(parameters, exception) -> onTableDelete((DropTableEventParameters) parameters).thenApply(ignore -> false));
- schemaManager.listen(SchemaEvent.CREATE, new EventListener<>() {
- @Override
- public CompletableFuture<Boolean> notify(@NotNull SchemaEventParameters parameters, @Nullable Throwable exception) {
- var eventParameters = new TableEventParameters(parameters.causalityToken(), parameters.tableId());
-
- return fireEvent(TableEvent.ALTER, eventParameters).thenApply(v -> false);
- }
- });
-
addMessageHandler(clusterService.messagingService());
}
@@ -591,13 +579,13 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/**
* Listener of table create configuration change.
*
- * @param ctx Table configuration context.
+ * @param evt Event parameters.
* @return A future.
*/
- private CompletableFuture<?> onTableCreate(CreateTableEventParameters ctx) {
+ private CompletableFuture<?> onTableCreate(CreateTableEventParameters evt) {
if (!busyLock.enterBusy()) {
fireEvent(TableEvent.CREATE,
- new TableEventParameters(ctx.causalityToken(), ctx.tableDescriptor().id()),
+ new TableEventParameters(evt.causalityToken(), evt.tableDescriptor().id()),
new NodeStoppingException()
);
@@ -605,10 +593,8 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
try {
- CatalogTableDescriptor tableDescriptor = ctx.tableDescriptor();
- CatalogZoneDescriptor zoneDescriptor = catalogManager.zone(tableDescriptor.zoneId(), ctx.catalogVersion());
-
- assert zoneDescriptor != null;
+ CatalogTableDescriptor tableDescriptor = evt.tableDescriptor();
+ CatalogZoneDescriptor zoneDescriptor = catalogManager.zone(tableDescriptor.zoneId(), evt.catalogVersion());
List<Set<Assignment>> assignments;
@@ -630,7 +616,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
assert !assignments.isEmpty() : "Couldn't create the table with empty assignments.";
CompletableFuture<?> createTableFut = createTableLocally(
- ctx.causalityToken(),
+ evt.causalityToken(),
tableDescriptor,
zoneDescriptor,
assignments
@@ -647,6 +633,10 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
writeTableAssignmentsToMetastore(tableId, assignments);
return createTableFut;
+ } catch (Throwable th) {
+ LOG.warn("Failed to process create table event.", th);
+
+ return failedFuture(th);
} finally {
busyLock.leaveBusy();
}
@@ -678,14 +668,14 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
/**
* Listener of table drop configuration change.
*
- * @param ctx Table configuration context.
+ * @param evt Event parameters.
* @return A future.
*/
- private CompletableFuture<?> onTableDelete(DropTableEventParameters ctx) {
+ private CompletableFuture<?> onTableDelete(DropTableEventParameters evt) {
if (!busyLock.enterBusy()) {
fireEvent(
TableEvent.DROP,
- new TableEventParameters(ctx.causalityToken(), ctx.tableId()),
+ new TableEventParameters(evt.causalityToken(), evt.tableId()),
new NodeStoppingException()
);
@@ -693,10 +683,16 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
try {
- CatalogTableDescriptor tableDescriptor = catalogManager.table(ctx.tableId(), ctx.catalogVersion() - 1);
- CatalogZoneDescriptor zoneDescriptor = catalogManager.zone(tableDescriptor.zoneId(), ctx.catalogVersion() - 1);
+ int previousCatalogVersion = evt.catalogVersion() - 1;
+
+ CatalogTableDescriptor tableDescriptor = catalogManager.table(evt.tableId(), previousCatalogVersion);
+ CatalogZoneDescriptor zoneDescriptor = catalogManager.zone(tableDescriptor.zoneId(), previousCatalogVersion);
- dropTableLocally(ctx.causalityToken(), tableDescriptor, zoneDescriptor);
+ dropTableLocally(evt.causalityToken(), tableDescriptor, zoneDescriptor);
+ } catch (Throwable th) {
+ LOG.warn("Failed to process drop table event.", th);
+
+ return failedFuture(th);
} finally {
busyLock.leaveBusy();
}
@@ -725,7 +721,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
// Empty assignments might be a valid case if tables are created from within cluster init HOCON
// configuration, which is not supported now.
- assert newAssignments != null : IgniteStringFormatter.format("Table [id={}] has empty assignments.", tableId);
+ assert newAssignments != null : format("Table [id={}] has empty assignments.", tableId);
int partitions = newAssignments.size();
@@ -1286,18 +1282,22 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
pendingTables.put(tableId, table);
startedTables.put(tableId, table);
- tablesById(causalityToken).thenAccept(ignored -> inBusyLock(busyLock, () -> {
+ createPartsFut.thenAccept(ignored -> inBusyLock(busyLock, () -> {
pendingTables.remove(tableId);
}));
- tablesById(causalityToken)
- .thenRun(() -> inBusyLock(busyLock, () -> completeApiCreateFuture(table)));
+ createPartsFut
+ .thenRunAsync(() -> inBusyLock(busyLock, () -> completeApiCreateFuture(table)));
// TODO should be reworked in IGNITE-16763
// We use the event notification future as the result so that dependent components can complete the schema updates.
// TODO: https://issues.apache.org/jira/browse/IGNITE-19913 Possible performance degradation.
- return allOf(createPartsFut, fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, tableId)));
+ CompletableFuture<?> eventFut = fireEvent(TableEvent.CREATE, new TableEventParameters(causalityToken, tableId));
+
+ // TODO: investigate why createParts and eventFutures hangs.
+// return allOf(createPartsFut, eventFut);
+ return completedFuture(false);
}
/**
@@ -1506,6 +1506,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
.changeColocationColumns(colocationKeys0.toArray(String[]::new)));
};
+
return catalogManager.createTable(parameters)
.thenApply(ignore -> catalogManager.table(tableName, Long.MAX_VALUE).id())
.thenCompose(tableId -> createTableAsyncInternal(tableId, tableName, zoneName, tblChanger));
@@ -2889,8 +2890,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
}
if (primaryCols.contains(colName)) {
- throw new SqlException(STMT_VALIDATION_ERR, IgniteStringFormatter
- .format("Can`t delete column, belongs to primary key: [name={}]", colName));
+ throw new SqlException(STMT_VALIDATION_ERR, format("Can`t delete column, belongs to primary key: [name={}]", colName));
}
}
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 22aa333a72..7fc676d9b3 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
@@ -23,7 +23,6 @@ import static org.apache.ignite.internal.distributionzones.DistributionZoneManag
import static org.apache.ignite.internal.distributionzones.DistributionZoneManager.DEFAULT_ZONE_NAME;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.getZoneById;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
-import static org.apache.ignite.internal.testframework.IgniteTestUtils.cause;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -38,8 +37,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -71,10 +68,7 @@ import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogServiceImpl;
import org.apache.ignite.internal.catalog.ClockWaiter;
import org.apache.ignite.internal.catalog.commands.AlterTableAddColumnParams;
-import org.apache.ignite.internal.catalog.commands.CreateHashIndexParams;
-import org.apache.ignite.internal.catalog.commands.CreateSortedIndexParams;
import org.apache.ignite.internal.catalog.commands.DropTableParams;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
@@ -85,7 +79,6 @@ import org.apache.ignite.internal.configuration.notifications.ConfigurationStora
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.configuration.testframework.InjectRevisionListenerHolder;
-import org.apache.ignite.internal.distributionzones.DistributionZoneConfigurationParameters;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClockImpl;
@@ -867,8 +860,6 @@ public class TableManagerTest extends IgniteAbstractTest {
when(vaultManager.get(any(ByteArray.class))).thenReturn(completedFuture(null));
when(vaultManager.put(any(ByteArray.class), any(byte[].class))).thenReturn(completedFuture(null));
- //TODO IGNITE-19082 drop mocked catalog manager.
-
CatalogManager catalogManager = new CatalogServiceImpl(
new UpdateLogImpl(StandaloneMetaStorageManager.create(vaultManager)),
new ClockWaiter("note", new HybridClockImpl()));
@@ -893,7 +884,7 @@ public class TableManagerTest extends IgniteAbstractTest {
workDir,
msm,
catalogManager,
- sm = new SchemaManager(revisionUpdater, tblsCfg, msm),
+ sm = new SchemaManager(revisionUpdater, msm, catalogManager),
budgetView -> new LocalLogStorageFactory(),
new HybridClockImpl(),
new OutgoingSnapshotsManager(clusterService.messagingService()),