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/08/01 20:40:30 UTC
[ignite-3] branch ignite-19497 created (now 22d25603af)
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a change to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at 22d25603af Styles.
This branch includes the following new commits:
new 8a00476c66 Fix incorrect IgniteSchemaTable.unwrap.
new bbf8eb5ed1 Use schema versions for queries. * Use actual schema for query in TX. * Pass schema version to remote fragment. * Wait for catalog schema ready on the remote side.
new a1ba302325 Fix HashFunctionFactoryImpl to use specific schema version. Fix SqlSchemaManager interface. Drop unused code.
new 82b1252d3d Replace IgniteTableImpl with IgniteCatalogImpl.
new 8276700f76 Replace IgniteIndex with IgniteSchemaIndex
new 22d25603af Styles.
The 6 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] 03/06: Fix HashFunctionFactoryImpl to use specific schema version. Fix SqlSchemaManager interface. Drop unused code.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit a1ba302325fd824a8a124d7a5b4b14523fda718e
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 21:01:30 2023 +0300
Fix HashFunctionFactoryImpl to use specific schema version.
Fix SqlSchemaManager interface.
Drop unused code.
---
.../ignite/internal/sql/api/ItCommonApiTest.java | 17 +-
.../internal/sql/engine/SqlQueryProcessor.java | 146 +-----
.../sql/engine/exec/ExecutionServiceImpl.java | 3 +-
.../sql/engine/schema/CatalogSqlSchemaManager.java | 18 +-
.../sql/engine/schema/SqlSchemaManager.java | 24 +-
.../sql/engine/schema/SqlSchemaManagerImpl.java | 584 ---------------------
.../internal/sql/engine/util/BaseQueryContext.java | 4 +-
.../sql/engine/util/HashFunctionFactoryImpl.java | 19 +-
.../sql/engine/exec/ExecutionServiceImplTest.java | 7 +-
.../engine/exec/schema/SqlSchemaManagerTest.java | 329 ------------
.../engine/framework/PredefinedSchemaManager.java | 16 +-
.../internal/sql/engine/framework/TestNode.java | 4 +-
.../engine/schema/CatalogSqlSchemaManagerTest.java | 22 +-
13 files changed, 55 insertions(+), 1138 deletions(-)
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItCommonApiTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItCommonApiTest.java
index 5b46b788bb..a35925eb5a 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItCommonApiTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItCommonApiTest.java
@@ -34,7 +34,6 @@ import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.sql.engine.ClusterPerClassIntegrationTest;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.sql.engine.exec.ExecutionCancelledException;
-import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -222,12 +221,6 @@ public class ItCommonApiTest extends ClusterPerClassIntegrationTest {
}
private static class ErroneousSchemaManager implements SqlSchemaManager {
- /** {@inheritDoc} */
- @Override
- public SchemaPlus schema(@Nullable String schema) {
- return null;
- }
-
/** {@inheritDoc} */
@Override
public SchemaPlus schema(@Nullable String name, int version) {
@@ -236,19 +229,13 @@ public class ItCommonApiTest extends ClusterPerClassIntegrationTest {
/** {@inheritDoc} */
@Override
- public IgniteTable tableById(int id) {
- throw new UnsupportedOperationException();
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<SchemaPlus> actualSchemaAsync(long ver) {
+ public CompletableFuture<?> actualSchemaAsync(long ver) {
throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Override
- public SchemaPlus activeSchema(@Nullable String name, long timestamp) {
+ public SchemaPlus latestSchema(@Nullable String name) {
throw new UnsupportedOperationException();
}
}
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 9098326e46..3b19070984 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
@@ -44,7 +44,6 @@ import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.index.IndexManager;
import org.apache.ignite.internal.index.event.IndexEvent;
-import org.apache.ignite.internal.index.event.IndexEventParameters;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.manager.Event;
@@ -68,8 +67,8 @@ 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;
import org.apache.ignite.internal.sql.engine.session.SessionId;
import org.apache.ignite.internal.sql.engine.session.SessionInfo;
@@ -85,7 +84,6 @@ import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.storage.DataStorageManager;
import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.table.event.TableEvent;
-import org.apache.ignite.internal.table.event.TableEventParameters;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
@@ -95,11 +93,10 @@ import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.lang.SchemaNotFoundException;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.sql.SqlException;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
- * SqlQueryProcessor.
+ * SqlQueryProcessor.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
public class SqlQueryProcessor implements QueryProcessor {
@@ -205,15 +202,10 @@ public class SqlQueryProcessor implements QueryProcessor {
this.clock = clock;
this.catalogManager = catalogManager;
- sqlSchemaManager = new SqlSchemaManagerImpl(
- tableManager,
- schemaManager,
- registry,
- busyLock
+ sqlSchemaManager = new CatalogSqlSchemaManager(
+ catalogManager,
+ 0 // TOOD fix cache size
);
-
- registerTableListener(TableEvent.CREATE, new TableCreatedListener((SqlSchemaManagerImpl) sqlSchemaManager));
- registerIndexListener(IndexEvent.CREATE, new IndexCreatedListener((SqlSchemaManagerImpl) sqlSchemaManager));
}
/** {@inheritDoc} */
@@ -245,8 +237,6 @@ public class SqlQueryProcessor implements QueryProcessor {
msgSrvc
));
- ((SqlSchemaManagerImpl) sqlSchemaManager).registerListener(prepareSvc);
-
this.prepareSvc = prepareSvc;
var ddlCommandHandler = new DdlCommandHandlerWrapper(
@@ -261,8 +251,6 @@ public class SqlQueryProcessor implements QueryProcessor {
var dependencyResolver = new ExecutionDependencyResolverImpl(executableTableRegistry);
- ((SqlSchemaManagerImpl) sqlSchemaManager).registerListener(executableTableRegistry);
-
var executionSrvc = registerService(ExecutionServiceImpl.create(
clusterSrvc.topologyService(),
msgSrvc,
@@ -280,11 +268,6 @@ public class SqlQueryProcessor implements QueryProcessor {
this.executionSrvc = executionSrvc;
- registerTableListener(TableEvent.ALTER, new TableUpdatedListener(((SqlSchemaManagerImpl) sqlSchemaManager)));
- registerTableListener(TableEvent.DROP, new TableDroppedListener(((SqlSchemaManagerImpl) sqlSchemaManager)));
-
- registerIndexListener(IndexEvent.DROP, new IndexDroppedListener(((SqlSchemaManagerImpl) sqlSchemaManager)));
-
services.forEach(LifecycleAware::start);
}
@@ -364,18 +347,6 @@ public class SqlQueryProcessor implements QueryProcessor {
return service;
}
- private void registerTableListener(TableEvent evt, AbstractTableEventListener lsnr) {
- evtLsnrs.add(Pair.of(evt, lsnr));
-
- tableManager.listen(evt, lsnr);
- }
-
- private void registerIndexListener(IndexEvent evt, AbstractIndexEventListener lsnr) {
- evtLsnrs.add(Pair.of(evt, lsnr));
-
- indexManager.listen(evt, lsnr);
- }
-
private CompletableFuture<AsyncSqlCursor<List<Object>>> querySingle0(
SessionId sessionId,
QueryContext context,
@@ -490,113 +461,6 @@ public class SqlQueryProcessor implements QueryProcessor {
return stage;
}
- private abstract static class AbstractTableEventListener implements EventListener<TableEventParameters> {
- protected final SqlSchemaManagerImpl schemaHolder;
-
- private AbstractTableEventListener(SqlSchemaManagerImpl schemaHolder) {
- this.schemaHolder = schemaHolder;
- }
- }
-
- private abstract static class AbstractIndexEventListener implements EventListener<IndexEventParameters> {
- protected final SqlSchemaManagerImpl schemaHolder;
-
- private AbstractIndexEventListener(SqlSchemaManagerImpl schemaHolder) {
- this.schemaHolder = schemaHolder;
- }
- }
-
- private static class TableCreatedListener extends AbstractTableEventListener {
- private TableCreatedListener(SqlSchemaManagerImpl schemaHolder) {
- super(schemaHolder);
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- return schemaHolder.onTableCreated(
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
- DEFAULT_SCHEMA_NAME,
- parameters.tableId(),
- parameters.causalityToken()
- )
- .thenApply(v -> false);
- }
- }
-
- private static class TableUpdatedListener extends AbstractTableEventListener {
- private TableUpdatedListener(SqlSchemaManagerImpl schemaHolder) {
- super(schemaHolder);
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- return schemaHolder.onTableUpdated(
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
- DEFAULT_SCHEMA_NAME,
- parameters.tableId(),
- parameters.causalityToken()
- )
- .thenApply(v -> false);
- }
- }
-
- private static class TableDroppedListener extends AbstractTableEventListener {
- private TableDroppedListener(SqlSchemaManagerImpl schemaHolder) {
- super(schemaHolder);
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<Boolean> notify(@NotNull TableEventParameters parameters, @Nullable Throwable exception) {
- return schemaHolder.onTableDropped(
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
- DEFAULT_SCHEMA_NAME,
- parameters.tableId(),
- parameters.causalityToken()
- )
- .thenApply(v -> false);
- }
- }
-
- private static class IndexDroppedListener extends AbstractIndexEventListener {
- private IndexDroppedListener(SqlSchemaManagerImpl schemaHolder) {
- super(schemaHolder);
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<Boolean> notify(@NotNull IndexEventParameters parameters, @Nullable Throwable exception) {
- return schemaHolder.onIndexDropped(
- // TODO: https://issues.apache.org/jira/browse/IGNITE-17694 Hardcoded schemas
- DEFAULT_SCHEMA_NAME,
- parameters.tableId(),
- parameters.indexId(),
- parameters.causalityToken()
- )
- .thenApply(v -> false);
- }
- }
-
- private static class IndexCreatedListener extends AbstractIndexEventListener {
- private IndexCreatedListener(SqlSchemaManagerImpl schemaHolder) {
- super(schemaHolder);
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<Boolean> notify(@NotNull IndexEventParameters parameters, @Nullable Throwable exception) {
- return schemaHolder.onIndexCreated(
- parameters.tableId(),
- parameters.indexId(),
- parameters.indexDescriptor(),
- parameters.causalityToken()
- )
- .thenApply(v -> false);
- }
- }
-
/** Returns {@code true} if this is data modification operation. */
private static boolean dataModificationOp(ParsedResult parsedResult) {
return parsedResult.queryType() == SqlQueryType.DML;
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 50428f0b91..e330ebd4ed 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
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
import org.apache.ignite.internal.sql.engine.rel.IgniteTableModify;
import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.rel.SourceAwareIgniteRel;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
@@ -172,7 +173,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
dependencyResolver,
(ctx, deps) -> new LogicalRelImplementor<>(
ctx,
- new HashFunctionFactoryImpl<>(sqlSchemaManager, handler),
+ new HashFunctionFactoryImpl<>(ctx.getRootSchema().unwrap(IgniteCatalogSchema.class), handler),
mailboxRegistry,
exchangeSrvc,
deps)
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 371f885abc..432ad5834c 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
@@ -67,13 +67,6 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
this.cache = Caffeine.newBuilder().maximumSize(cacheSize).<Map.Entry<String, Integer>, SchemaPlus>build().asMap();
}
- /** {@inheritDoc} */
- @Override
- public SchemaPlus schema(@Nullable String schema) {
- // Should be removed -schema(name, version) must be used instead
- throw new UnsupportedOperationException();
- }
-
/** {@inheritDoc} */
@Override
public SchemaPlus schema(String name, int version) {
@@ -83,13 +76,6 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
return cache.computeIfAbsent(entry, (e) -> createSqlSchema(e.getValue(), catalogManager.schema(e.getKey(), e.getValue())));
}
- /** {@inheritDoc} */
- @Override
- public IgniteTable tableById(int id) {
- // Should be removed - this method is used to obtain native types from a table.
- throw new UnsupportedOperationException();
- }
-
/** {@inheritDoc} */
@Override
public CompletableFuture<?> actualSchemaAsync(long ver) {
@@ -98,10 +84,10 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus activeSchema(@Nullable String name, long timestamp) {
+ public SchemaPlus latestSchema(@Nullable String name) {
String schemaName = name == null ? DEFAULT_SCHEMA_NAME : name;
- int version = catalogManager.activeCatalogVersion(timestamp);
+ int version = catalogManager.activeCatalogVersion(Long.MAX_VALUE);
CatalogSchemaDescriptor descriptor = catalogManager.schema(schemaName, version);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
index e2c1fb6e1b..25eb750e36 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
@@ -20,29 +20,19 @@ package org.apache.ignite.internal.sql.engine.schema;
import java.util.concurrent.CompletableFuture;
import org.apache.calcite.schema.SchemaPlus;
import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.TestOnly;
/**
* Sql schemas operations interface.
*/
public interface SqlSchemaManager {
- /**
- * Returns a required schema if specified, or default schema otherwise.
- */
- SchemaPlus schema(@Nullable String schema);
-
/**
* Returns schema with given name and by the given version, if name is not specified, returns default schema of the given version.
- */
- SchemaPlus schema(@Nullable String name, int version);
-
- /**
- * Returns a table by given id.
- *
- * @param id An id of required table.
*
- * @return The table.
+ * @param schemaName Schema name. If {@code null}, then default schema name will be used.
+ * @param version Catalog version.
*/
- IgniteTable tableById(int id);
+ SchemaPlus schema(@Nullable String schemaName, int version);
/**
* Wait for {@code ver} schema version, just a stub, need to be removed after IGNITE-18733.
@@ -51,7 +41,9 @@ public interface SqlSchemaManager {
CompletableFuture<?> actualSchemaAsync(long ver);
/**
- * Returns a required schema if specified, or default schema otherwise.
+ * Returns a schema corresponding to the latest Catalog version, which is available on node locally.
+ * Note: For test purposes only. Method does NOT wait for actual Catalog version in the grid.
*/
- SchemaPlus activeSchema(@Nullable String name, long timestamp);
+ @TestOnly
+ SchemaPlus latestSchema(@Nullable String name);
}
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
deleted file mode 100644
index 1722d86074..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
+++ /dev/null
@@ -1,584 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.schema;
-
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.sql.engine.SqlQueryProcessor.DEFAULT_SCHEMA_NAME;
-import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
-import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
-import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
-import static org.apache.ignite.lang.IgniteStringFormatter.format;
-
-import it.unimi.dsi.fastutil.ints.IntArrayList;
-import it.unimi.dsi.fastutil.ints.IntList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.function.Consumer;
-import java.util.function.DoubleSupplier;
-import java.util.function.LongFunction;
-import java.util.stream.Collectors;
-import org.apache.calcite.schema.SchemaPlus;
-import org.apache.calcite.tools.Frameworks;
-import org.apache.ignite.internal.causality.CompletableVersionedValue;
-import org.apache.ignite.internal.causality.IncrementalVersionedValue;
-import org.apache.ignite.internal.causality.OutdatedTokenException;
-import org.apache.ignite.internal.index.HashIndex;
-import org.apache.ignite.internal.index.Index;
-import org.apache.ignite.internal.index.IndexDescriptor;
-import org.apache.ignite.internal.index.SortedIndexDescriptor;
-import org.apache.ignite.internal.index.SortedIndexImpl;
-import org.apache.ignite.internal.schema.Column;
-import org.apache.ignite.internal.schema.DefaultValueProvider;
-import org.apache.ignite.internal.schema.DefaultValueProvider.Type;
-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.sql.engine.trait.IgniteDistribution;
-import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.apache.ignite.internal.table.InternalTable;
-import org.apache.ignite.internal.table.TableImpl;
-import org.apache.ignite.internal.table.distributed.TableManager;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.lang.IgniteInternalException;
-import org.apache.ignite.lang.NodeStoppingException;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Holds actual schema and mutates it on schema change, requested by Ignite.
- */
-public class SqlSchemaManagerImpl implements SqlSchemaManager {
- private final IncrementalVersionedValue<Map<String, IgniteSchema>> schemasVv;
-
- private final IncrementalVersionedValue<Map<Integer, IgniteTable>> tablesVv;
-
- private final Map<Integer, CompletableFuture<?>> pkIdxReady = new ConcurrentHashMap<>();
-
- private final IncrementalVersionedValue<Map<Integer, IgniteIndex>> indicesVv;
-
- private final TableManager tableManager;
- private final SchemaManager schemaManager;
-
- private final CompletableVersionedValue<SchemaPlus> calciteSchemaVv;
-
- private final Set<SchemaUpdateListener> listeners = new CopyOnWriteArraySet<>();
-
- /** Busy lock for stop synchronisation. */
- private final IgniteSpinBusyLock busyLock;
-
- /**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public SqlSchemaManagerImpl(
- TableManager tableManager,
- SchemaManager schemaManager,
- Consumer<LongFunction<CompletableFuture<?>>> registry,
- IgniteSpinBusyLock busyLock
- ) {
- this.tableManager = tableManager;
- this.schemaManager = schemaManager;
-
- schemasVv = new IncrementalVersionedValue<>(registry, HashMap::new);
- tablesVv = new IncrementalVersionedValue<>(registry, HashMap::new);
- indicesVv = new IncrementalVersionedValue<>(registry, HashMap::new);
- this.busyLock = busyLock;
-
- calciteSchemaVv = new CompletableVersionedValue<>(() -> {
- SchemaPlus newCalciteSchema = Frameworks.createRootSchema(false);
- newCalciteSchema.add(DEFAULT_SCHEMA_NAME, new IgniteSchema(DEFAULT_SCHEMA_NAME));
- return newCalciteSchema;
- });
-
- schemasVv.whenComplete((token, stringIgniteSchemaMap, throwable) -> {
- if (!busyLock.enterBusy()) {
- calciteSchemaVv.completeExceptionally(token, new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
-
- return;
- }
- try {
- if (throwable != null) {
- calciteSchemaVv.completeExceptionally(
- token,
- new IgniteInternalException(
- INTERNAL_ERR, "Couldn't evaluate sql schemas for causality token: " + token, throwable)
- );
-
- return;
- }
-
- SchemaPlus newCalciteSchema = rebuild(stringIgniteSchemaMap);
-
- listeners.forEach(SchemaUpdateListener::onSchemaUpdated);
-
- calciteSchemaVv.complete(token, newCalciteSchema);
- } finally {
- busyLock.leaveBusy();
- }
- });
- }
-
- /** {@inheritDoc} */
- @Override
- public SchemaPlus schema(@Nullable String schema) {
- // stub for waiting pk indexes, more clear place is IgniteSchema
- CompletableFuture.allOf(pkIdxReady.values().toArray(CompletableFuture[]::new)).join();
-
- SchemaPlus schemaPlus = calciteSchemaVv.latest();
-
- return schema != null ? schemaPlus.getSubSchema(schema) : schemaPlus.getSubSchema(DEFAULT_SCHEMA_NAME);
- }
-
- /** {@inheritDoc} */
- @Override
- public SchemaPlus schema(String name, int version) {
- throw new UnsupportedOperationException();
- }
-
- /** {@inheritDoc} */
- @Override
- public CompletableFuture<?> actualSchemaAsync(long ver) {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
- try {
- if (ver == IgniteSchema.INITIAL_VERSION) {
- return completedFuture(calciteSchemaVv.latest());
- }
-
- CompletableFuture<SchemaPlus> lastSchemaFut;
-
- try {
- lastSchemaFut = calciteSchemaVv.get(ver);
- } catch (OutdatedTokenException e) {
- return completedFuture(null);
- }
-
- return lastSchemaFut;
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- /** {@inheritDoc} */
- @Override
- public SchemaPlus activeSchema(@Nullable String name, long timestamp) {
- throw new UnsupportedOperationException();
- }
-
- /** {@inheritDoc} */
- @Override
- public IgniteTable tableById(int id) {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
- try {
- IgniteTable table = tablesVv.latest().get(id);
-
- if (table == null) {
- throw new IgniteInternalException(INTERNAL_ERR,
- format("Table not found [tableId={}]", id));
- }
-
- return table;
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- public void registerListener(SchemaUpdateListener listener) {
- listeners.add(listener);
- }
-
- /**
- * OnSqlTypeCreated.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public CompletableFuture<?> onTableCreated(String schemaName, int tableId, long causalityToken) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
- }
-
- try {
- pkIdxReady.computeIfAbsent(tableId, k -> new CompletableFuture<>());
-
- CompletableFuture<Map<Integer, IgniteTable>> updatedTables = tablesVv.update(causalityToken, (tables, e) ->
- inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return tableManager.tableAsync(causalityToken, tableId)
- .thenCompose(table -> convert(causalityToken, table))
- .thenApply(igniteTable -> {
- Map<Integer, IgniteTable> resTbls = new HashMap<>(tables);
-
- IgniteTable oldTable = resTbls.put(igniteTable.id(), igniteTable);
-
- // looks like this is UPDATE operation
- if (oldTable != null) {
- for (var index : oldTable.indexes().values()) {
- igniteTable.addIndex(index);
- }
- }
-
- return resTbls;
- });
- })
- );
-
- schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return updatedTables.thenApply(tables -> {
- IgniteTable igniteTable = tables.get(tableId);
-
- Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
- IgniteSchema schema = res.compute(schemaName,
- (k, v) -> v == null ? new IgniteSchema(schemaName, causalityToken) : IgniteSchema.copy(v, causalityToken));
-
- schema.addTable(igniteTable);
-
- return res;
- });
- }));
-
- // calciteSchemaVv depends on all other Versioned Values and is completed last.
- return calciteSchemaVv.get(causalityToken);
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- /**
- * OnSqlTypeUpdated.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public CompletableFuture<?> onTableUpdated(String schemaName, int tableId, long causalityToken) {
- return onTableCreated(schemaName, tableId, causalityToken);
- }
-
- /**
- * OnSqlTypeDropped.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public CompletableFuture<?> onTableDropped(String schemaName, int tableId, long causalityToken) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
- }
-
- try {
- CompletableFuture<IgniteTable> removedTableFuture = new CompletableFuture<>();
-
- tablesVv.update(causalityToken, (tables, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- removedTableFuture.completeExceptionally(e);
-
- return failedFuture(e);
- }
-
- Map<Integer, IgniteTable> resTbls = new HashMap<>(tables);
-
- IgniteTable removedTable = resTbls.remove(tableId);
-
- removedTableFuture.complete(removedTable);
-
- return completedFuture(resTbls);
- }));
-
- schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return removedTableFuture.thenApply(table -> {
- if (table == null) {
- return schemas;
- }
-
- Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
- IgniteSchema schema = res.compute(schemaName,
- (k, v) -> v == null ? new IgniteSchema(schemaName, causalityToken) : IgniteSchema.copy(v, causalityToken));
-
- schema.removeTable(table.name());
-
- pkIdxReady.remove(table.id());
-
- return res;
- });
- }));
-
- // calciteSchemaVv depends on all other Versioned Values and is completed last.
- return calciteSchemaVv.get(causalityToken);
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- /**
- * Rebuilds Calcite schemas.
- *
- * @param schemas Ignite schemas.
- */
- private SchemaPlus rebuild(Map<String, IgniteSchema> schemas) {
- SchemaPlus newCalciteSchema = Frameworks.createRootSchema(false);
-
- newCalciteSchema.add(DEFAULT_SCHEMA_NAME, new IgniteSchema(DEFAULT_SCHEMA_NAME));
-
- schemas.forEach(newCalciteSchema::add);
-
- return newCalciteSchema;
- }
-
- private CompletableFuture<IgniteTableImpl> convert(long causalityToken, TableImpl table) {
- return schemaManager.schemaRegistry(causalityToken, table.tableId())
- .thenApply(schemaRegistry -> inBusyLock(busyLock, () -> convert(table, schemaRegistry, causalityToken)));
- }
-
- private IgniteTableImpl convert(TableImpl table, SchemaRegistry schemaRegistry, long schemaVersion) {
- SchemaDescriptor descriptor = schemaRegistry.schema();
-
- List<ColumnDescriptor> colDescriptors = descriptor.columnNames().stream()
- .map(descriptor::column)
- .sorted(Comparator.comparingInt(Column::columnOrder))
- .map(col -> new ColumnDescriptorImpl(
- col.name(),
- descriptor.isKeyColumn(col.schemaIndex()),
- col.nullable(),
- col.columnOrder(),
- col.schemaIndex(),
- col.type(),
- convertDefaultValueProvider(col.defaultValueProvider()),
- col::defaultValue
- ))
- .collect(Collectors.toList());
-
- IntList colocationColumns = new IntArrayList();
-
- for (Column column : descriptor.colocationColumns()) {
- colocationColumns.add(column.columnOrder());
- }
-
- // TODO Use the actual zone ID after implementing https://issues.apache.org/jira/browse/IGNITE-18426.
- IgniteDistribution distribution = IgniteDistributions.affinity(colocationColumns, table.tableId(), table.tableId());
-
- InternalTable internalTable = table.internalTable();
- DoubleSupplier rowCount = IgniteTableImpl.rowCountStatistic(internalTable);
-
- return new IgniteTableImpl(
- new TableDescriptorImpl(colDescriptors, distribution),
- internalTable.tableId(),
- internalTable.name(),
- schemaRegistry.lastSchemaVersion(),
- rowCount
- );
- }
-
- private DefaultValueStrategy convertDefaultValueProvider(DefaultValueProvider defaultValueProvider) {
- return defaultValueProvider.type() == Type.CONSTANT
- ? DefaultValueStrategy.DEFAULT_CONSTANT
- : DefaultValueStrategy.DEFAULT_COMPUTED;
- }
-
- /**
- * Index created callback method register index in Calcite schema.
- *
- * @param tableId Table ID.
- * @param indexId Index ID.
- * @param causalityToken Causality token.
- * @return Schema registration future.
- */
- public CompletableFuture<?> onIndexCreated(int tableId, int indexId, IndexDescriptor indexDescriptor, long causalityToken) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
- }
-
- try {
- CompletableFuture<Map<Integer, IgniteIndex>> updatedIndices = indicesVv.update(causalityToken, (indices, e) ->
- inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return tableManager.tableAsync(causalityToken, tableId).thenApply(table -> {
- var igniteIndex = new IgniteIndex(newIndex(table, indexId, indexDescriptor));
-
- Map<Integer, IgniteIndex> resIdxs = new HashMap<>(indices);
-
- resIdxs.put(indexId, igniteIndex);
-
- return resIdxs;
- });
- }));
-
- CompletableFuture<Map<Integer, IgniteTable>> updatedTables = tablesVv.update(causalityToken, (tables, e) ->
- inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return updatedIndices.thenApply(indices -> {
- IgniteIndex igniteIndex = indices.get(indexId);
-
- Map<Integer, IgniteTable> resTbls = new HashMap<>(tables);
-
- IgniteTable igniteTable = resTbls.computeIfPresent(tableId,
- (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
-
- assert igniteTable != null : "Table " + tableId + " was not found";
-
- igniteTable.addIndex(igniteIndex);
-
- return resTbls;
- });
- }));
-
- schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- return updatedTables.thenCombine(updatedIndices, (tables, indices) -> inBusyLock(busyLock, () -> {
- Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
- IgniteSchema schema = res.compute(DEFAULT_SCHEMA_NAME,
- (k, v) -> v == null ? new IgniteSchema(k, causalityToken) : IgniteSchema.copy(v, causalityToken));
-
- schema.addTable(tables.get(tableId));
-
- schema.addIndex(indexId, indices.get(indexId));
-
- return res;
- }));
- }));
-
- // this stub is necessary for observing pk index creation.
- schemasVv.whenComplete((token, stringIgniteSchemaMap, throwable) -> {
- CompletableFuture<?> pkFut = pkIdxReady.get(tableId);
- // this listener is called repeatedly on node stop.
- if (pkFut != null) {
- pkFut.complete(null);
- }
- });
-
- return calciteSchemaVv.get(causalityToken);
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- private static Index<?> newIndex(TableImpl table, int indexId, IndexDescriptor descriptor) {
- if (descriptor instanceof SortedIndexDescriptor) {
- return new SortedIndexImpl(indexId, table.internalTable(), (SortedIndexDescriptor) descriptor);
- } else {
- return new HashIndex(indexId, table.internalTable(), descriptor);
- }
- }
-
- /**
- * Index dropped callback method deregisters index from Calcite schema.
- *
- * @param schemaName Schema name.
- * @param indexId Index id.
- * @param causalityToken Causality token.
- * @return Schema registration future.
- */
- public CompletableFuture<?> onIndexDropped(String schemaName, int tableId, int indexId, long causalityToken) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()));
- }
-
- try {
- CompletableFuture<IgniteIndex> removedIndexFuture = new CompletableFuture<>();
-
- indicesVv.update(causalityToken, (indices, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- removedIndexFuture.completeExceptionally(e);
-
- return failedFuture(e);
- }
-
- Map<Integer, IgniteIndex> resIdxs = new HashMap<>(indices);
-
- IgniteIndex rmvIdx = resIdxs.remove(indexId);
-
- removedIndexFuture.complete(rmvIdx);
-
- return completedFuture(resIdxs);
- }));
-
- CompletableFuture<Map<Integer, IgniteTable>> updatedTables = tablesVv.update(causalityToken, (tables, e) ->
- inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- Map<Integer, IgniteTable> resTbls = new HashMap<>(tables);
-
- IgniteTable table = resTbls.computeIfPresent(tableId, (k, v) -> IgniteTableImpl.copyOf((IgniteTableImpl) v));
-
- if (table == null) {
- return completedFuture(resTbls);
- } else {
- return removedIndexFuture.thenApply(rmvIndex -> {
- table.removeIndex(rmvIndex.name());
-
- return resTbls;
- });
- }
- }));
-
- schemasVv.update(causalityToken, (schemas, e) -> inBusyLock(busyLock, () -> {
- if (e != null) {
- return failedFuture(e);
- }
-
- Map<String, IgniteSchema> res = new HashMap<>(schemas);
-
- IgniteSchema schema = res.compute(schemaName,
- (k, v) -> v == null ? new IgniteSchema(schemaName, causalityToken) : IgniteSchema.copy(v, causalityToken));
-
- schema.removeIndex(indexId);
-
- return updatedTables.thenApply(tables -> {
- IgniteTable table = tables.get(tableId);
-
- if (table != null) {
- schema.addTable(tables.get(tableId));
- }
-
- return res;
- });
- }));
-
- return calciteSchemaVv.get(causalityToken);
- } finally {
- busyLock.leaveBusy();
- }
- }
-}
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/main/java/org/apache/ignite/internal/sql/engine/util/HashFunctionFactoryImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/HashFunctionFactoryImpl.java
index 4ad5484257..6eac606773 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/HashFunctionFactoryImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/HashFunctionFactoryImpl.java
@@ -26,7 +26,8 @@ import org.apache.ignite.internal.schema.NativeType;
import org.apache.ignite.internal.schema.NativeTypeSpec;
import org.apache.ignite.internal.sql.engine.exec.RowHandler;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
-import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
+import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.util.ColocationUtils;
import org.apache.ignite.internal.util.HashCalculator;
@@ -35,11 +36,11 @@ import org.apache.ignite.internal.util.HashCalculator;
* Factory for creating a function to calculate the hash of the specified fields of a row.
*/
public class HashFunctionFactoryImpl<T> implements HashFunctionFactory<T> {
- private final SqlSchemaManager sqlSchemaManager;
private final RowHandler<T> rowHandler;
+ private final IgniteCatalogSchema schema;
- public HashFunctionFactoryImpl(SqlSchemaManager sqlSchemaManager, RowHandler<T> rowHandler) {
- this.sqlSchemaManager = sqlSchemaManager;
+ public HashFunctionFactoryImpl(IgniteCatalogSchema schema, RowHandler<T> rowHandler) {
+ this.schema = schema;
this.rowHandler = rowHandler;
}
@@ -48,7 +49,15 @@ public class HashFunctionFactoryImpl<T> implements HashFunctionFactory<T> {
public RowHashFunction<T> create(int[] fields, int tableId) {
int fieldCnt = fields.length;
NativeType[] fieldTypes = new NativeType[fieldCnt];
- TableDescriptor tblDesc = sqlSchemaManager.tableById(tableId).descriptor();
+ //TODO: optimize this
+ TableDescriptor tblDesc = schema.getTableNames().stream()
+ .map(schema::getTable)
+ .map(IgniteTable.class::cast)
+ .filter(t -> t.id() == tableId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No table found: tableId=" + tableId))
+ .descriptor();
+
ImmutableIntList colocationColumns = tblDesc.distribution().getKeys();
assert colocationColumns.size() == fieldCnt : "fieldsCount=" + fieldCnt + ", colocationColumns=" + colocationColumns;
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index ec61eca2fb..627c2e81a3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -89,6 +89,7 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptorImpl;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptorImpl;
@@ -590,10 +591,6 @@ public class ExecutionServiceImplTest {
when(topologyService.localMember()).thenReturn(clusterNode);
- when(schemaManagerMock.tableById(anyInt())).thenReturn(table);
-
- when(schemaManagerMock.actualSchemaAsync(isA(long.class))).thenReturn(CompletableFuture.completedFuture(null));
-
TestExecutableTableRegistry executableTableRegistry = new TestExecutableTableRegistry();
executableTableRegistry.setColocatioGroupProvider((tableId) -> {
// Make sure the exception is handled properly if it occurs during the mapping phase.
@@ -785,7 +782,7 @@ public class ExecutionServiceImplTest {
MailboxRegistry mailboxRegistry,
ExchangeService exchangeService,
ResolvedDependencies deps) {
- HashFunctionFactory<Object[]> funcFactory = new HashFunctionFactoryImpl<>(mock(SqlSchemaManager.class), ctx.rowHandler());
+ HashFunctionFactory<Object[]> funcFactory = new HashFunctionFactoryImpl<>(ctx.getRootSchema().unwrap(IgniteCatalogSchema.class), ctx.rowHandler());
return new LogicalRelImplementor<>(ctx, funcFactory, mailboxRegistry, exchangeService, deps) {
@Override
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
deleted file mode 100644
index eb63c18ebe..0000000000
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java
+++ /dev/null
@@ -1,329 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.exec.schema;
-
-import static java.util.concurrent.CompletableFuture.allOf;
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
-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.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Consumer;
-import java.util.function.LongFunction;
-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;
-import org.apache.ignite.internal.schema.Column;
-import org.apache.ignite.internal.schema.NativeTypes;
-import org.apache.ignite.internal.schema.SchemaDescriptor;
-import org.apache.ignite.internal.schema.SchemaManager;
-import org.apache.ignite.internal.schema.registry.SchemaRegistryImpl;
-import org.apache.ignite.internal.sql.engine.planner.AbstractPlannerTest.TestHashIndex;
-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.SqlSchemaManagerImpl;
-import org.apache.ignite.internal.table.InternalTable;
-import org.apache.ignite.internal.table.TableImpl;
-import org.apache.ignite.internal.table.distributed.TableManager;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-/**
- * Tests to verify {@link SqlSchemaManagerImpl}.
- */
-@ExtendWith(MockitoExtension.class)
-public class SqlSchemaManagerTest extends BaseIgniteAbstractTest {
- private final int tableId = 1;
-
- private final int indexId = 2;
-
- private final SchemaDescriptor schemaDescriptor = new SchemaDescriptor(
- 1,
- new Column[]{new Column(0, "ID", NativeTypes.INT64, false)},
- new Column[]{new Column(1, "VAL", NativeTypes.INT64, false)}
- );
-
- @Mock
- private TableManager tableManager;
-
- @Mock
- private SchemaManager schemaManager;
-
- @Mock
- private TableImpl table;
-
- @Mock
- private Index<IndexDescriptor> index;
-
- @Mock
- private SchemaRegistryImpl schemaRegistry;
-
- private SqlSchemaManagerImpl sqlSchemaManager;
-
- private TestRevisionRegister testRevisionRegister;
-
- /** Busy lock for stop synchronisation. */
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- @BeforeEach
- public void setup() {
- Mockito.reset(tableManager);
-
- when(tableManager.tableAsync(anyLong(), eq(tableId))).thenReturn(completedFuture(table));
-
- testRevisionRegister = new TestRevisionRegister();
-
- sqlSchemaManager = new SqlSchemaManagerImpl(
- tableManager,
- schemaManager,
- testRevisionRegister,
- busyLock
- );
-
- testRevisionRegister.moveForward();
- }
-
- @Test
- public void testOnTableDroppedHandler() {
- InternalTable mock = mock(InternalTable.class);
- when(mock.tableId()).thenReturn(tableId);
- when(mock.name()).thenReturn("T");
-
- when(table.internalTable()).thenReturn(mock);
- when(table.tableId()).thenReturn(tableId);
- when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
- when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
-
- when(schemaManager.schemaRegistry(anyLong(), anyInt())).thenReturn(completedFuture(schemaRegistry));
-
- sqlSchemaManager.onTableCreated("PUBLIC", tableId, testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- TestHashIndex testHashIndex = TestHashIndex.create(List.of("ID"), "pk_idx", tableId);
-
- sqlSchemaManager.onIndexCreated(
- testHashIndex.tableId(),
- testHashIndex.id(),
- testHashIndex.descriptor(),
- testRevisionRegister.actualToken() + 1
- );
-
- testRevisionRegister.moveForward();
-
- Table schemaTable = sqlSchemaManager.schema("PUBLIC").getTable("T");
-
- assertNotNull(schemaTable);
- IgniteTableImpl igniteTable = assertInstanceOf(IgniteTableImpl.class, schemaTable);
- assertEquals(tableId, igniteTable.id());
-
- sqlSchemaManager.onTableDropped("PUBLIC", tableId, testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- assertNull(sqlSchemaManager.schema("PUBLIC").getTable("T"));
- }
-
- @Test
- public void testIndexEventHandler() {
- InternalTable mock = mock(InternalTable.class);
- when(mock.tableId()).thenReturn(tableId);
- when(mock.name()).thenReturn("T");
-
- when(table.internalTable()).thenReturn(mock);
- when(table.tableId()).thenReturn(tableId);
- when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
- when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- when(schemaManager.schemaRegistry(anyLong(), anyInt())).thenReturn(completedFuture(schemaRegistry));
-
- sqlSchemaManager.onTableCreated("PUBLIC", tableId, testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- TestHashIndex testHashIndex = TestHashIndex.create(List.of("ID"), "pk_idx", tableId);
-
- sqlSchemaManager.onIndexCreated(
- testHashIndex.tableId(),
- testHashIndex.id(),
- testHashIndex.descriptor(),
- testRevisionRegister.actualToken() + 1
- );
-
- testRevisionRegister.moveForward();
-
- assertEquals(1, ((IgniteTableImpl) sqlSchemaManager.schema("PUBLIC").getTable("T")).indexes().size());
-
- IndexDescriptor descMock = mock(IndexDescriptor.class);
- when(descMock.columns()).thenReturn(List.of());
- when(descMock.name()).thenReturn("PUBLIC.I");
-
- sqlSchemaManager.onIndexCreated(tableId, indexId, descMock, testRevisionRegister.actualToken() + 1);
-
- testRevisionRegister.moveForward();
-
- IgniteSchema schema = sqlSchemaManager.schema("PUBLIC").unwrap(IgniteSchema.class);
- Table schemaTable = schema.getTable("T");
- IgniteIndex igniteIndex = schema.index(indexId);
-
- assertNotNull(igniteIndex);
-
- IgniteTableImpl igniteTable = assertInstanceOf(IgniteTableImpl.class, schemaTable);
-
- assertEquals(igniteTable.id(), igniteIndex.tableId());
- assertSame(igniteIndex, igniteTable.indexes().get("PUBLIC.I"));
-
- sqlSchemaManager.onIndexDropped("PUBLIC", igniteTable.id(), indexId, testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- assertNull(sqlSchemaManager.schema("PUBLIC").unwrap(IgniteSchema.class).index(indexId));
-
- verifyNoMoreInteractions(tableManager);
- }
-
-
- @Test
- public void testIndexEventsProcessed() {
- InternalTable mock = mock(InternalTable.class);
- when(mock.tableId()).thenReturn(tableId);
- when(mock.name()).thenReturn("T");
-
- when(table.internalTable()).thenReturn(mock);
- when(table.tableId()).thenReturn(tableId);
- when(schemaRegistry.schema()).thenReturn(schemaDescriptor);
- when(schemaRegistry.lastSchemaVersion()).thenReturn(schemaDescriptor.version());
- when(schemaManager.schemaRegistry(anyLong(), anyInt())).thenReturn(completedFuture(schemaRegistry));
-
- sqlSchemaManager.onTableCreated("PUBLIC", table.tableId(), testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- TestHashIndex testHashIndex = TestHashIndex.create(List.of("ID"), "pk_idx", tableId);
-
- sqlSchemaManager.onIndexCreated(
- testHashIndex.tableId(),
- testHashIndex.id(),
- testHashIndex.descriptor(),
- testRevisionRegister.actualToken() + 1
- );
-
- testRevisionRegister.moveForward();
-
- String idxName = "I";
-
- IndexDescriptor descMock = mock(IndexDescriptor.class);
- when(descMock.columns()).thenReturn(List.of());
- when(descMock.name()).thenReturn(idxName);
-
- {
- SchemaPlus schema1 = sqlSchemaManager.schema("PUBLIC");
-
- sqlSchemaManager.onIndexCreated(tableId, indexId, descMock, testRevisionRegister.actualToken() + 1);
- testRevisionRegister.moveForward();
-
- SchemaPlus schema2 = sqlSchemaManager.schema("PUBLIC");
-
- // 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(((IgniteTable) schema1.getTable("T")).getIndex(idxName));
- assertNotNull(((IgniteTable) schema2.getTable("T")).getIndex(idxName));
- }
- {
- sqlSchemaManager.onIndexDropped("PUBLIC", table.tableId(), indexId, testRevisionRegister.actualToken() + 1);
- SchemaPlus schema1 = sqlSchemaManager.schema("PUBLIC");
- testRevisionRegister.moveForward();
-
- SchemaPlus schema2 = sqlSchemaManager.schema("PUBLIC");
-
- // 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));
-
- assertNull(((IgniteTable) schema2.getTable("T")).getIndex(idxName));
- assertNotNull(((IgniteTable) schema1.getTable("T")).getIndex(idxName));
- }
-
- verifyNoMoreInteractions(tableManager);
- }
-
- /**
- * Test revision register.
- */
- private static class TestRevisionRegister implements Consumer<LongFunction<CompletableFuture<?>>> {
- AtomicLong token = new AtomicLong(-1);
-
- /** Revision consumer. */
- private LongFunction<CompletableFuture<?>> moveRevision;
-
- /**
- * Moves forward token.
- */
- void moveForward() {
- await(moveRevision.apply(token.incrementAndGet()));
- }
-
- /**
- * Gets an actual token.
- *
- * @return Actual token.
- */
- long actualToken() {
- return token.get();
- }
-
- /** {@inheritDoc} */
- @Override
- public void accept(LongFunction<CompletableFuture<?>> function) {
- if (moveRevision == null) {
- moveRevision = function;
- } else {
- LongFunction<CompletableFuture<?>> old = moveRevision;
-
- moveRevision = rev -> allOf(
- old.apply(rev),
- function.apply(rev)
- );
- }
- }
- }
-}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
index 73ddc9098b..582087733d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.sql.engine.framework;
+import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_SCHEMA_NAME;
+
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -66,12 +68,6 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
}
}
- /** {@inheritDoc} */
- @Override
- public SchemaPlus schema(@Nullable String schema) {
- return schema == null ? root : root.getSubSchema(schema);
- }
-
/** {@inheritDoc} */
@Override
public SchemaPlus schema(String name, int version) {
@@ -86,13 +82,11 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus activeSchema(@Nullable String name, long timestamp) {
+ public SchemaPlus latestSchema(@Nullable String name) {
return schema(name);
}
- /** {@inheritDoc} */
- @Override
- public IgniteTable tableById(int id) {
- return tableById.get(id);
+ private SchemaPlus schema(@Nullable String schemaName) {
+ return root.getSubSchema(schemaName == null ? DEFAULT_SCHEMA_NAME : schemaName);
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
index 88f270c581..031f7a8ac3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
@@ -27,7 +27,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
-import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.internal.sql.engine.AsyncCursor;
import org.apache.ignite.internal.sql.engine.QueryCancel;
@@ -59,6 +58,7 @@ import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
import org.apache.ignite.internal.sql.engine.rel.IgniteIndexScan;
import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserService;
@@ -131,7 +131,7 @@ public class TestNode implements LifecycleAware {
dependencyResolver,
(ctx, deps) -> new LogicalRelImplementor<Object[]>(
ctx,
- new HashFunctionFactoryImpl<>(schemaManager, rowHandler),
+ new HashFunctionFactoryImpl<>(ctx.getRootSchema().unwrap(IgniteCatalogSchema.class), rowHandler),
mailboxRegistry,
exchangeService,
deps
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
index 18b64e3644..0fa0228799 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
@@ -99,7 +99,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
assertEquals(testSchema.name, schema.getName());
@@ -156,7 +156,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteSchemaTable table = getTable(schema, testTable);
@@ -197,7 +197,7 @@ public class CatalogSqlSchemaManagerTest {
SqlSchemaManager sqlSchemaManager = newSchemaManager();
{
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
assertEquals(DEFAULT_SCHEMA_NAME, schema.getName());
@@ -205,7 +205,7 @@ public class CatalogSqlSchemaManagerTest {
}
{
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(null, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(null);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
assertEquals(DEFAULT_SCHEMA_NAME, schema.getName());
@@ -234,7 +234,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteTable table = getTable(schema, testTable);
@@ -274,7 +274,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
@@ -302,7 +302,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
@@ -336,7 +336,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
@@ -369,7 +369,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteSchemaTable table = (IgniteSchemaTable) schema.getTable(testTable.name);
@@ -413,7 +413,7 @@ public class CatalogSqlSchemaManagerTest {
testSchema.init(catalogManager);
SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.activeSchema(testSchema.name, testSchema.timestamp);
+ SchemaPlus schemaPlus = sqlSchemaManager.latestSchema(testSchema.name);
IgniteCatalogSchema schema = unwrapSchema(schemaPlus);
IgniteSchemaTable table = (IgniteSchemaTable) schema.getTable(testTable.name);
@@ -469,7 +469,7 @@ public class CatalogSqlSchemaManagerTest {
void init(CatalogManager catalogManager) {
CatalogSchemaDescriptor schemaDescriptor = newSchemaDescriptor(version);
- when(catalogManager.activeCatalogVersion(timestamp)).thenReturn(version);
+ when(catalogManager.activeCatalogVersion(Long.MAX_VALUE)).thenReturn(version);
when(catalogManager.schema(name != null ? name : DEFAULT_SCHEMA_NAME, version)).thenReturn(schemaDescriptor);
}
[ignite-3] 05/06: Replace IgniteIndex with IgniteSchemaIndex
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 8276700f76d01321bb297daa70c855d90c71a54d
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 22:18:28 2023 +0300
Replace IgniteIndex with IgniteSchemaIndex
---
.../sql/engine/exec/LogicalRelImplementor.java | 8 +++---
.../internal/sql/engine/exec/RowConverter.java | 2 +-
.../internal/sql/engine/exec/ScannableTable.java | 4 +--
.../sql/engine/exec/ScannableTableImpl.java | 5 ++--
.../sql/engine/exec/rel/IndexScanNode.java | 8 +++---
.../engine/rel/logical/IgniteLogicalIndexScan.java | 6 ++--
.../sql/engine/rule/LogicalScanConverterRule.java | 7 ++---
.../sql/engine/rule/logical/ExposeIndexRule.java | 4 +--
.../sql/engine/schema/CatalogSqlSchemaManager.java | 33 ++++++++++++++--------
.../sql/engine/schema/IgniteSchemaIndex.java | 27 +++++++++++++++++-
.../exec/rel/IndexScanNodeExecutionTest.java | 15 ++++++++--
.../engine/exec/rel/ScannableTableSelfTest.java | 17 ++++-------
.../internal/sql/engine/framework/TestIndex.java | 1 +
13 files changed, 87 insertions(+), 50 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
index 464a0c0d5a..0b4e3e7cd0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
@@ -103,8 +103,8 @@ import org.apache.ignite.internal.sql.engine.rel.agg.IgniteReduceHashAggregate;
import org.apache.ignite.internal.sql.engine.rel.agg.IgniteReduceSortAggregate;
import org.apache.ignite.internal.sql.engine.rel.set.IgniteSetOp;
import org.apache.ignite.internal.sql.engine.rule.LogicalScanConverterRule;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.trait.Destination;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
@@ -329,7 +329,7 @@ public class LogicalRelImplementor<RowT> implements IgniteRelVisitor<Node<RowT>>
RelDataType rowType = tbl.getRowType(typeFactory, requiredColumns);
ScannableTable scannableTable = resolvedDependencies.scannableTable(tbl.id());
- IgniteIndex idx = tbl.getIndex(rel.indexName());
+ IgniteSchemaIndex idx = tbl.indexes().get(rel.indexName());
List<SearchBounds> searchBounds = rel.searchBounds();
RexNode condition = rel.condition();
@@ -343,8 +343,8 @@ public class LogicalRelImplementor<RowT> implements IgniteRelVisitor<Node<RowT>>
if (searchBounds != null) {
Comparator<RowT> searchRowComparator = null;
- if (idx.collations() != null) {
- searchRowComparator = expressionFactory.comparator(TraitUtils.createCollation(idx.collations()));
+ if (idx.type() != Type.SORTED) {
+ searchRowComparator = expressionFactory.comparator(idx.collation());
}
ranges = expressionFactory.ranges(searchBounds, idx.getRowType(typeFactory, tbl.descriptor()), searchRowComparator);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowConverter.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowConverter.java
index 3abd4cd5c9..ff6df49cdd 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowConverter.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowConverter.java
@@ -40,7 +40,7 @@ public final class RowConverter {
/**
* Creates binary tuple schema for index rows.
*/
- public static BinaryTupleSchema createIndexRowSchema(List<String> indexedColumns, TableDescriptor tableDescriptor) {
+ public static BinaryTupleSchema createIndexRowSchema(List<Integer> indexedColumns, TableDescriptor tableDescriptor) {
Element[] elements = indexedColumns.stream()
.map(tableDescriptor::columnDescriptor)
.map(colDesc -> new Element(colDesc.physicalType(), true))
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTable.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTable.java
index 1ea160ca8e..7d52acc2e9 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTable.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTable.java
@@ -57,7 +57,7 @@ public interface ScannableTable {
* @return A publisher that produces rows.
*/
<RowT> Publisher<RowT> indexRangeScan(ExecutionContext<RowT> ctx, PartitionWithTerm partWithTerm,
- RowFactory<RowT> rowFactory, int indexId, List<String> columns,
+ RowFactory<RowT> rowFactory, int indexId, List<Integer> columns,
@Nullable RangeCondition<RowT> cond, @Nullable BitSet requiredColumns);
/**
@@ -74,6 +74,6 @@ public interface ScannableTable {
* @return A publisher that produces rows.
*/
<RowT> Publisher<RowT> indexLookup(ExecutionContext<RowT> ctx, PartitionWithTerm partWithTerm,
- RowFactory<RowT> rowFactory, int indexId, List<String> columns,
+ RowFactory<RowT> rowFactory, int indexId, List<Integer> columns,
RowT key, @Nullable BitSet requiredColumns);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
index 147086a52b..b20e934673 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
@@ -83,7 +83,7 @@ public class ScannableTableImpl implements ScannableTable {
PartitionWithTerm partWithTerm,
RowFactory<RowT> rowFactory,
int indexId,
- List<String> columns,
+ List<Integer> columns,
@Nullable RangeCondition<RowT> cond,
@Nullable BitSet requiredColumns
) {
@@ -143,7 +143,8 @@ public class ScannableTableImpl implements ScannableTable {
PartitionWithTerm partWithTerm,
RowFactory<RowT> rowFactory,
int indexId,
- List<String> columns, RowT key,
+ List<Integer> columns,
+ RowT key,
@Nullable BitSet requiredColumns
) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNode.java
index 55ad834cdf..52647241eb 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNode.java
@@ -31,7 +31,7 @@ import org.apache.ignite.internal.sql.engine.exec.ScannableTable;
import org.apache.ignite.internal.sql.engine.exec.exp.RangeCondition;
import org.apache.ignite.internal.sql.engine.exec.exp.RangeIterable;
import org.apache.ignite.internal.sql.engine.metadata.PartitionWithTerm;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.util.SubscriptionUtils;
@@ -43,7 +43,7 @@ import org.jetbrains.annotations.Nullable;
*/
public class IndexScanNode<RowT> extends StorageScanNode<RowT> {
/** Schema index. */
- private final IgniteIndex schemaIndex;
+ private final IgniteSchemaIndex schemaIndex;
private final ScannableTable table;
@@ -75,7 +75,7 @@ public class IndexScanNode<RowT> extends StorageScanNode<RowT> {
public IndexScanNode(
ExecutionContext<RowT> ctx,
RowHandler.RowFactory<RowT> rowFactory,
- IgniteIndex schemaIndex,
+ IgniteSchemaIndex schemaIndex,
ScannableTable table,
TableDescriptor tableDescriptor,
Collection<PartitionWithTerm> partsWithTerms,
@@ -124,7 +124,7 @@ public class IndexScanNode<RowT> extends StorageScanNode<RowT> {
private Publisher<RowT> partitionPublisher(PartitionWithTerm partWithTerm, @Nullable RangeCondition<RowT> cond) {
int indexId = schemaIndex.id();
- List<String> columns = schemaIndex.columns();
+ List<Integer> columns = schemaIndex.collation().getKeys();
ExecutionContext<RowT> ctx = context();
switch (schemaIndex.type()) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
index d76a263c68..25ba268f2a 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
@@ -28,9 +28,9 @@ import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.internal.sql.engine.prepare.bounds.SearchBounds;
import org.apache.ignite.internal.sql.engine.rel.AbstractIndexScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
-import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.RexUtils;
@@ -53,8 +53,8 @@ public class IgniteLogicalIndexScan extends AbstractIndexScan {
) {
IgniteTable tbl = table.unwrap(IgniteTable.class);
IgniteTypeFactory typeFactory = Commons.typeFactory(cluster);
- IgniteIndex index = tbl.getIndex(idxName);
- RelCollation collation = TraitUtils.createCollation(index.columns(), index.collations(), tbl.descriptor());
+ IgniteSchemaIndex index = tbl.indexes().get(idxName);
+ RelCollation collation = index.collation();
List<SearchBounds> searchBounds;
if (index.type() == Type.HASH) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/LogicalScanConverterRule.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/LogicalScanConverterRule.java
index 836cfe19de..42632f6469 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/LogicalScanConverterRule.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/LogicalScanConverterRule.java
@@ -39,10 +39,9 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.rel.ProjectableFilterableTableScan;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalIndexScan;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
-import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
/**
* LogicalScanConverterRule.
@@ -62,10 +61,10 @@ public abstract class LogicalScanConverterRule<T extends ProjectableFilterableTa
) {
RelOptCluster cluster = rel.getCluster();
IgniteTable table = rel.getTable().unwrap(IgniteTable.class);
- IgniteIndex index = table.getIndex(rel.indexName());
+ IgniteSchemaIndex index = table.indexes().get(rel.indexName());
RelDistribution distribution = table.distribution();
- RelCollation collation = TraitUtils.createCollation(index.columns(), index.collations(), table.descriptor());
+ RelCollation collation = index.collation();
RelCollation outputCollation = index.type() == Type.HASH ? RelCollations.EMPTY : collation;
if (rel.projects() != null || rel.requiredColumns() != null) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/ExposeIndexRule.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/ExposeIndexRule.java
index ffd52389dd..2eb00d359b 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/ExposeIndexRule.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/ExposeIndexRule.java
@@ -32,8 +32,8 @@ import org.apache.calcite.util.ImmutableBitSet;
import org.apache.ignite.internal.sql.engine.prepare.bounds.SearchBounds;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalIndexScan;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.immutables.value.Value;
@@ -85,7 +85,7 @@ public class ExposeIndexRule extends RelRule<ExposeIndexRule.Config> {
/** Filter pre known not applicable variants. Significant shrink search space in some cases. */
private static boolean filter(IgniteTable table, String idxName, List<SearchBounds> searchBounds) {
- IgniteIndex index = table.getIndex(idxName);
+ IgniteSchemaIndex index = table.indexes().get(idxName);
return index.type() == Type.SORTED || (searchBounds != null
&& searchBounds.stream().noneMatch(bound -> bound.type() == SearchBounds.Type.RANGE));
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 432ad5834c..861018ba3b 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
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.schema.DefaultValueGenerator;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -116,21 +117,11 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
TableDescriptorImpl tableDescriptorImpl = tableDescriptorMap.get(tableId);
assert tableDescriptorImpl != null : "Table is not found in schema: " + tableId;
- String indexName = indexDescriptor.name();
Map<String, IgniteSchemaIndex> tableIndexes = schemaTableIndexes.computeIfAbsent(tableId, id -> new LinkedHashMap<>());
- Type type;
- if (indexDescriptor instanceof CatalogSortedIndexDescriptor) {
- type = Type.SORTED;
- } else if (indexDescriptor instanceof CatalogHashIndexDescriptor) {
- type = Type.HASH;
- } else {
- throw new IllegalArgumentException("Unexpected index type: " + indexDescriptor);
- }
+ IgniteSchemaIndex schemaIndex = createSchemaIndex(indexDescriptor, tableDescriptorImpl);
- RelCollation indexCollation = IgniteSchemaIndex.createIndexCollation(indexDescriptor, tableDescriptorImpl);
- IgniteSchemaIndex schemaIndex = new IgniteSchemaIndex(indexName, type, tableDescriptorImpl.distribution(), indexCollation);
- tableIndexes.put(indexName, schemaIndex);
+ tableIndexes.put(schemaIndex.name(), schemaIndex);
schemaTableIndexes.put(tableId, tableIndexes);
}
@@ -156,6 +147,24 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
return rootSchema.add(schemaName, igniteSchema);
}
+ private static IgniteSchemaIndex createSchemaIndex(CatalogIndexDescriptor indexDescriptor, TableDescriptorImpl tableDescriptorImpl) {
+ Type type;
+ if (indexDescriptor instanceof CatalogSortedIndexDescriptor) {
+ type = IgniteIndex.Type.SORTED;
+ } else if (indexDescriptor instanceof CatalogHashIndexDescriptor) {
+ type = IgniteIndex.Type.HASH;
+ } else {
+ throw new IllegalArgumentException("Unexpected index type: " + indexDescriptor);
+ }
+
+ int indexId = indexDescriptor.id();
+ String indexName = indexDescriptor.name();
+ IgniteDistribution distribution = tableDescriptorImpl.distribution();
+ RelCollation indexCollation = IgniteSchemaIndex.createIndexCollation(indexDescriptor, tableDescriptorImpl);
+
+ return new IgniteSchemaIndex(indexId, indexName, type, distribution, indexCollation);
+ }
+
private static TableDescriptorImpl createTableDescriptor(CatalogTableDescriptor descriptor) {
List<ColumnDescriptor> colDescriptors = new ArrayList<>();
List<Integer> colocationColumns = new ArrayList<>();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
index 5a5171bca4..09870c0fdc 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.sql.engine.schema;
+import static org.apache.ignite.internal.sql.engine.util.TypeUtils.native2relationalType;
+
import java.util.ArrayList;
import java.util.List;
import org.apache.calcite.plan.Convention;
@@ -28,6 +30,8 @@ import org.apache.calcite.rel.RelCollations;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelFieldCollation.Direction;
import org.apache.calcite.rel.RelFieldCollation.NullDirection;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
@@ -37,12 +41,15 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogSortedIndexDescript
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalIndexScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
+import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
/**
* Auxiliary data structure to represent a table index.
*/
public class IgniteSchemaIndex {
+ private final int id;
+
private final String name;
private final IgniteDistribution tableDistribution;
@@ -52,13 +59,19 @@ public class IgniteSchemaIndex {
private final Type type;
/** Constructor. */
- public IgniteSchemaIndex(String name, Type type, IgniteDistribution tableDistribution, RelCollation collation) {
+ public IgniteSchemaIndex(int indexId, String name, Type type, IgniteDistribution tableDistribution, RelCollation collation) {
+ this.id = indexId;
this.name = name;
this.type = type;
this.tableDistribution = tableDistribution;
this.collation = collation;
}
+ /** Returns an id of the index. */
+ public int id() {
+ return id;
+ }
+
/** Returns the name of this index. */
public String name() {
return name;
@@ -140,4 +153,16 @@ public class IgniteSchemaIndex {
throw new IllegalArgumentException("Unexpected index type: " + descriptor);
}
}
+
+ //TODO: cache rowType as it can't be changed.
+ public RelDataType getRowType(IgniteTypeFactory typeFactory, TableDescriptor tableDescriptor) {
+ RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(typeFactory);
+
+ for (RelFieldCollation fieldCollation : collation.getFieldCollations()) {
+ ColumnDescriptor colDesc = tableDescriptor.columnDescriptor(fieldCollation.getFieldIndex());
+ b.add(colDesc.name(), native2relationalType(typeFactory, colDesc.physicalType(), colDesc.nullable()));
+ }
+
+ return b.build();
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNodeExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNodeExecutionTest.java
index 1b6fb59cde..86525e3aaf 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNodeExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexScanNodeExecutionTest.java
@@ -32,6 +32,7 @@ import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.stream.Collectors;
+import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeFactory.Builder;
@@ -49,8 +50,11 @@ import org.apache.ignite.internal.sql.engine.exec.exp.RangeCondition;
import org.apache.ignite.internal.sql.engine.metadata.PartitionWithTerm;
import org.apache.ignite.internal.sql.engine.planner.AbstractPlannerTest.TestTableDescriptor;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
+import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
+import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.table.InternalTable;
import org.jetbrains.annotations.Nullable;
@@ -177,7 +181,12 @@ public class IndexScanNodeExecutionTest extends AbstractExecutionTest {
TableDescriptor tableDescriptor = new TestTableDescriptor(IgniteDistributions::single, rowType);
- IgniteIndex schemaIndex = new IgniteIndex(index);
+ //TODO: avoid IgniteIndex creation.
+ IgniteIndex idx = new IgniteIndex(index);
+ RelCollation collations = TraitUtils.createCollation(idx.columns(), idx.collations(), tableDescriptor);
+ IgniteDistribution distribution = tableDescriptor.distribution();
+
+ IgniteSchemaIndex schemaIndex = new IgniteSchemaIndex(index.id(), index.name(), idx.type(), distribution, collations);
RowFactory<Object[]> rowFactory = ctx.rowHandler().factory(ctx.getTypeFactory(), rowType);
SingleRangeIterable<Object[]> conditions = new SingleRangeIterable<>(new Object[]{}, null, false, false);
List<PartitionWithTerm> partitions = scannableTable.getPartitions();
@@ -212,7 +221,7 @@ public class IndexScanNodeExecutionTest extends AbstractExecutionTest {
/** {@inheritDoc} */
@Override
public <RowT> Publisher<RowT> indexRangeScan(ExecutionContext<RowT> ctx, PartitionWithTerm partWithTerm,
- RowFactory<RowT> rowFactory, int indexId, List<String> columns,
+ RowFactory<RowT> rowFactory, int indexId, List<Integer> columns,
@Nullable RangeCondition<RowT> cond, @Nullable BitSet requiredColumns) {
List<T> list = partitionedData.get(partWithTerm.partId());
@@ -221,7 +230,7 @@ public class IndexScanNodeExecutionTest extends AbstractExecutionTest {
@Override
public <RowT> Publisher<RowT> indexLookup(ExecutionContext<RowT> ctx, PartitionWithTerm partWithTerm,
- RowFactory<RowT> rowFactory, int indexId, List<String> columns,
+ RowFactory<RowT> rowFactory, int indexId, List<Integer> columns,
RowT key, @Nullable BitSet requiredColumns) {
return newPublisher(ctx, partWithTerm, rowFactory);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ScannableTableSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ScannableTableSelfTest.java
index 0e0dfd5d9d..fd91cfabbf 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ScannableTableSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ScannableTableSelfTest.java
@@ -48,11 +48,11 @@ import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory.Builder;
-import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.index.SortedIndex;
@@ -641,7 +641,7 @@ public class ScannableTableSelfTest {
RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
RowFactory<Object[]> rowFactory = rowHandler.factory(TYPE_FACTORY, input.rowType);
RangeCondition<Object[]> rangeCondition = condition.asRangeCondition();
- List<String> indexColumns = input.getIndexColumns();
+ List<Integer> indexColumns = input.getIndexColumns();
Publisher<Object[]> publisher = scannableTable.indexRangeScan(ctx, new PartitionWithTerm(partitionId, term), rowFactory,
indexId, indexColumns, rangeCondition, requiredFields);
@@ -675,7 +675,7 @@ public class ScannableTableSelfTest {
RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
RowFactory<Object[]> rowFactory = rowHandler.factory(TYPE_FACTORY, input.rowType);
- List<String> indexColumns = input.getIndexColumns();
+ List<Integer> indexColumns = input.getIndexColumns();
Publisher<Object[]> publisher = scannableTable.indexLookup(ctx, new PartitionWithTerm(partitionId, term), rowFactory,
indexId, indexColumns, key, requiredFields);
@@ -743,15 +743,8 @@ public class ScannableTableSelfTest {
publisher.closeExceptionally(t);
}
- private List<String> getIndexColumns() {
- List<String> columns = new ArrayList<>();
-
- indexColumns.stream().forEach(i -> {
- RelDataTypeField field = rowType.getFieldList().get(i);
- columns.add(field.getName());
- });
-
- return columns;
+ private List<Integer> getIndexColumns() {
+ return indexColumns.stream().boxed().collect(Collectors.toList());
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
index f9da45677b..dd8449e6bb 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
@@ -66,6 +66,7 @@ public class TestIndex extends IgniteIndex {
}
/** Returns an id of the index. */
+ @Override
public int id() {
return id;
}
[ignite-3] 01/06: Fix incorrect IgniteSchemaTable.unwrap.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 8a00476c660e8a7f330133aab78a9e54bb1bd8e9
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 18:53:18 2023 +0300
Fix incorrect IgniteSchemaTable.unwrap.
---
.../apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
index b27768ddca..74cee2f891 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
@@ -187,8 +187,8 @@ public final class IgniteSchemaTable extends AbstractTable implements IgniteTabl
/** {@inheritDoc} */
@Override
public <C> @Nullable C unwrap(Class<C> cls) {
- if (cls.isInstance(this)) {
- return cls.cast(this);
+ if (cls.isInstance(desc)) {
+ return cls.cast(desc);
}
return super.unwrap(cls);
[ignite-3] 04/06: Replace IgniteTableImpl with IgniteCatalogImpl.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 82b1252d3de270a5f10989b598127e1aa753042f
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 21:21:29 2023 +0300
Replace IgniteTableImpl with IgniteCatalogImpl.
---
.../engine/rule/logical/LogicalOrToUnionRule.java | 7 +-
.../sql/engine/schema/IgniteSchemaTable.java | 15 +-
.../internal/sql/engine/schema/IgniteTable.java | 2 +-
.../sql/engine/schema/IgniteTableImpl.java | 276 ---------------------
.../sql/engine/exec/ExecutionServiceImplTest.java | 4 +-
.../engine/framework/PredefinedSchemaManager.java | 7 +-
.../sql/engine/framework/TestBuilders.java | 10 +-
.../internal/sql/engine/framework/TestTable.java | 5 +-
.../sql/engine/planner/AbstractPlannerTest.java | 5 +-
.../engine/schema/CatalogSqlSchemaManagerTest.java | 4 +-
10 files changed, 26 insertions(+), 309 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
index 2c58e626fc..e7f1463ffe 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
@@ -26,6 +26,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.RelRule;
import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rex.RexBuilder;
@@ -38,6 +39,7 @@ import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
@@ -138,9 +140,8 @@ public class LogicalOrToUnionRule extends RelRule<LogicalOrToUnionRule.Config> {
BitSet idxsFirstFields = new BitSet(fieldCnt);
- for (IgniteIndex idx : tbl.indexes().values()) {
- List<RelFieldCollation> fieldCollations = TraitUtils.createCollation(idx.columns(), idx.collations(), tbl.descriptor())
- .getFieldCollations();
+ for (IgniteSchemaIndex idx : tbl.indexes().values()) {
+ List<RelFieldCollation> fieldCollations = idx.collation().getFieldCollations();
if (!CollectionUtils.nullOrEmpty(fieldCollations)) {
idxsFirstFields.set(fieldCollations.get(0).getFieldIndex());
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
index 74cee2f891..9ef7ac6a9a 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaTable.java
@@ -58,7 +58,7 @@ public final class IgniteSchemaTable extends AbstractTable implements IgniteTabl
private final Map<String, IgniteSchemaIndex> indexMap;
/** Constructor. */
- public IgniteSchemaTable(String name, int tableId, int version, TableDescriptor desc,
+ public IgniteSchemaTable(String name, int tableId, int version, TableDescriptor desc,
IgniteStatistic statistic, Map<String, IgniteSchemaIndex> indexMap) {
this.id = tableId;
@@ -186,7 +186,7 @@ public final class IgniteSchemaTable extends AbstractTable implements IgniteTabl
/** {@inheritDoc} */
@Override
- public <C> @Nullable C unwrap(Class<C> cls) {
+ public <C> @Nullable C unwrap(Class<C> cls) {
if (cls.isInstance(desc)) {
return cls.cast(desc);
}
@@ -196,8 +196,8 @@ public final class IgniteSchemaTable extends AbstractTable implements IgniteTabl
/** {@inheritDoc} */
@Override
- public Map<String, IgniteIndex> indexes() {
- throw new UnsupportedOperationException("getIndexes() should be used instead");
+ public Map<String, IgniteSchemaIndex> indexes() {
+ return indexMap;
}
/** {@inheritDoc} */
@@ -217,9 +217,4 @@ public final class IgniteSchemaTable extends AbstractTable implements IgniteTabl
public void removeIndex(String idxName) {
throw new UnsupportedOperationException("IndexMap is not modifiable");
}
-
- /** Returns a map of indexes defined for this table. */
- public Map<String, IgniteSchemaIndex> getIndexes() {
- return indexMap;
- }
-}
+}
\ No newline at end of file
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTable.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTable.java
index 1faaed98af..166490ba8b 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTable.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTable.java
@@ -192,7 +192,7 @@ public interface IgniteTable extends TranslatableTable, Wrapper {
*
* @return Indexes for the current table.
*/
- Map<String, IgniteIndex> indexes();
+ Map<String, IgniteSchemaIndex> indexes();
/**
* Adds index to table.
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
deleted file mode 100644
index 3b4003d442..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
+++ /dev/null
@@ -1,276 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.schema;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.DoubleSupplier;
-import org.apache.calcite.plan.Convention;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptTable;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollation;
-import org.apache.calcite.rel.RelCollations;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeFactory;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.schema.Statistic;
-import org.apache.calcite.schema.impl.AbstractTable;
-import org.apache.calcite.util.ImmutableBitSet;
-import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
-import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalIndexScan;
-import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
-import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
-import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
-import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
-import org.apache.ignite.internal.storage.MvPartitionStorage;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
-import org.apache.ignite.internal.table.InternalTable;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Ignite table implementation.
- */
-public class IgniteTableImpl extends AbstractTable implements IgniteTable {
-
- private final TableDescriptor desc;
-
- private final int ver;
-
- private final int id;
-
- private final String name;
-
- private final Statistic statistic;
-
- private final Map<String, IgniteIndex> indexes = new HashMap<>();
-
- /**
- * Constructor.
- *
- * @param desc Table descriptor.
- * @param tableId Table id.
- * @param name Table name.
- */
- IgniteTableImpl(TableDescriptor desc, int tableId, String name, int version,
- DoubleSupplier rowCount) {
- this.ver = version;
- this.desc = desc;
- this.id = tableId;
- this.name = name;
- this.statistic = new IgniteStatistic(rowCount, desc.distribution());
- }
-
- private IgniteTableImpl(IgniteTableImpl t) {
- this.desc = t.desc;
- this.ver = t.ver;
- this.id = t.id;
- this.name = t.name;
- this.statistic = t.statistic;
- this.indexes.putAll(t.indexes);
- }
-
- public static IgniteTableImpl copyOf(IgniteTableImpl v) {
- return new IgniteTableImpl(v);
- }
-
- /** {@inheritDoc} */
- @Override
- public int id() {
- return id;
- }
-
- /** {@inheritDoc} */
- @Override
- public int version() {
- return ver;
- }
-
- /** {@inheritDoc} */
- @Override
- public String name() {
- return name;
- }
-
- /** {@inheritDoc} */
- @Override
- public RelDataType getRowType(RelDataTypeFactory typeFactory, ImmutableBitSet requiredColumns) {
- return desc.rowType((IgniteTypeFactory) typeFactory, requiredColumns);
- }
-
- /** {@inheritDoc} */
- @Override
- public Statistic getStatistic() {
- return statistic;
- }
-
- /** {@inheritDoc} */
- @Override
- public TableDescriptor descriptor() {
- return desc;
- }
-
- /** {@inheritDoc} */
- @Override
- public IgniteLogicalTableScan toRel(
- RelOptCluster cluster,
- RelOptTable relOptTbl,
- List<RelHint> hints,
- @Nullable List<RexNode> proj,
- @Nullable RexNode cond,
- @Nullable ImmutableBitSet requiredColumns
- ) {
- RelTraitSet traitSet = cluster.traitSetOf(distribution());
-
- return IgniteLogicalTableScan.create(cluster, traitSet, hints, relOptTbl, proj, cond, requiredColumns);
- }
-
- /** {@inheritDoc} */
- @Override
- public IgniteLogicalIndexScan toRel(
- RelOptCluster cluster,
- RelOptTable relOptTable,
- String idxName,
- List<RexNode> proj,
- RexNode condition,
- ImmutableBitSet requiredCols
- ) {
- IgniteIndex index = getIndex(idxName);
-
- RelCollation collation = TraitUtils.createCollation(index.columns(), index.collations(), descriptor());
-
- RelTraitSet traitSet = cluster.traitSetOf(Convention.Impl.NONE)
- .replace(distribution())
- .replace(index.type() == Type.HASH ? RelCollations.EMPTY : collation);
-
- return IgniteLogicalIndexScan.create(cluster, traitSet, relOptTable, idxName, proj, condition, requiredCols);
- }
-
- /** {@inheritDoc} */
- @Override
- public IgniteDistribution distribution() {
- return desc.distribution();
- }
-
- /** {@inheritDoc} */
- @Override
- public Map<String, IgniteIndex> indexes() {
- return Collections.unmodifiableMap(indexes);
- }
-
- /** {@inheritDoc} */
- @Override
- public void addIndex(IgniteIndex idxTbl) {
- indexes.put(idxTbl.name(), idxTbl);
- }
-
- /** {@inheritDoc} */
- @Override
- public IgniteIndex getIndex(String idxName) {
- return indexes.get(idxName);
- }
-
- /** {@inheritDoc} */
- @Override
- public void removeIndex(String idxName) {
- indexes.remove(idxName);
- }
-
- /** {@inheritDoc} */
- @Override
- public <C> C unwrap(Class<C> cls) {
- if (cls.isInstance(desc)) {
- return cls.cast(desc);
- }
-
- return super.unwrap(cls);
- }
-
- static DoubleSupplier rowCountStatistic(InternalTable table) {
- return new RowCountStatistic(table);
- }
-
- private static final class RowCountStatistic implements DoubleSupplier {
- private static final int UPDATE_THRESHOLD = DistributionZoneManager.DEFAULT_PARTITION_COUNT;
-
- private final AtomicLong lastUpd = new AtomicLong();
-
- private volatile long localRowCnt = 0L;
-
- private final InternalTable table;
-
- private RowCountStatistic(InternalTable table) {
- this.table = table;
- }
-
- /** {@inheritDoc} */
- @Override
- // TODO: need to be refactored https://issues.apache.org/jira/browse/IGNITE-19558
- public double getAsDouble() {
- int parts = table.storage().getTableDescriptor().getPartitions();
-
- long partitionsRevisionCounter = 0L;
-
- for (int p = 0; p < parts; ++p) {
- @Nullable MvPartitionStorage part = table.storage().getMvPartition(p);
-
- if (part == null) {
- continue;
- }
-
- long upd = part.lastAppliedIndex();
-
- partitionsRevisionCounter += upd;
- }
-
- long prev = lastUpd.get();
-
- if (partitionsRevisionCounter - prev > UPDATE_THRESHOLD) {
- synchronized (this) {
- if (lastUpd.compareAndSet(prev, partitionsRevisionCounter)) {
- long size = 0L;
-
- for (int p = 0; p < parts; ++p) {
- @Nullable MvPartitionStorage part = table.storage().getMvPartition(p);
-
- if (part == null) {
- continue;
- }
-
- try {
- size += part.rowsCount();
- } catch (StorageRebalanceException ignore) {
- // No-op.
- }
- }
-
- localRowCnt = size;
- }
- }
- }
-
- // Forbid zero result, to prevent zero cost for table and index scans.
- return Math.max(10_000.0, (double) localRowCnt);
- }
- }
-}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index 627c2e81a3..319f27736f 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -141,7 +141,7 @@ public class ExecutionServiceImplTest {
private final TestTable table = createTable("TEST_TBL", 1_000_000, IgniteDistributions.random(),
"ID", NativeTypes.INT32, "VAL", NativeTypes.INT32);
- private final IgniteSchema schema = new IgniteSchema("PUBLIC", Map.of(table.name(), table), null, SCHEMA_VERSION);
+ private final IgniteCatalogSchema schema = new IgniteCatalogSchema("PUBLIC", SCHEMA_VERSION, Map.of(table.name(), table));
private final List<CapturingMailboxRegistry> mailboxes = new ArrayList<>();
@@ -640,7 +640,7 @@ public class ExecutionServiceImplTest {
.build();
}
- private SchemaPlus wrap(IgniteSchema schema) {
+ private SchemaPlus wrap(IgniteCatalogSchema schema) {
var schemaPlus = Frameworks.createRootSchema(false);
schemaPlus.add(schema.getName(), schema);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
index 582087733d..637e77058e 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
@@ -28,6 +28,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.Frameworks;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
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.SqlSchemaManager;
@@ -47,16 +48,16 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
private final Map<Integer, IgniteTable> tableById;
/** Constructs schema manager from a single schema. */
- PredefinedSchemaManager(IgniteSchema schema) {
+ PredefinedSchemaManager(IgniteCatalogSchema schema) {
this(List.of(schema));
}
/** Constructs schema manager from a collection of schemas. */
- PredefinedSchemaManager(Collection<IgniteSchema> schemas) {
+ PredefinedSchemaManager(Collection<IgniteCatalogSchema> schemas) {
this.root = Frameworks.createRootSchema(false);
this.tableById = new HashMap<>();
- for (IgniteSchema schema : schemas) {
+ for (IgniteCatalogSchema schema : schemas) {
root.add(schema.getName(), schema);
tableById.putAll(
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
index db75948eac..270da0b545 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
@@ -45,10 +45,9 @@ import org.apache.ignite.internal.sql.engine.metadata.FragmentDescription;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptorImpl;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Collation;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
-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.TableDescriptorImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
@@ -355,12 +354,7 @@ public class TestBuilders {
.map(ClusterTableBuilderImpl::build)
.collect(Collectors.toMap(TestTable::name, Function.identity()));
- Map<Integer, IgniteIndex> indexMap = tableMap.values().stream()
- .map(TestTable.class::cast)
- .flatMap(t -> t.indexes().values().stream().map(TestIndex.class::cast))
- .collect(Collectors.toMap(TestIndex::id, Function.identity()));
-
- var schemaManager = new PredefinedSchemaManager(new IgniteSchema("PUBLIC", tableMap, indexMap, SCHEMA_VERSION));
+ var schemaManager = new PredefinedSchemaManager(new IgniteCatalogSchema("PUBLIC", SCHEMA_VERSION, tableMap));
var colocationGroupProvider = new TestColocationGroupProvider(tableBuilders, tableMap, nodeNames);
Map<String, TestNode> nodes = nodeNames.stream()
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
index ec46c53ae6..4e860d7745 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.sql.engine.metadata.ColocationGroup;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalIndexScan;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
@@ -239,8 +240,8 @@ public class TestTable implements IgniteTable {
/** {@inheritDoc} */
@Override
- public Map<String, IgniteIndex> indexes() {
- return Collections.unmodifiableMap(indexes);
+ public Map<String, IgniteSchemaIndex> indexes() {
+ throw new UnsupportedOperationException("Not implemented yet");
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
index c8da086c5f..712ec8e466 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
@@ -113,6 +113,7 @@ import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
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.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
@@ -979,8 +980,8 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
/** {@inheritDoc} */
@Override
- public Map<String, IgniteIndex> indexes() {
- return Collections.unmodifiableMap(indexes);
+ public Map<String, IgniteSchemaIndex> indexes() {
+ throw new UnsupportedOperationException("Not implemented yet");
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
index 0fa0228799..5dd2b561db 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
@@ -375,7 +375,7 @@ public class CatalogSqlSchemaManagerTest {
IgniteSchemaTable table = (IgniteSchemaTable) schema.getTable(testTable.name);
assertNotNull(table);
- IgniteSchemaIndex testIdx = table.getIndexes().get(testIndex.name);
+ IgniteSchemaIndex testIdx = table.indexes().get(testIndex.name);
assertEquals(testIndex.name, testIdx.name());
assertEquals(Type.HASH, testIdx.type());
@@ -419,7 +419,7 @@ public class CatalogSqlSchemaManagerTest {
IgniteSchemaTable table = (IgniteSchemaTable) schema.getTable(testTable.name);
assertNotNull(table);
- IgniteSchemaIndex testIdx = table.getIndexes().get(testIndex.name);
+ IgniteSchemaIndex testIdx = table.indexes().get(testIndex.name);
assertEquals(testIndex.name, testIdx.name());
assertEquals(Type.SORTED, testIdx.type());
[ignite-3] 06/06: Styles.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 22d25603af24089ec408d0a3e40e080afe396fc7
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 23:34:56 2023 +0300
Styles.
---
.../sql/engine/exec/ExecutionServiceImpl.java | 2 +-
.../internal/sql/engine/rel/AbstractIndexScan.java | 3 +-
.../engine/rel/logical/IgniteLogicalIndexScan.java | 5 +-
.../engine/rule/logical/LogicalOrToUnionRule.java | 3 --
.../sql/engine/schema/CatalogSqlSchemaManager.java | 1 -
.../sql/engine/schema/IgniteSchemaIndex.java | 13 +++++
.../sql/engine/exec/ExecutionServiceImplTest.java | 6 +--
.../engine/framework/PredefinedSchemaManager.java | 10 ++--
.../sql/engine/framework/TestBuilders.java | 56 ++++++++++++++--------
.../internal/sql/engine/framework/TestIndex.java | 40 +++++-----------
.../internal/sql/engine/framework/TestNode.java | 2 +-
.../internal/sql/engine/framework/TestTable.java | 19 ++++----
12 files changed, 85 insertions(+), 75 deletions(-)
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 e330ebd4ed..a89cf33464 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
@@ -428,7 +428,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private DistributedQueryManager getOrCreateQueryManager(QueryStartRequest msg) {
return queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
- BaseQueryContext ctx = createQueryContext(key, msg.schema(), (int)msg.schemaVersion(), msg.parameters());
+ 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/rel/AbstractIndexScan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIndexScan.java
index d7e416f6ce..27617f13ba 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIndexScan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/AbstractIndexScan.java
@@ -41,7 +41,6 @@ import org.apache.ignite.internal.sql.engine.metadata.cost.IgniteCost;
import org.apache.ignite.internal.sql.engine.prepare.bounds.MultiBounds;
import org.apache.ignite.internal.sql.engine.prepare.bounds.SearchBounds;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.jetbrains.annotations.Nullable;
@@ -126,7 +125,7 @@ public abstract class AbstractIndexScan extends ProjectableFilterableTableScan {
double cost = 0;
- if (type == Type.HASH) {
+ if (type == IgniteIndex.Type.HASH) {
boolean notExact = (searchBounds() == null)
|| searchBounds().stream().anyMatch(bound -> bound.type() == SearchBounds.Type.RANGE);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
index 25ba268f2a..3bcc96c058 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rel/logical/IgniteLogicalIndexScan.java
@@ -27,9 +27,8 @@ import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.internal.sql.engine.prepare.bounds.SearchBounds;
import org.apache.ignite.internal.sql.engine.rel.AbstractIndexScan;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.apache.ignite.internal.sql.engine.util.Commons;
@@ -111,7 +110,7 @@ public class IgniteLogicalIndexScan extends AbstractIndexScan {
RelTraitSet traits,
RelOptTable tbl,
String idxName,
- IgniteIndex.Type type,
+ Type type,
@Nullable List<RexNode> proj,
@Nullable RexNode cond,
@Nullable List<SearchBounds> searchBounds,
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
index e7f1463ffe..26d905975d 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/LogicalOrToUnionRule.java
@@ -26,7 +26,6 @@ import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.RelRule;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rex.RexBuilder;
@@ -38,10 +37,8 @@ import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.internal.sql.engine.rel.logical.IgniteLogicalTableScan;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
-import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.util.CollectionUtils;
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 861018ba3b..c0c7ce63cb 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
@@ -50,7 +50,6 @@ import org.apache.ignite.internal.schema.DefaultValueGenerator;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
index 09870c0fdc..6a237efe97 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchemaIndex.java
@@ -155,6 +155,19 @@ public class IgniteSchemaIndex {
}
//TODO: cache rowType as it can't be changed.
+ /**
+ * Returns index row type.
+ *
+ * <p>This is a struct type whose fields describe the names and types of indexed columns.</p>
+ *
+ * <p>The implementer must use the type factory provided. This ensures that
+ * the type is converted into a canonical form; other equal types in the same
+ * query will use the same object.</p>
+ *
+ * @param typeFactory Type factory with which to create the type
+ * @param tableDescriptor Table descriptor.
+ * @return Row type.
+ */
public RelDataType getRowType(IgniteTypeFactory typeFactory, TableDescriptor tableDescriptor) {
RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(typeFactory);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index 319f27736f..4d6d1054b9 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -34,7 +34,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -90,7 +89,6 @@ import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptorImpl;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptorImpl;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
@@ -782,7 +780,9 @@ public class ExecutionServiceImplTest {
MailboxRegistry mailboxRegistry,
ExchangeService exchangeService,
ResolvedDependencies deps) {
- HashFunctionFactory<Object[]> funcFactory = new HashFunctionFactoryImpl<>(ctx.getRootSchema().unwrap(IgniteCatalogSchema.class), ctx.rowHandler());
+ IgniteCatalogSchema schema = ctx.getRootSchema().unwrap(IgniteCatalogSchema.class);
+
+ HashFunctionFactory<Object[]> funcFactory = new HashFunctionFactoryImpl<>(schema, ctx.rowHandler());
return new LogicalRelImplementor<>(ctx, funcFactory, mailboxRegistry, exchangeService, deps) {
@Override
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
index 637e77058e..612a7e2ff0 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
@@ -71,8 +71,8 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(String name, int version) {
- return schema(name);
+ public @Nullable SchemaPlus schema(@Nullable String name, int version) {
+ return getSchema(name);
}
/** {@inheritDoc} */
@@ -83,11 +83,11 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus latestSchema(@Nullable String name) {
- return schema(name);
+ public @Nullable SchemaPlus latestSchema(@Nullable String name) {
+ return getSchema(name);
}
- private SchemaPlus schema(@Nullable String schemaName) {
+ private @Nullable SchemaPlus getSchema(@Nullable String schemaName) {
return root.getSubSchema(schemaName == null ? DEFAULT_SCHEMA_NAME : schemaName);
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
index 270da0b545..69f30e2673 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
@@ -33,6 +33,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.schema.Table;
import org.apache.ignite.internal.schema.NativeType;
import org.apache.ignite.internal.sql.engine.exec.ArrayRowHandler;
@@ -47,10 +48,12 @@ import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptorImpl;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
import org.apache.ignite.internal.sql.engine.schema.IgniteCatalogSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Collation;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
+import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptorImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
+import org.apache.ignite.internal.sql.engine.trait.TraitUtils;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.network.ClusterNode;
import org.jetbrains.annotations.Nullable;
@@ -424,16 +427,19 @@ public class TestBuilders {
throw new IllegalArgumentException("Table must contain at least one column");
}
- TestTable testTable = new TestTable(
- new TableDescriptorImpl(columns, distribution),
+ TableDescriptorImpl tableDescriptor = new TableDescriptorImpl(columns, distribution);
+
+ Map<String, IgniteSchemaIndex> indexMap = indexBuilders.stream()
+ .map(idx -> idx.build(tableDescriptor))
+ .collect(Collectors.toMap(TestIndex::name, Function.identity()));
+
+ return new TestTable(
+ tableDescriptor,
Objects.requireNonNull(name),
Map.of(),
- size
+ size,
+ indexMap
);
-
- indexBuilders.stream().map(AbstractIndexBuilderImpl::build).forEach(testTable::addIndex);
-
- return testTable;
}
/** {@inheritDoc} */
@@ -477,11 +483,13 @@ public class TestBuilders {
}
private TestTable build() {
- TestTable testTable = new TestTable(new TableDescriptorImpl(columns, distribution), name, dataProviders, size);
+ TableDescriptorImpl tableDescriptor = new TableDescriptorImpl(columns, distribution);
- indexBuilders.forEach(idx -> testTable.addIndex(idx.build()));
+ Map<String, IgniteSchemaIndex> indexMap = indexBuilders.stream()
+ .map(idx -> idx.build(tableDescriptor))
+ .collect(Collectors.toMap(TestIndex::name, Function.identity()));
- return testTable;
+ return new TestTable(tableDescriptor, name, dataProviders, size, indexMap);
}
}
@@ -509,7 +517,7 @@ public class TestBuilders {
/** {@inheritDoc} */
@Override
- public TestIndex build() {
+ public TestIndex build(TableDescriptor desc) {
if (name == null) {
throw new IllegalArgumentException("Name is not specified");
}
@@ -522,7 +530,9 @@ public class TestBuilders {
throw new IllegalArgumentException("Collation must be specified for each of columns.");
}
- return new TestIndex(name, Type.SORTED, columns, collations, dataProviders);
+ RelCollation collation = TraitUtils.createCollation(columns, collations, desc);
+
+ return TestIndex.createSorted(name, desc.distribution(), collation, dataProviders);
}
}
@@ -549,7 +559,7 @@ public class TestBuilders {
/** {@inheritDoc} */
@Override
- public TestIndex build() {
+ public TestIndex build(TableDescriptor desc) {
if (name == null) {
throw new IllegalArgumentException("Name is not specified");
}
@@ -560,7 +570,9 @@ public class TestBuilders {
assert collations == null : "Collation is not supported.";
- return new TestIndex(name, Type.HASH, columns, null, dataProviders);
+ RelCollation collation = TraitUtils.createCollation(columns, collations, desc);
+
+ return TestIndex.createHash(name, desc.distribution(), collation, dataProviders);
}
}
@@ -587,10 +599,12 @@ public class TestBuilders {
}
@Override
- TestIndex build() {
+ TestIndex build(TableDescriptor desc) {
assert collations.size() == columns.size();
- return TestIndex.createSorted(name, columns, collations, dataProviders);
+ RelCollation collation = TraitUtils.createCollation(columns, collations, desc);
+
+ return TestIndex.createSorted(name, desc.distribution(), collation, dataProviders);
}
}
@@ -617,10 +631,12 @@ public class TestBuilders {
}
@Override
- TestIndex build() {
+ TestIndex build(TableDescriptor desc) {
assert collations == null;
- return TestIndex.createHash(name, columns, dataProviders);
+ RelCollation collation = TraitUtils.createCollation(columns, collations, desc);
+
+ return TestIndex.createHash(name, desc.distribution(), collation, dataProviders);
}
}
@@ -723,7 +739,7 @@ public class TestBuilders {
return self();
}
- abstract TestIndex build();
+ abstract TestIndex build(TableDescriptor desc);
}
private abstract static class AbstractDataSourceBuilderImpl<ChildT> {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
index dd8449e6bb..f7678459d3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
@@ -19,64 +19,50 @@ package org.apache.ignite.internal.sql.engine.framework;
import static org.apache.ignite.lang.IgniteStringFormatter.format;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
-import org.jetbrains.annotations.Nullable;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchemaIndex;
+import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
/**
* A test index that implements all the necessary for the optimizer methods to be used to prepare a query, as well as provides access to the
* data to use this index in execution-related scenarios.
*/
-public class TestIndex extends IgniteIndex {
+public class TestIndex extends IgniteSchemaIndex {
private static final String DATA_PROVIDER_NOT_CONFIGURED_MESSAGE_TEMPLATE =
"DataProvider is not configured [index={}, node={}]";
/** Factory method for creating hash-index. */
- static TestIndex createHash(String name, List<String> indexedColumns, Map<String, DataProvider<?>> dataProviders) {
- return new TestIndex(name, Type.HASH, indexedColumns, null, dataProviders);
+ static TestIndex createHash(String name, IgniteDistribution distribution, RelCollation collation,
+ Map<String, DataProvider<?>> dataProviders) {
+ return new TestIndex(name, Type.HASH, distribution, collation, dataProviders);
}
/** Factory method for creating sorted-index. */
- static TestIndex createSorted(String name, List<String> columns, List<Collation> collations,
+ static TestIndex createSorted(String name, IgniteDistribution distribution, RelCollation collation,
Map<String, DataProvider<?>> dataProviders) {
- return new TestIndex(name, Type.SORTED, columns, collations, dataProviders);
+ return new TestIndex(name, Type.SORTED, distribution, collation, dataProviders);
}
private static final AtomicInteger ID = new AtomicInteger();
- private final int id = ID.incrementAndGet();
- private final String name;
-
private final Map<String, DataProvider<?>> dataProviders;
/** Constructor. */
TestIndex(
String name,
Type type,
- List<String> columns,
- @Nullable List<Collation> collations,
+ IgniteDistribution distribution,
+ RelCollation collation,
Map<String, DataProvider<?>> dataProviders
) {
- super(type, columns, collations);
+ super(ID.incrementAndGet(), name, type, distribution, collation);
- this.name = name;
this.dataProviders = dataProviders;
}
- /** Returns an id of the index. */
- @Override
- public int id() {
- return id;
- }
-
- /** {@inheritDoc} */
- @Override
- public String name() {
- return name;
- }
-
/**
* Returns the data provider for the given node.
*
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
index 031f7a8ac3..440ca7ed6d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
@@ -146,7 +146,7 @@ public class TestNode implements LifecycleAware {
@Override
public Node<Object[]> visit(IgniteIndexScan rel) {
TestTable tbl = rel.getTable().unwrap(TestTable.class);
- TestIndex idx = (TestIndex) tbl.getIndex(rel.indexName());
+ TestIndex idx = (TestIndex) tbl.indexes().get(rel.indexName());
DataProvider<Object[]> dataProvider = idx.dataProvider(ctx.localNode().name());
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
index 4e860d7745..87ad26f24d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestTable.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.sql.engine.framework;
import static org.apache.ignite.lang.IgniteStringFormatter.format;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -51,9 +50,8 @@ import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.jetbrains.annotations.Nullable;
/**
- * A test table that implements all the necessary for the optimizer methods to be used
- * to prepare a query, as well as provides access to the data to use this table in
- * execution-related scenarios.
+ * A test table that implements all the necessary for the optimizer methods to be used to prepare a query, as well as provides access to the
+ * data to use this table in execution-related scenarios.
*/
public class TestTable implements IgniteTable {
private static final String DATA_PROVIDER_NOT_CONFIGURED_MESSAGE_TEMPLATE =
@@ -62,7 +60,7 @@ public class TestTable implements IgniteTable {
private static final AtomicInteger ID = new AtomicInteger();
private final int id = ID.incrementAndGet();
- private final Map<String, IgniteIndex> indexes = new HashMap<>();
+ private final Map<String, IgniteSchemaIndex> indexes;
private final String name;
private final double rowCnt;
@@ -83,6 +81,7 @@ public class TestTable implements IgniteTable {
this.rowCnt = rowCnt;
this.colocationGroup = colocationGroup;
+ indexes = Map.of();
dataProviders = Collections.emptyMap();
}
@@ -91,12 +90,14 @@ public class TestTable implements IgniteTable {
TableDescriptor descriptor,
String name,
Map<String, DataProvider<?>> dataProviders,
- double rowCnt
+ double rowCnt,
+ Map<String, IgniteSchemaIndex> indexMap
) {
this.descriptor = descriptor;
this.name = name;
this.rowCnt = rowCnt;
this.dataProviders = dataProviders;
+ indexes = indexMap;
this.colocationGroup = ColocationGroup.forNodes(List.copyOf(dataProviders.keySet()));
}
@@ -241,19 +242,19 @@ public class TestTable implements IgniteTable {
/** {@inheritDoc} */
@Override
public Map<String, IgniteSchemaIndex> indexes() {
- throw new UnsupportedOperationException("Not implemented yet");
+ return indexes;
}
/** {@inheritDoc} */
@Override
public void addIndex(IgniteIndex idxTbl) {
- indexes.put(idxTbl.name(), idxTbl);
+ throw new UnsupportedOperationException("Not implemented yet");
}
/** {@inheritDoc} */
@Override
public IgniteIndex getIndex(String idxName) {
- return indexes.get(idxName);
+ throw new UnsupportedOperationException("Not implemented yet");
}
/** {@inheritDoc} */
[ignite-3] 02/06: Use schema versions for queries. * Use actual schema for query in TX. * Pass schema version to remote fragment. * Wait for catalog schema ready on the remote side.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-19497
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit bbf8eb5ed167fbbe41624285f81fe843afaec846
Author: amashenkov <an...@gmail.com>
AuthorDate: Tue Aug 1 20:53:41 2023 +0300
Use schema versions for queries.
* Use actual schema for query in TX.
* Pass schema version to remote fragment.
* Wait for catalog schema ready on the remote side.
---
.../java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java | 5 +++++
.../java/org/apache/ignite/internal/catalog/CatalogService.java | 6 ++++++
.../org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java | 4 +++-
.../ignite/internal/sql/engine/exec/ExecutionServiceImpl.java | 6 +++---
.../ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java | 2 +-
.../apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java | 1 +
.../ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java | 2 +-
.../internal/sql/engine/framework/PredefinedSchemaManager.java | 2 +-
.../org/apache/ignite/internal/sql/engine/framework/TestNode.java | 6 +++---
9 files changed, 24 insertions(+), 10 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
index eb606160e3..16a9a1e8e0 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
@@ -275,6 +275,11 @@ public class CatalogManagerImpl extends Producer<CatalogEvent, CatalogEventParam
return catalogByVer.lastEntry().getKey();
}
+ @Override
+ public CompletableFuture<Void> catalogReadyFuture(int ver) {
+ return versionTracker.waitFor(ver);
+ }
+
private Catalog catalog(int version) {
return catalogByVer.get(version);
}
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 c1d30f1b4c..314ad24075 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
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.catalog;
import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
@@ -79,4 +80,9 @@ public interface CatalogService {
int latestCatalogVersion();
void listen(CatalogEvent evt, EventListener<CatalogEventParameters> closure);
+
+ /**
+ * Returns a future, which completes, when catalog of given version will be available.
+ */
+ CompletableFuture<Void> catalogReadyFuture(int ver);
}
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 1ba38288b1..9098326e46 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
@@ -423,7 +423,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 6b483f8b0e..50428f0b91 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
@@ -239,13 +239,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 catalogVersion, Object[] params) {
return BaseQueryContext.builder()
.queryId(queryId)
.parameters(params)
.frameworkConfig(
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(sqlSchemaManager.schema(schema))
+ .defaultSchema(sqlSchemaManager.schema(schema, catalogVersion))
.build()
)
.logger(LOG)
@@ -427,7 +427,7 @@ 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());
+ 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..371f885abc 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
@@ -93,7 +93,7 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
public CompletableFuture<?> actualSchemaAsync(long ver) {
- return CompletableFuture.completedFuture(catalogManager.schema((int) ver));
+ return catalogManager.catalogReadyFuture((int) ver);
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
index ef45bc8598..e2c1fb6e1b 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java
@@ -47,6 +47,7 @@ public interface SqlSchemaManager {
/**
* Wait for {@code ver} schema version, just a stub, need to be removed after IGNITE-18733.
*/
+ @Deprecated
CompletableFuture<?> actualSchemaAsync(long ver);
/**
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index e57170b284..ec61eca2fb 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -610,7 +610,7 @@ public class ExecutionServiceImplTest {
rootSch.add(schema.getName(), schema);
SchemaPlus plus = rootSch.plus().getSubSchema(schema.getName());
- when(schemaManagerMock.schema(any())).thenReturn(plus);
+ when(schemaManagerMock.schema(any(), anyInt())).thenReturn(plus);
var executionService = new ExecutionServiceImpl<>(
messageService,
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
index c98b745510..73ddc9098b 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java
@@ -81,7 +81,7 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
public CompletableFuture<?> actualSchemaAsync(long ver) {
- return CompletableFuture.completedFuture(root);
+ return CompletableFuture.completedFuture(null);
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
index de4e900d96..88f270c581 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
@@ -79,12 +79,12 @@ import org.apache.ignite.network.TopologyService;
*/
public class TestNode implements LifecycleAware {
private final String nodeName;
- private final SchemaPlus schema;
private final PrepareService prepareService;
private final ExecutionService executionService;
private final ParserService parserService;
private final List<LifecycleAware> services = new ArrayList<>();
+ private final SqlSchemaManager schemaManager;
/**
* Constructs the object.
@@ -101,7 +101,7 @@ public class TestNode implements LifecycleAware {
) {
this.nodeName = nodeName;
this.prepareService = registerService(new PrepareServiceImpl(nodeName, 0, mock(DdlSqlToCommandConverter.class), PLANNING_TIMEOUT));
- this.schema = schemaManager.schema("PUBLIC");
+ this.schemaManager = schemaManager;
TopologyService topologyService = clusterService.topologyService();
MessagingService messagingService = clusterService.messagingService();
@@ -223,7 +223,7 @@ public class TestNode implements LifecycleAware {
.cancel(new QueryCancel())
.frameworkConfig(
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(schema)
+ .defaultSchema(schemaManager.latestSchema("PUBLIC"))
.build()
)
.build();