You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ko...@apache.org on 2023/10/23 16:09:27 UTC
[ignite-3] branch main updated: IGNITE-20439 Sql. Support multiple schemas in CatalogSqlSchemaManager (#2719)
This is an automated email from the ASF dual-hosted git repository.
korlov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 0304a5745a IGNITE-20439 Sql. Support multiple schemas in CatalogSqlSchemaManager (#2719)
0304a5745a is described below
commit 0304a5745a3bbe20f6accd1ab0a21e3ecb6d3865
Author: korlov42 <ko...@gridgain.com>
AuthorDate: Mon Oct 23 19:09:21 2023 +0300
IGNITE-20439 Sql. Support multiple schemas in CatalogSqlSchemaManager (#2719)
---
.../internal/catalog/CatalogManagerImpl.java | 3 +-
.../ignite/internal/catalog/CatalogService.java | 2 +
.../internal/catalog/CatalogTestUtilsTest.java | 102 +++
.../ignite/internal/catalog/CatalogTestUtils.java | 89 +++
.../ignite/internal/sql/api/ItCommonApiTest.java | 11 +-
.../internal/sql/engine/ItSystemViewsTest.java | 204 +++---
.../internal/sql/engine/SqlQueryProcessor.java | 19 +-
.../sql/engine/exec/ExecutableTableRegistry.java | 6 +-
.../engine/exec/ExecutableTableRegistryImpl.java | 25 +-
.../internal/sql/engine/exec/ExecutionContext.java | 2 +-
.../engine/exec/ExecutionDependencyResolver.java | 3 +-
.../exec/ExecutionDependencyResolverImpl.java | 20 +-
.../sql/engine/exec/ExecutionServiceImpl.java | 17 +-
.../internal/sql/engine/exec/rel/AbstractNode.java | 17 +-
.../sql/engine/message/QueryStartRequest.java | 5 -
.../sql/engine/prepare/PrepareServiceImpl.java | 9 +-
.../internal/sql/engine/schema/IgniteSchema.java | 8 +-
.../sql/engine/schema/SqlSchemaManager.java | 18 +-
...chemaManager.java => SqlSchemaManagerImpl.java} | 186 ++---
.../internal/sql/engine/util/BaseQueryContext.java | 4 -
.../engine/util/cache/CaffeineCacheFactory.java | 2 +-
.../exec/ExecutableTableRegistrySelfTest.java | 20 +-
.../exec/ExecutionDependencyResolverSelfTest.java | 32 +-
.../sql/engine/exec/ExecutionServiceImplTest.java | 6 +-
.../engine/exec/NoOpExecutableTableRegistry.java | 2 +-
.../engine/framework/PredefinedSchemaManager.java | 20 +-
.../internal/sql/engine/framework/TestNode.java | 10 +-
.../sql/engine/planner/PlannerTimeoutTest.java | 4 +-
.../engine/schema/CatalogSqlSchemaManagerTest.java | 756 ---------------------
.../engine/schema/SqlSchemaManagerImplTest.java | 708 +++++++++++++++++++
.../sql/metrics/PlanningCacheMetricsTest.java | 4 +-
.../internal/sql/engine/util/QueryChecker.java | 2 +
.../internal/sql/engine/util/QueryCheckerImpl.java | 88 ++-
33 files changed, 1310 insertions(+), 1094 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 9a6ead5252..180057fd31 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
@@ -296,7 +296,8 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
return versionTracker.waitFor(version);
}
- private Catalog catalog(int version) {
+ @Override
+ public 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 f514b310f6..b5b11d804c 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
@@ -49,6 +49,8 @@ public interface CatalogService extends EventProducer<CatalogEvent, CatalogEvent
String DEFAULT_ZONE_NAME = "Default";
+ Catalog catalog(int version);
+
@Nullable CatalogTableDescriptor table(String tableName, long timestamp);
@Nullable CatalogTableDescriptor table(int tableId, long timestamp);
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogTestUtilsTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogTestUtilsTest.java
new file mode 100644
index 0000000000..0fd4d1fbb4
--- /dev/null
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogTestUtilsTest.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.catalog;
+
+import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.hasSize;
+
+import java.util.Collection;
+import java.util.List;
+import org.apache.ignite.internal.catalog.commands.ColumnParams;
+import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
+import org.apache.ignite.internal.catalog.commands.CreateTableCommandBuilder;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests to verify {@link CatalogTestUtils}.
+ */
+class CatalogTestUtilsTest extends BaseIgniteAbstractTest {
+
+ /**
+ * Simple smoke test to verify test manager is able to process several versions of catalog,
+ * and returned instance follows the contract.
+ */
+ @Test
+ void testManagerWorksAsExpected() throws Exception {
+ CatalogManager manager = CatalogTestUtils.createCatalogManagerWithTestUpdateLog("test", new HybridClockImpl());
+
+ manager.start();
+
+ CreateTableCommandBuilder createTableTemplate = CreateTableCommand.builder()
+ .schemaName("PUBLIC")
+ .columns(List.of(
+ ColumnParams.builder().name("C1").type(ColumnType.INT32).build(),
+ ColumnParams.builder().name("C2").type(ColumnType.INT32).build()
+ ))
+ .primaryKeyColumns(List.of("C1"));
+
+ await(manager.execute(
+ createTableTemplate.tableName("T1").build()
+ ));
+
+ int version1 = manager.latestCatalogVersion();
+
+ await(manager.execute(
+ createTableTemplate.tableName("T2").build()
+ ));
+
+ int version2 = manager.latestCatalogVersion();
+
+ Collection<CatalogTableDescriptor> tablesOfVersion1 = manager.tables(version1);
+
+ assertThat(tablesOfVersion1, hasSize(1));
+ assertThat(tablesOfVersion1, hasItem(descriptorWithName("T1")));
+
+ Collection<CatalogTableDescriptor> tablesOfVersion2 = manager.tables(version2);
+
+ assertThat(tablesOfVersion2, hasSize(2));
+ assertThat(tablesOfVersion2, hasItem(descriptorWithName("T1")));
+ assertThat(tablesOfVersion2, hasItem(descriptorWithName("T2")));
+
+ manager.stop();
+ }
+
+ private static Matcher<CatalogTableDescriptor> descriptorWithName(String name) {
+ return new BaseMatcher<>() {
+ @Override
+ public boolean matches(Object actual) {
+ return actual instanceof CatalogTableDescriptor && name.equals(((CatalogTableDescriptor) actual).name());
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText(format("should have name '{}'", name));
+ }
+ };
+ }
+}
diff --git a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
index 475a73e79d..141cbbb148 100644
--- a/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
+++ b/modules/catalog/src/testFixtures/java/org/apache/ignite/internal/catalog/CatalogTestUtils.java
@@ -17,24 +17,30 @@
package org.apache.ignite.internal.catalog;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_SCHEMA_NAME;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.catalog.commands.AlterTableAddColumnCommand;
import org.apache.ignite.internal.catalog.commands.AlterTableDropColumnCommand;
import org.apache.ignite.internal.catalog.commands.ColumnParams;
import org.apache.ignite.internal.catalog.commands.ColumnParams.Builder;
import org.apache.ignite.internal.catalog.commands.DropTableCommand;
+import org.apache.ignite.internal.catalog.storage.UpdateLog;
import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
+import org.apache.ignite.internal.catalog.storage.VersionedUpdate;
import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
+import org.apache.ignite.lang.ErrorGroups.Common;
import org.apache.ignite.sql.ColumnType;
/**
@@ -124,6 +130,46 @@ public class CatalogTestUtils {
};
}
+ /**
+ * Create the same {@link CatalogManager} as for normal operations, but with {@link UpdateLog} that
+ * simply notifies the manager without storing any updates in metastore.
+ *
+ * <p>Particular configuration of manager pretty fast (in terms of awaiting of certain safe time) and lightweight.
+ * It doesn't contain any mocks from {@link org.mockito.Mockito}.
+ *
+ * @param nodeName Name of the node that is meant to own this manager. Any thread spawned by returned instance
+ * will have it as thread's name prefix.
+ * @param clock This clock is used to assign activation timestamp for incoming updates, thus make it possible
+ * to acquired schema that was valid at give time.
+ * @return An instance of {@link CatalogManager catalog manager}.
+ */
+ public static CatalogManager createCatalogManagerWithTestUpdateLog(String nodeName, HybridClock clock) {
+ var clockWaiter = new ClockWaiter(nodeName, clock);
+
+ return new CatalogManagerImpl(new TestUpdateLog(clock), clockWaiter) {
+ @Override
+ public void start() {
+ clockWaiter.start();
+
+ super.start();
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ super.beforeNodeStop();
+
+ clockWaiter.beforeNodeStop();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ super.stop();
+
+ clockWaiter.stop();
+ }
+ };
+ }
+
/** Default nullable behavior. */
public static final boolean DEFAULT_NULLABLE = false;
@@ -209,4 +255,47 @@ public class CatalogTestUtils {
static CatalogCommand addColumnParams(String tableName, ColumnParams... columns) {
return AlterTableAddColumnCommand.builder().schemaName(DEFAULT_SCHEMA_NAME).tableName(tableName).columns(List.of(columns)).build();
}
+
+ private static class TestUpdateLog implements UpdateLog {
+ private final HybridClock clock;
+
+ private long lastSeenVersion = 0;
+
+ private volatile OnUpdateHandler onUpdateHandler;
+
+ private TestUpdateLog(HybridClock clock) {
+ this.clock = clock;
+ }
+
+ @Override
+ public synchronized CompletableFuture<Boolean> append(VersionedUpdate update) {
+ if (update.version() - 1 != lastSeenVersion) {
+ return completedFuture(false);
+ }
+
+ lastSeenVersion = update.version();
+
+ return onUpdateHandler.handle(update, clock.now(), update.version()).thenApply(ignored -> true);
+ }
+
+ @Override
+ public void registerUpdateHandler(OnUpdateHandler handler) {
+ this.onUpdateHandler = handler;
+ }
+
+ @Override
+ public void start() throws IgniteInternalException {
+ if (onUpdateHandler == null) {
+ throw new IgniteInternalException(
+ Common.INTERNAL_ERR,
+ "Handler must be registered prior to component start"
+ );
+ }
+ }
+
+ @Override
+ public void stop() throws Exception {
+
+ }
+ }
}
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 7be8377d89..368b21c76d 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
@@ -35,6 +35,7 @@ import org.apache.ignite.Ignite;
import org.apache.ignite.internal.sql.engine.ClusterPerClassIntegrationTest;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
+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;
@@ -205,13 +206,13 @@ public class ItCommonApiTest extends ClusterPerClassIntegrationTest {
/** {@inheritDoc} */
@Override
- public @Nullable SchemaPlus schema(@Nullable String name, int version) {
+ public @Nullable SchemaPlus schema(int version) {
return null;
}
/** {@inheritDoc} */
@Override
- public @Nullable SchemaPlus schema(@Nullable String name, long timestamp) {
+ public @Nullable SchemaPlus schema(long timestamp) {
return null;
}
@@ -220,5 +221,11 @@ public class ItCommonApiTest extends ClusterPerClassIntegrationTest {
public CompletableFuture<Void> schemaReadyFuture(int version) {
throw new UnsupportedOperationException();
}
+
+ /** {@inheritDoc} */
+ @Override
+ public @Nullable IgniteTable table(int schemaVersion, int tableId) {
+ return null;
+ }
}
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSystemViewsTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSystemViewsTest.java
index d73782450a..2a8cad2cd4 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSystemViewsTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSystemViewsTest.java
@@ -20,18 +20,9 @@ package org.apache.ignite.internal.sql.engine;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.sql.engine.ItSystemViewsTest.KnownSystemView.SYSTEM_VIEWS;
import static org.apache.ignite.internal.sql.engine.ItSystemViewsTest.KnownSystemView.SYSTEM_VIEW_COLUMNS;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import java.util.List;
import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
import org.apache.ignite.sql.ColumnType;
-import org.apache.ignite.sql.ResultSet;
-import org.apache.ignite.sql.ResultSetMetadata;
-import org.apache.ignite.sql.Session;
-import org.apache.ignite.sql.SqlRow;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
@@ -39,7 +30,6 @@ import org.junit.jupiter.params.provider.EnumSource;
/**
* End-to-end tests to verify system views.
*/
-@SuppressWarnings("DataFlowIssue")
public class ItSystemViewsTest extends ClusterPerClassIntegrationTest {
enum KnownSystemView {
SYSTEM_VIEWS("SYSTEM", "SYSTEM_VIEWS"),
@@ -53,130 +43,116 @@ public class ItSystemViewsTest extends ClusterPerClassIntegrationTest {
this.schema = schema;
this.name = name;
}
+
+ String canonicalName() {
+ return schema + "." + name;
+ }
}
@ParameterizedTest
@EnumSource(KnownSystemView.class)
public void systemViewWithGivenNameExists(KnownSystemView view) {
- try (Session session = CLUSTER_NODES.get(0).sql().sessionBuilder().defaultSchema(view.schema).build()) {
- ResultSet<SqlRow> rs = session.execute(null, format("SELECT count(*) FROM {}", view.name));
-
- // for this test it's enough to check presence of the row,
- // because we are interested in whether the view is available for
- // querying at all
- assertThat(rs.hasNext(), is(true));
- }
+ assertQuery(format("SELECT count(*) FROM {}", view.canonicalName()))
+ // for this test it's enough to check presence of the row,
+ // because we are interested in whether the view is available for
+ // querying at all
+ .returnSomething()
+ .check();
}
@ParameterizedTest
@EnumSource(KnownSystemView.class)
public void systemViewWithGivenNamePresentedInSystemViewsView(KnownSystemView view) {
- try (Session session = CLUSTER_NODES.get(0).sql().sessionBuilder().defaultSchema(SYSTEM_VIEWS.schema).build()) {
- ResultSet<SqlRow> rs = session.execute(null,
- format("SELECT count(*) FROM {} WHERE schema = '{}' AND name = '{}'",
- SYSTEM_VIEWS.name, view.schema, view.name));
-
- assertThat(rs.hasNext(), is(true));
-
- SqlRow row = rs.next();
-
- assertThat(row.value(0), is(1L));
- }
+ assertQuery(format("SELECT * FROM {} WHERE schema = ? AND name = ?", SYSTEM_VIEWS.canonicalName()))
+ .withParams(view.schema, view.name)
+ .returnSomething()
+ .check();
}
@Test
public void systemViewsViewMetadataTest() {
- KnownSystemView view = SYSTEM_VIEWS;
-
- try (Session session = CLUSTER_NODES.get(0).sql().sessionBuilder().defaultSchema(view.schema).build()) {
- ResultSet<SqlRow> rs = session.execute(null, format("SELECT * FROM {}", view.name));
-
- ResultSetMetadata metadata = rs.metadata();
-
- assertMetadata(metadata, List.of(
- new MetadataMatcher()
- .name("ID")
- .type(ColumnType.INT32)
- .nullable(true),
-
- new MetadataMatcher()
- .name("SCHEMA")
- .type(ColumnType.STRING)
- .precision(Short.MAX_VALUE)
- .nullable(true),
-
- new MetadataMatcher()
- .name("NAME")
- .type(ColumnType.STRING)
- .precision(Short.MAX_VALUE)
- .nullable(true),
-
- new MetadataMatcher()
- .name("TYPE")
- .type(ColumnType.STRING)
- .precision(Short.MAX_VALUE)
- .nullable(true)
- ));
- }
+ assertQuery(format("SELECT * FROM {}", SYSTEM_VIEWS.canonicalName()))
+ .columnMetadata(
+ new MetadataMatcher()
+ .name("ID")
+ .type(ColumnType.INT32)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("SCHEMA")
+ .type(ColumnType.STRING)
+ .precision(Short.MAX_VALUE)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("NAME")
+ .type(ColumnType.STRING)
+ .precision(Short.MAX_VALUE)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("TYPE")
+ .type(ColumnType.STRING)
+ .precision(Short.MAX_VALUE)
+ .nullable(true)
+ )
+ .check();
}
@Test
public void systemViewColumnsViewMetadataTest() {
- KnownSystemView view = SYSTEM_VIEW_COLUMNS;
-
- try (Session session = CLUSTER_NODES.get(0).sql().sessionBuilder().defaultSchema(view.schema).build()) {
- ResultSet<SqlRow> rs = session.execute(null, format("SELECT * FROM {}", view.name));
-
- ResultSetMetadata metadata = rs.metadata();
-
- assertMetadata(metadata, List.of(
- new MetadataMatcher()
- .name("VIEW_ID")
- .type(ColumnType.INT32)
- .nullable(true),
-
- new MetadataMatcher()
- .name("NAME")
- .type(ColumnType.STRING)
- .precision(Short.MAX_VALUE)
- .nullable(true),
-
- new MetadataMatcher()
- .name("TYPE")
- .type(ColumnType.STRING)
- .precision(Short.MAX_VALUE)
- .nullable(true),
-
- new MetadataMatcher()
- .name("NULLABLE")
- .type(ColumnType.BOOLEAN)
- .nullable(true),
-
- new MetadataMatcher()
- .name("PRECISION")
- .type(ColumnType.INT32)
- .nullable(true),
-
- new MetadataMatcher()
- .name("SCALE")
- .type(ColumnType.INT32)
- .nullable(true),
-
- new MetadataMatcher()
- .name("LENGTH")
- .type(ColumnType.INT32)
- .nullable(true)
- ));
-
- }
+ assertQuery(format("SELECT * FROM {}", SYSTEM_VIEW_COLUMNS.canonicalName()))
+ .columnMetadata(
+ new MetadataMatcher()
+ .name("VIEW_ID")
+ .type(ColumnType.INT32)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("NAME")
+ .type(ColumnType.STRING)
+ .precision(Short.MAX_VALUE)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("TYPE")
+ .type(ColumnType.STRING)
+ .precision(Short.MAX_VALUE)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("NULLABLE")
+ .type(ColumnType.BOOLEAN)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("PRECISION")
+ .type(ColumnType.INT32)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("SCALE")
+ .type(ColumnType.INT32)
+ .nullable(true),
+
+ new MetadataMatcher()
+ .name("LENGTH")
+ .type(ColumnType.INT32)
+ .nullable(true))
+ .check();
}
- private static void assertMetadata(ResultSetMetadata actual, List<MetadataMatcher> matchers) {
- assertThat(actual, notNullValue());
- assertThat(actual.columns(), hasSize(matchers.size()));
+ @Test
+ public void systemViewAndTableCrossQuery() {
+ KnownSystemView view = SYSTEM_VIEWS;
- for (int i = 0; i < matchers.size(); i++) {
- matchers.get(i).check(actual.columns().get(i));
- }
+ sql("CREATE TABLE known_views(view_id INT primary key, name varchar(256))");
+ sql(format("INSERT INTO known_views SELECT id, SUBSTRING(name, 256) FROM {}", view.canonicalName()));
+
+ assertQuery(format("SELECT * FROM known_views kv JOIN {} sv"
+ + " ON kv.view_id = sv.id WHERE sv.name = ?", view.canonicalName()))
+ .withParam(view.name)
+ .returnSomething()
+ .check();
}
}
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 cacad79fb3..3afc2a48ad 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
@@ -72,10 +72,10 @@ import org.apache.ignite.internal.sql.engine.prepare.PrepareServiceImpl;
import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
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.IgniteSystemView;
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.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;
@@ -88,6 +88,7 @@ import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
+import org.apache.ignite.internal.sql.engine.util.cache.CacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.sql.metrics.SqlClientMetricSource;
import org.apache.ignite.internal.storage.DataStorageManager;
@@ -135,15 +136,17 @@ public class SqlQueryProcessor implements QueryProcessor {
private static final long DEFAULT_SESSION_IDLE_TIMEOUT = TimeUnit.MINUTES.toMillis(15);
/** Name of the default schema. */
- public static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
+ private static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
private static final PropertiesHolder DEFAULT_PROPERTIES = PropertiesHelper.newBuilder()
.set(QueryProperty.DEFAULT_SCHEMA, DEFAULT_SCHEMA_NAME)
.set(SessionProperty.IDLE_TIMEOUT, DEFAULT_SESSION_IDLE_TIMEOUT)
.build();
+ private static final CacheFactory CACHE_FACTORY = CaffeineCacheFactory.INSTANCE;
+
private final ParserService parserService = new ParserServiceImpl(
- PARSED_RESULT_CACHE_SIZE, CaffeineCacheFactory.INSTANCE
+ PARSED_RESULT_CACHE_SIZE, CACHE_FACTORY
);
private final List<LifecycleAware> services = new ArrayList<>();
@@ -220,8 +223,9 @@ public class SqlQueryProcessor implements QueryProcessor {
this.metricManager = metricManager;
this.systemViewManager = systemViewManager;
- sqlSchemaManager = new CatalogSqlSchemaManager(
+ sqlSchemaManager = new SqlSchemaManagerImpl(
catalogManager,
+ CACHE_FACTORY,
SCHEMA_CACHE_SIZE
);
}
@@ -242,6 +246,7 @@ public class SqlQueryProcessor implements QueryProcessor {
var prepareSvc = registerService(PrepareServiceImpl.create(
nodeName,
PLAN_CACHE_SIZE,
+ CACHE_FACTORY,
dataStorageManager,
dataStorageFieldsSupplier.get(),
metricManager
@@ -263,7 +268,9 @@ public class SqlQueryProcessor implements QueryProcessor {
var ddlCommandHandler = new DdlCommandHandler(catalogManager);
- var executableTableRegistry = new ExecutableTableRegistryImpl(tableManager, schemaManager, replicaService, clock, TABLE_CACHE_SIZE);
+ var executableTableRegistry = new ExecutableTableRegistryImpl(
+ tableManager, schemaManager, sqlSchemaManager, replicaService, clock, TABLE_CACHE_SIZE
+ );
var dependencyResolver = new ExecutionDependencyResolverImpl(
executableTableRegistry,
@@ -472,7 +479,7 @@ public class SqlQueryProcessor implements QueryProcessor {
private CompletableFuture<SchemaPlus> waitForActualSchema(String schemaName, HybridTimestamp timestamp) {
try {
return schemaSyncService.waitForMetadataCompleteness(timestamp).thenApply(unused -> {
- SchemaPlus schema = sqlSchemaManager.schema(schemaName, timestamp.longValue());
+ SchemaPlus schema = sqlSchemaManager.schema(timestamp.longValue()).getSubSchema(schemaName);
if (schema == null) {
throw new SchemaNotFoundException(schemaName);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
index d7633ad230..b33c667aed 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistry.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.sql.engine.exec;
import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
/**
* Provides access to read/write APIs for SQL engine.
@@ -28,10 +27,9 @@ public interface ExecutableTableRegistry {
/**
* Returns an table that can be used for both read and write operations.
*
+ * @param schemaVersion Version of the schema.
* @param tableId Table Id.
- * @param tableVersion Table version.
- * @param tableDescriptor Table descriptor.
* @return An operation that returns executable table.
*/
- CompletableFuture<ExecutableTable> getTable(int tableId, int tableVersion, TableDescriptor tableDescriptor);
+ CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
index e8c2c35421..8120ef2ea2 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistryImpl.java
@@ -26,6 +26,8 @@ import org.apache.ignite.internal.replicator.ReplicaService;
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.schema.IgniteTable;
+import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.distributed.TableManager;
@@ -37,6 +39,8 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry {
private final TableManager tableManager;
+ private final SqlSchemaManager sqlSchemaManager;
+
private final SchemaManager schemaManager;
private final ReplicaService replicaService;
@@ -47,9 +51,10 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry {
final ConcurrentMap<CacheKey, CompletableFuture<ExecutableTable>> tableCache;
/** Constructor. */
- public ExecutableTableRegistryImpl(TableManager tableManager, SchemaManager schemaManager,
+ public ExecutableTableRegistryImpl(TableManager tableManager, SchemaManager schemaManager, SqlSchemaManager sqlSchemaManager,
ReplicaService replicaService, HybridClock clock, int cacheSize) {
+ this.sqlSchemaManager = sqlSchemaManager;
this.tableManager = tableManager;
this.schemaManager = schemaManager;
this.replicaService = replicaService;
@@ -61,15 +66,19 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry {
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int tableId, int tableVersion, TableDescriptor tableDescriptor) {
- return tableCache.computeIfAbsent(cacheKey(tableId, tableVersion), (k) -> loadTable(tableId, tableVersion, tableDescriptor));
+ public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
+ IgniteTable sqlTable = sqlSchemaManager.table(schemaVersion, tableId);
+
+ return tableCache.computeIfAbsent(cacheKey(tableId, sqlTable.version()), (k) -> loadTable(sqlTable));
}
- private CompletableFuture<ExecutableTable> loadTable(int tableId, int tableVersion, TableDescriptor tableDescriptor) {
- return tableManager.tableAsync(tableId)
+ private CompletableFuture<ExecutableTable> loadTable(IgniteTable sqlTable) {
+ return tableManager.tableAsync(sqlTable.id())
.thenApply((table) -> {
- SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(tableId);
- SchemaDescriptor schemaDescriptor = schemaRegistry.schema(tableVersion);
+ TableDescriptor tableDescriptor = sqlTable.descriptor();
+
+ SchemaRegistry schemaRegistry = schemaManager.schemaRegistry(sqlTable.id());
+ SchemaDescriptor schemaDescriptor = schemaRegistry.schema(sqlTable.version());
TableRowConverterFactory converterFactory = requiredColumns -> new TableRowConverterImpl(
schemaRegistry, schemaDescriptor, tableDescriptor, requiredColumns
);
@@ -77,7 +86,7 @@ public class ExecutableTableRegistryImpl implements ExecutableTableRegistry {
InternalTable internalTable = table.internalTable();
ScannableTable scannableTable = new ScannableTableImpl(internalTable, converterFactory, tableDescriptor);
- UpdatableTableImpl updatableTable = new UpdatableTableImpl(tableId, tableDescriptor, internalTable.partitions(),
+ UpdatableTableImpl updatableTable = new UpdatableTableImpl(sqlTable.id(), tableDescriptor, internalTable.partitions(),
replicaService, clock, converterFactory.create(null), schemaDescriptor);
return new ExecutableTableImpl(scannableTable, updatableTable);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
index 810ef49859..91c5bdfc83 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
@@ -215,7 +215,7 @@ public class ExecutionContext<RowT> extends AbstractQueryContext implements Data
/** {@inheritDoc} */
@Override
public SchemaPlus getRootSchema() {
- return qctx.schema();
+ throw new AssertionError("getRootSchema");
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
index 2fe2dbf802..c86475a561 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolver.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.sql.engine.exec;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
/**
* Resolves components required for execution.
@@ -30,5 +29,5 @@ public interface ExecutionDependencyResolver {
/**
* Resolves dependencies required to execute the given list of relations.
*/
- CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, IgniteSchema schema);
+ CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int schemaVersion);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
index c1351ca8eb..3297117dc9 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverImpl.java
@@ -32,10 +32,8 @@ import org.apache.ignite.internal.sql.engine.rel.IgniteSystemViewScan;
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.IgniteTrimExchange;
-import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteSystemView;
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.DistributionFunction;
import org.apache.ignite.internal.sql.engine.trait.DistributionFunction.AffinityDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
@@ -61,7 +59,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
* {@inheritDoc}
*/
@Override
- public CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, IgniteSchema schema) {
+ public CompletableFuture<ResolvedDependencies> resolveDependencies(Iterable<IgniteRel> rels, int schemaVersion) {
Map<Integer, CompletableFuture<ExecutableTable>> tableMap = new HashMap<>();
Map<Integer, ScannableDataSource> dataSources = new HashMap<>();
@@ -88,7 +86,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteTableModify rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(igniteTable);
+ resolveTable(schemaVersion, igniteTable.id());
return super.visit(rel);
}
@@ -97,7 +95,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteIndexScan rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(igniteTable);
+ resolveTable(schemaVersion, igniteTable.id());
return rel;
}
@@ -106,7 +104,7 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
public IgniteRel visit(IgniteTableScan rel) {
IgniteTable igniteTable = rel.getTable().unwrapOrThrow(IgniteTable.class);
- resolveTable(igniteTable);
+ resolveTable(schemaVersion, igniteTable.id());
return rel;
}
@@ -128,16 +126,12 @@ public class ExecutionDependencyResolverImpl implements ExecutionDependencyResol
if (function.affinity()) {
int tableId = ((AffinityDistribution) function).tableId();
- resolveTable(schema.getTable(tableId));
+ resolveTable(schemaVersion, tableId);
}
}
- private void resolveTable(IgniteTable igniteTable) {
- int tableId = igniteTable.id();
- TableDescriptor tableDescriptor = igniteTable.descriptor();
- int tableVersion = igniteTable.version();
-
- tableMap.computeIfAbsent(tableId, (id) -> registry.getTable(tableId, tableVersion, tableDescriptor));
+ private void resolveTable(int schemaVersion, int tableId) {
+ tableMap.computeIfAbsent(tableId, (id) -> registry.getTable(schemaVersion, tableId));
}
};
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 ca0be42a80..4ea4bbf74e 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
@@ -77,7 +77,6 @@ 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.IgniteSchema;
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;
@@ -236,13 +235,13 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return queryManager.execute(tx, plan);
}
- private BaseQueryContext createQueryContext(UUID queryId, int schemaVersion, @Nullable String schema, Object[] params) {
+ private BaseQueryContext createQueryContext(UUID queryId, int schemaVersion, Object[] params) {
return BaseQueryContext.builder()
.queryId(queryId)
.parameters(params)
.frameworkConfig(
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(sqlSchemaManager.schema(schema, schemaVersion))
+ .defaultSchema(sqlSchemaManager.schema(schemaVersion))
.build()
)
.logger(LOG)
@@ -423,7 +422,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private void submitFragment(String nodeName, QueryStartRequest msg) {
DistributedQueryManager queryManager = getOrCreateQueryManager(msg);
- queryManager.submitFragment(nodeName, msg.root(), msg.fragmentDescription(), msg.txAttributes());
+ queryManager.submitFragment(nodeName, msg.schemaVersion(), msg.root(), msg.fragmentDescription(), msg.txAttributes());
}
private void handleError(Throwable ex, String nodeName, QueryStartRequest msg) {
@@ -434,7 +433,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private DistributedQueryManager getOrCreateQueryManager(QueryStartRequest msg) {
return queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
- BaseQueryContext ctx = createQueryContext(key, msg.schemaVersion(), msg.schema(), msg.parameters());
+ BaseQueryContext ctx = createQueryContext(key, msg.schemaVersion(), msg.parameters());
return new DistributedQueryManager(ctx);
});
@@ -489,7 +488,6 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
QueryStartRequest request = FACTORY.queryStartRequest()
.queryId(ctx.queryId())
.fragmentId(desc.fragmentId())
- .schema(ctx.schemaName())
.root(serialisedFragment)
.fragmentDescription(desc)
.parameters(ctx.parameters())
@@ -605,6 +603,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private void submitFragment(
String initiatorNode,
+ int schemaVersion,
String fragmentString,
FragmentDescription desc,
TxAttributes txAttributes
@@ -618,11 +617,9 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
start.thenCompose(none -> {
IgniteRel treeRoot = relationalTreeFromJsonString(fragmentString, ctx);
- IgniteSchema igniteSchema = ctx.schema().unwrap(IgniteSchema.class);
- return dependencyResolver.resolveDependencies(List.of(treeRoot), igniteSchema).thenComposeAsync(deps -> {
- return executeFragment(treeRoot, deps, context);
- }, exec);
+ return dependencyResolver.resolveDependencies(List.of(treeRoot), schemaVersion)
+ .thenComposeAsync(deps -> executeFragment(treeRoot, deps, context), exec);
}).exceptionally(ex -> {
handleError(ex, initiatorNode, desc.fragmentId());
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNode.java
index 29c40b724c..908b2e9f32 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNode.java
@@ -20,8 +20,6 @@ package org.apache.ignite.internal.sql.engine.exec.rel;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
-import java.util.Collection;
-import java.util.Comparator;
import java.util.List;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
@@ -40,18 +38,11 @@ public abstract class AbstractNode<RowT> implements Node<RowT> {
protected final int inBufSize = Commons.IN_BUFFER_SIZE;
+ private final ExecutionContext<RowT> ctx;
+
/** For debug purpose. */
private volatile Thread thread;
- /**
- * Execution context.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- *
- * {@link Inbox} node may not have proper context at creation time in case it creates on first message received from a remote source.
- * This case the context sets in scope of {@link Inbox#init(ExecutionContext, Collection, Comparator)} method call.
- */
- private ExecutionContext<RowT> ctx;
-
private Downstream<RowT> downstream;
private boolean closed;
@@ -74,10 +65,6 @@ public abstract class AbstractNode<RowT> implements Node<RowT> {
return ctx;
}
- protected void context(ExecutionContext<RowT> ctx) {
- this.ctx = ctx;
- }
-
/** {@inheritDoc} */
@Override
public void register(List<Node<RowT>> sources) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
index 9f046e8b5a..8255182032 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/QueryStartRequest.java
@@ -27,11 +27,6 @@ import org.apache.ignite.network.annotations.Transferable;
*/
@Transferable(value = SqlQueryMessageGroup.QUERY_START_REQUEST)
public interface QueryStartRequest extends ExecutionContextAwareMessage {
- /**
- * Get schema name.
- */
- String schema();
-
/**
* Get fragment description.
*/
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
index 6bbd0a352c..6ca87e2f56 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
@@ -53,7 +53,7 @@ import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.sql.engine.util.cache.Cache;
-import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
+import org.apache.ignite.internal.sql.engine.util.cache.CacheFactory;
import org.apache.ignite.internal.sql.metrics.SqlPlanCacheMetricSource;
import org.apache.ignite.internal.storage.DataStorageManager;
import org.apache.ignite.internal.thread.NamedThreadFactory;
@@ -101,6 +101,7 @@ public class PrepareServiceImpl implements PrepareService {
*
* @param nodeName Name of the current Ignite node. Will be used in thread factory as part of the thread name.
* @param cacheSize Size of the cache of query plans. Should be non negative.
+ * @param cacheFactory A factory to create cache of query plans.
* @param dataStorageManager Data storage manager.
* @param dataStorageFields Data storage fields. Mapping: Data storage name -> field name -> field type.
* @param metricManager Metric manager.
@@ -108,6 +109,7 @@ public class PrepareServiceImpl implements PrepareService {
public static PrepareServiceImpl create(
String nodeName,
int cacheSize,
+ CacheFactory cacheFactory,
DataStorageManager dataStorageManager,
Map<String, Map<String, Class<?>>> dataStorageFields,
MetricManager metricManager
@@ -115,6 +117,7 @@ public class PrepareServiceImpl implements PrepareService {
return new PrepareServiceImpl(
nodeName,
cacheSize,
+ cacheFactory,
new DdlSqlToCommandConverter(dataStorageFields, DataStorageManager::defaultDataStorage),
DEFAULT_PLANNER_TIMEOUT,
metricManager
@@ -126,6 +129,7 @@ public class PrepareServiceImpl implements PrepareService {
*
* @param nodeName Name of the current Ignite node. Will be used in thread factory as part of the thread name.
* @param cacheSize Size of the cache of query plans. Should be non negative.
+ * @param cacheFactory A factory to create cache of query plans.
* @param ddlConverter A converter of the DDL-related AST to the actual command.
* @param plannerTimeout Timeout in milliseconds to planning.
* @param metricManager Metric manager.
@@ -133,6 +137,7 @@ public class PrepareServiceImpl implements PrepareService {
public PrepareServiceImpl(
String nodeName,
int cacheSize,
+ CacheFactory cacheFactory,
DdlSqlToCommandConverter ddlConverter,
long plannerTimeout,
MetricManager metricManager
@@ -143,7 +148,7 @@ public class PrepareServiceImpl implements PrepareService {
this.metricManager = metricManager;
sqlPlanCacheMetricSource = new SqlPlanCacheMetricSource();
- cache = CaffeineCacheFactory.INSTANCE.create(cacheSize, sqlPlanCacheMetricSource);
+ cache = cacheFactory.create(cacheSize, sqlPlanCacheMetricSource);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
index 8d61c31a37..ff5f521d49 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteSchema.java
@@ -26,6 +26,7 @@ import java.util.stream.Collectors;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
/**
* Schema implementation for sql engine.
@@ -65,9 +66,12 @@ public class IgniteSchema extends AbstractSchema {
}
/** Returns table by given id. */
- public IgniteTable getTable(int tableId) {
+ @Nullable IgniteTable tableByIdOpt(int tableId) {
IgniteDataSource dataSource = tableById.get(tableId);
- assert dataSource instanceof IgniteTable : "Expected table but got " + dataSource;
+
+ if (!(dataSource instanceof IgniteTable)) {
+ return null;
+ }
return (IgniteTable) dataSource;
}
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 50849c5265..2490f88a48 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
@@ -19,21 +19,29 @@ package org.apache.ignite.internal.sql.engine.schema;
import java.util.concurrent.CompletableFuture;
import org.apache.calcite.schema.SchemaPlus;
-import org.jetbrains.annotations.Nullable;
/**
* Sql schemas operations interface.
*/
public interface SqlSchemaManager {
/**
- * Returns schema with given name and by the given version, if name is not specified, returns default schema of the given version.
+ * Returns root schema of the given version.
*/
- SchemaPlus schema(@Nullable String name, int version);
+ SchemaPlus schema(int version);
/**
- * Returns schema with given name and by the given timestamp, if name is not specified, returns default schema of the given version.
+ * Returns root schema by the given timestamp.
*/
- SchemaPlus schema(@Nullable String name, long timestamp);
+ SchemaPlus schema(long timestamp);
+
+ /**
+ * Returns table by given id, which version correspond to the one from schema of given version.
+ *
+ * @param schemaVersion Version of the schema.
+ * @param tableId An identifier of a table of interest.
+ * @return A table.
+ */
+ IgniteTable table(int schemaVersion, int tableId);
/**
* Returns a future to wait for given SQL schema version readiness.
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/SqlSchemaManagerImpl.java
similarity index 71%
rename from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManager.java
rename to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java
index fd8ab94ef0..19f9d5e707 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/SqlSchemaManagerImpl.java
@@ -17,24 +17,23 @@
package org.apache.ignite.internal.sql.engine.schema;
-import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_SCHEMA_NAME;
-
-import com.github.benmanes.caffeine.cache.Caffeine;
import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentMap;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.tools.Frameworks;
+import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.catalog.commands.DefaultValue;
@@ -48,47 +47,49 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescripto
import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor.SystemViewType;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.lang.IgniteInternalException;
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.Nullable;
+import org.apache.ignite.internal.sql.engine.util.cache.Cache;
+import org.apache.ignite.internal.sql.engine.util.cache.CacheFactory;
+import org.apache.ignite.lang.ErrorGroups.Common;
/**
* Implementation of {@link SqlSchemaManager} backed by {@link CatalogService}.
*/
-public class CatalogSqlSchemaManager implements SqlSchemaManager {
+public class SqlSchemaManagerImpl implements SqlSchemaManager {
private final CatalogManager catalogManager;
- private final ConcurrentMap<CacheKey, SchemaPlus> cache;
+ private final Cache<Integer, SchemaPlus> schemaCache;
+
+ private final Cache<Long, IgniteTable> tableCache;
/** Constructor. */
- public CatalogSqlSchemaManager(CatalogManager catalogManager, int cacheSize) {
+ public SqlSchemaManagerImpl(CatalogManager catalogManager, CacheFactory factory, int cacheSize) {
this.catalogManager = catalogManager;
- this.cache = Caffeine.newBuilder()
- .initialCapacity(cacheSize)
- .maximumSize(cacheSize)
- .<CacheKey, SchemaPlus>build()
- .asMap();
+ this.schemaCache = factory.create(cacheSize);
+ this.tableCache = factory.create(cacheSize);
}
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(@Nullable String name, int schemaVersion) {
- String schemaName = name == null ? DEFAULT_SCHEMA_NAME : name;
-
- return cache.computeIfAbsent(cacheKey(schemaVersion, schemaName),
- (e) -> createSqlSchema(e.schemaVersion(), catalogManager.schema(e.schemaName(), e.schemaVersion())));
+ public SchemaPlus schema(int schemaVersion) {
+ return schemaCache.get(
+ schemaVersion,
+ version -> createRootSchema(catalogManager.catalog(version))
+ );
}
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(@Nullable String name, long timestamp) {
+ public SchemaPlus schema(long timestamp) {
int catalogVersion = catalogManager.activeCatalogVersion(timestamp);
- return schema(name, catalogVersion);
+ return schema(catalogVersion);
}
/** {@inheritDoc} */
@@ -98,7 +99,63 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
return catalogManager.catalogReadyFuture(version);
}
- private static SchemaPlus createSqlSchema(int catalogVersion, CatalogSchemaDescriptor schemaDescriptor) {
+ @Override
+ public IgniteTable table(int schemaVersion, int tableId) {
+ return tableCache.get(tableCacheKey(schemaVersion, tableId), key -> {
+ SchemaPlus rootSchema = schemaCache.get(schemaVersion);
+
+ if (rootSchema != null) {
+ for (String name : rootSchema.getSubSchemaNames()) {
+ SchemaPlus subSchema = rootSchema.getSubSchema(name);
+
+ assert subSchema != null : name;
+
+ IgniteSchema schema = subSchema.unwrap(IgniteSchema.class);
+
+ assert schema != null : "unknown schema " + subSchema;
+
+ IgniteTable table = schema.tableByIdOpt(tableId);
+
+ if (table != null) {
+ return table;
+ }
+ }
+ }
+
+ Catalog catalog = catalogManager.catalog(schemaVersion);
+
+ if (catalog == null) {
+ throw new IgniteInternalException(Common.INTERNAL_ERR, "Catalog of given version not found: " + schemaVersion);
+ }
+
+ CatalogTableDescriptor tableDescriptor = catalog.table(tableId);
+
+ if (tableDescriptor == null) {
+ throw new IgniteInternalException(Common.INTERNAL_ERR, "Table with given id not found: " + tableId);
+ }
+
+ return createTable(tableDescriptor, createTableDescriptorForTable(tableDescriptor), Map.of());
+ });
+ }
+
+ private static long tableCacheKey(int schemaVersion, int tableId) {
+ long cacheKey = schemaVersion;
+ cacheKey <<= 32;
+ return cacheKey | tableId;
+ }
+
+ private static SchemaPlus createRootSchema(Catalog catalog) {
+ SchemaPlus rootSchema = Frameworks.createRootSchema(false);
+
+ for (CatalogSchemaDescriptor schemaDescriptor : catalog.schemas()) {
+ IgniteSchema igniteSchema = createSqlSchema(catalog.version(), schemaDescriptor);
+ rootSchema.add(igniteSchema.getName(), igniteSchema);
+ }
+
+ return rootSchema;
+ }
+
+ private static IgniteSchema createSqlSchema(int catalogVersion, CatalogSchemaDescriptor schemaDescriptor) {
String schemaName = schemaDescriptor.name();
int numTables = schemaDescriptor.tables().length;
@@ -131,23 +188,12 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
// Assemble tables.
for (CatalogTableDescriptor tableDescriptor : schemaDescriptor.tables()) {
int tableId = tableDescriptor.id();
- String tableName = tableDescriptor.name();
TableDescriptor descriptor = tableDescriptorMap.get(tableId);
assert descriptor != null;
- //TODO IGNITE-19558: The table is not available at planning stage.
- // Let's fix table statistics keeping in mind IGNITE-19558 issue.
- IgniteStatistic statistic = new IgniteStatistic(() -> 0.0d, descriptor.distribution());
Map<String, IgniteIndex> tableIndexMap = schemaTableIndexes.getOrDefault(tableId, Collections.emptyMap());
- IgniteTable schemaTable = new IgniteTableImpl(
- tableName,
- tableId,
- tableDescriptor.tableVersion(),
- descriptor,
- statistic,
- tableIndexMap
- );
+ IgniteTable schemaTable = createTable(tableDescriptor, descriptor, tableIndexMap);
schemaDataSources.add(schemaTable);
}
@@ -166,10 +212,7 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
schemaDataSources.add(schemaTable);
}
- // create root schema
- SchemaPlus rootSchema = Frameworks.createRootSchema(false);
- IgniteSchema igniteSchema = new IgniteSchema(schemaName, catalogVersion, schemaDataSources);
- return rootSchema.add(schemaName, igniteSchema);
+ return new IgniteSchema(schemaName, catalogVersion, schemaDataSources);
}
private static IgniteIndex createSchemaIndex(CatalogIndexDescriptor indexDescriptor, TableDescriptor tableDescriptor) {
@@ -188,21 +231,23 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
private static TableDescriptor createTableDescriptorForTable(CatalogTableDescriptor descriptor) {
List<ColumnDescriptor> colDescriptors = new ArrayList<>();
- List<Integer> colocationColumns = new ArrayList<>(descriptor.colocationColumns().size());
List<CatalogTableColumnDescriptor> columns = descriptor.columns();
+ Object2IntMap<String> columnToIndex = new Object2IntOpenHashMap<>();
for (int i = 0; i < columns.size(); i++) {
CatalogTableColumnDescriptor col = columns.get(i);
boolean key = descriptor.isPrimaryKeyColumn(col.name());
CatalogColumnDescriptor columnDescriptor = createColumnDescriptor(col, key, i);
- if (descriptor.isColocationColumn(col.name())) {
- colocationColumns.add(i);
- }
+ columnToIndex.put(col.name(), i);
colDescriptors.add(columnDescriptor);
}
+ List<Integer> colocationColumns = descriptor.colocationColumns().stream()
+ .map(columnToIndex::getInt)
+ .collect(Collectors.toList());
+
// TODO Use the actual zone ID after implementing https://issues.apache.org/jira/browse/IGNITE-18426.
int tableId = descriptor.id();
IgniteDistribution distribution = IgniteDistributions.affinity(colocationColumns, tableId, tableId);
@@ -289,42 +334,25 @@ public class CatalogSqlSchemaManager implements SqlSchemaManager {
return columnDescriptor;
}
- private static CacheKey cacheKey(int schemaVersion, String schemaName) {
- return new CacheKey(schemaVersion, schemaName);
- }
-
- private static class CacheKey {
- private final int schemaVersion;
- private final String schemaName;
-
- CacheKey(int schemaVersion, String schemaName) {
- this.schemaVersion = schemaVersion;
- this.schemaName = schemaName;
- }
-
- public int schemaVersion() {
- return schemaVersion;
- }
-
- public String schemaName() {
- return schemaName;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- CacheKey cacheKey = (CacheKey) o;
- return schemaVersion == cacheKey.schemaVersion && Objects.equals(schemaName, cacheKey.schemaName);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(schemaVersion, schemaName);
- }
+ private static IgniteTable createTable(
+ CatalogTableDescriptor catalogTableDescriptor,
+ TableDescriptor tableDescriptor,
+ Map<String, IgniteIndex> indexes
+ ) {
+ int tableId = catalogTableDescriptor.id();
+ String tableName = catalogTableDescriptor.name();
+
+ //TODO IGNITE-19558: The table is not available at planning stage.
+ // Let's fix table statistics keeping in mind IGNITE-19558 issue.
+ IgniteStatistic statistic = new IgniteStatistic(() -> 0.0d, tableDescriptor.distribution());
+
+ return new IgniteTableImpl(
+ tableName,
+ tableId,
+ catalogTableDescriptor.tableVersion(),
+ tableDescriptor,
+ statistic,
+ indexes
+ );
}
}
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 7c2d517c85..bf5eb82144 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
@@ -210,10 +210,6 @@ public final class BaseQueryContext extends AbstractQueryContext {
return log;
}
- public String schemaName() {
- return schema().getName();
- }
-
public SchemaPlus schema() {
return cfg.getDefaultSchema();
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
index 585c4127fe..8541349890 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
@@ -28,7 +28,7 @@ import org.jetbrains.annotations.Nullable;
* Factory that creates caches backed by {@link Caffeine} cache.
*/
public class CaffeineCacheFactory implements CacheFactory {
- public static CacheFactory INSTANCE = new CaffeineCacheFactory();
+ public static final CacheFactory INSTANCE = new CaffeineCacheFactory();
private CaffeineCacheFactory() {
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
index cb04a37fb8..411dae61dd 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutableTableRegistrySelfTest.java
@@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.when;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.TestHybridClock;
import org.apache.ignite.internal.hlc.HybridClock;
@@ -30,6 +31,10 @@ import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaRegistry;
+import org.apache.ignite.internal.sql.engine.framework.TestStatistic;
+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.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableImpl;
@@ -59,6 +64,9 @@ public class ExecutableTableRegistrySelfTest extends BaseIgniteAbstractTest {
@Mock
private SchemaManager schemaManager;
+ @Mock
+ private SqlSchemaManager sqlSchemaManager;
+
@Mock
private TableDescriptor descriptor;
@@ -121,7 +129,7 @@ public class ExecutableTableRegistrySelfTest extends BaseIgniteAbstractTest {
}
Tester(int cacheSize) {
- registry = new ExecutableTableRegistryImpl(tableManager, schemaManager, replicaService, clock, cacheSize);
+ registry = new ExecutableTableRegistryImpl(tableManager, schemaManager, sqlSchemaManager, replicaService, clock, cacheSize);
}
CompletableFuture<ExecutableTable> getTable(int tableId) {
@@ -133,10 +141,16 @@ public class ExecutableTableRegistrySelfTest extends BaseIgniteAbstractTest {
when(tableManager.tableAsync(tableId)).thenReturn(CompletableFuture.completedFuture(table));
when(schemaManager.schemaRegistry(tableId)).thenReturn(schemaRegistry);
- when(schemaRegistry.schema(tableVersion)).thenReturn(schemaDescriptor);
when(descriptor.iterator()).thenReturn(emptyIterator());
+ when(schemaRegistry.schema(tableVersion)).thenReturn(schemaDescriptor);
+
+ IgniteTable sqlTable = new IgniteTableImpl(
+ table.name(), tableId, tableVersion, descriptor, new TestStatistic(1_000.0), Map.of()
+ );
+
+ when(sqlSchemaManager.table(schemaVersion, tableId)).thenReturn(sqlTable);
- return registry.getTable(tableId, tableVersion, descriptor);
+ return registry.getTable(schemaVersion, tableId);
}
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
index 39887d7962..a7625b1c29 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionDependencyResolverSelfTest.java
@@ -20,10 +20,8 @@ package org.apache.ignite.internal.sql.engine.exec;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -40,7 +38,6 @@ import org.apache.ignite.internal.sql.engine.framework.TestTable;
import org.apache.ignite.internal.sql.engine.planner.AbstractPlannerTest;
import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
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.TableDescriptor;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.type.NativeTypes;
@@ -82,8 +79,6 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
int t1Id = testTable1.id();
int t2Id = testTable2.id();
- TableDescriptor td1 = testTable1.descriptor();
- TableDescriptor td2 = testTable2.descriptor();
Tester tester = new Tester(createSchema(testTable1, testTable2));
@@ -96,8 +91,8 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
tester.checkDependencies(deps, t1Id);
tester.checkDependencies(deps, t2Id);
- verify(registry, times(1)).getTable(eq(t1Id), anyInt(), same(td1));
- verify(registry, times(1)).getTable(eq(t2Id), anyInt(), same(td2));
+ verify(registry, times(1)).getTable(anyInt(), eq(t1Id));
+ verify(registry, times(1)).getTable(anyInt(), eq(t2Id));
}
/**
@@ -108,7 +103,6 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
TestTable table = createTestTable("TEST1", addHashIndex("ID"));
int t1Id = table.id();
- TableDescriptor td1 = table.descriptor();
Tester tester = new Tester(createSchema(table));
tester.setDependencies(t1Id, table1, update1);
@@ -116,7 +110,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
CompletableFuture<ResolvedDependencies> f = tester.resolveDependencies("SELECT * FROM test1 WHERE id=1");
tester.checkDependencies(f.join(), t1Id);
- verify(registry, times(1)).getTable(eq(t1Id), anyInt(), same(td1));
+ verify(registry, times(1)).getTable(anyInt(), eq(t1Id));
}
/**
@@ -129,8 +123,6 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
int t1Id = testTable1.id();
int t2Id = testTable2.id();
- TableDescriptor td1 = testTable1.descriptor();
- TableDescriptor td2 = testTable2.descriptor();
Tester tester = new Tester(createSchema(testTable1, testTable2));
@@ -144,8 +136,8 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
tester.checkDependencies(deps, t1Id);
tester.checkDependencies(deps, t2Id);
- verify(registry, times(1)).getTable(eq(t1Id), anyInt(), same(td1));
- verify(registry, times(1)).getTable(eq(t2Id), anyInt(), same(td2));
+ verify(registry, times(1)).getTable(anyInt(), eq(t1Id));
+ verify(registry, times(1)).getTable(anyInt(), eq(t2Id));
}
/**
@@ -166,7 +158,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
ResolvedDependencies deps = f.join();
tester.checkDependencies(deps, t1Id);
- verify(registry, times(1)).getTable(anyInt(), anyInt(), any(TableDescriptor.class));
+ verify(registry, times(1)).getTable(anyInt(), anyInt());
}
/**
@@ -205,14 +197,14 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
CompletableFuture<ExecutableTable> f = CompletableFuture.completedFuture(executableTable);
- when(registry.getTable(eq(tableId), anyInt(), any(TableDescriptor.class))).thenReturn(f);
+ when(registry.getTable(anyInt(), eq(tableId))).thenReturn(f);
}
void setError(int tableId, Throwable err) {
CompletableFuture<ExecutableTable> f = new CompletableFuture<>();
f.completeExceptionally(err);
- when(registry.getTable(eq(tableId), anyInt(), any(TableDescriptor.class))).thenReturn(f);
+ when(registry.getTable(anyInt(), eq(tableId))).thenReturn(f);
}
CompletableFuture<ResolvedDependencies> resolveDependencies(String sql) {
@@ -225,7 +217,7 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
throw new IllegalStateException("Unable to plan: " + sql, e);
}
- return resolver.resolveDependencies(List.of(rel), igniteSchema);
+ return resolver.resolveDependencies(List.of(rel), igniteSchema.version());
}
void checkDependencies(ResolvedDependencies dependencies, int tableId) {
@@ -234,12 +226,6 @@ public class ExecutionDependencyResolverSelfTest extends AbstractPlannerTest {
assertEquals(executableTable.scannableTable(), dependencies.scannableTable(tableId));
assertEquals(executableTable.updatableTable(), dependencies.updatableTable(tableId));
}
-
- TableDescriptor tableDescriptor(String tableName) {
- IgniteTable table = (IgniteTable) igniteSchema.getTable(tableName);
- return table.descriptor();
- }
-
}
private static final class TestExecutableTable implements ExecutableTable {
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 fb1524a9b3..741f6d8578 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
@@ -38,7 +38,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.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;
@@ -118,6 +117,7 @@ import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.HashFunctionFactory;
import org.apache.ignite.internal.sql.engine.util.HashFunctionFactoryImpl;
+import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.IgniteTestUtils.RunnableX;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -184,7 +184,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
public void init() {
testCluster = new TestCluster();
executionServices = nodeNames.stream().map(this::create).collect(Collectors.toList());
- prepareService = new PrepareServiceImpl("test", 0, null, PLANNING_TIMEOUT, new MetricManager());
+ prepareService = new PrepareServiceImpl("test", 0, CaffeineCacheFactory.INSTANCE, null, PLANNING_TIMEOUT, new MetricManager());
parserService = new ParserServiceImpl(0, EmptyCacheFactory.INSTANCE);
prepareService.start();
@@ -737,7 +737,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
rootSch.add(schema.getName(), schema);
SchemaPlus plus = rootSch.plus().getSubSchema(schema.getName());
- when(schemaManagerMock.schema(any(), anyInt())).thenReturn(plus);
+ when(schemaManagerMock.schema(anyInt())).thenReturn(plus);
var targetProvider = new ExecutionTargetProvider() {
@Override
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
index cf82bba832..0639502f5a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
@@ -24,7 +24,7 @@ import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
public final class NoOpExecutableTableRegistry implements ExecutableTableRegistry {
/** {@inheritDoc} */
@Override
- public CompletableFuture<ExecutableTable> getTable(int tableId, int tableVersion, TableDescriptor tableDescriptor) {
+ public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
return CompletableFuture.completedFuture(new NoOpExecutableTable(tableId));
}
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 0628cd50a2..2be85c6860 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
@@ -31,7 +31,6 @@ import org.apache.calcite.tools.Frameworks;
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;
-import org.jetbrains.annotations.Nullable;
/**
* Dummy wrapper for predefined collection of schemas.
@@ -70,14 +69,14 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(@Nullable String name, int version) {
- return name == null ? root : root.getSubSchema(name);
+ public SchemaPlus schema(int version) {
+ return root;
}
/** {@inheritDoc} */
@Override
- public SchemaPlus schema(@Nullable String name, long timestamp) {
- return name == null ? root : root.getSubSchema(name);
+ public SchemaPlus schema(long timestamp) {
+ return root;
}
/** {@inheritDoc} */
@@ -85,4 +84,15 @@ public class PredefinedSchemaManager implements SqlSchemaManager {
public CompletableFuture<Void> schemaReadyFuture(int version) {
return completedFuture(null);
}
+
+ @Override
+ public IgniteTable table(int schemaVersion, int tableId) {
+ IgniteTable table = tableById.get(tableId);
+
+ if (table == null) {
+ throw new RuntimeException("Table not found: " + tableId);
+ }
+
+ return table;
+ }
}
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 2621a0e35d..29359eaebc 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
@@ -64,6 +64,7 @@ import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.HashFunctionFactoryImpl;
+import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.util.AsyncCursor;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.IgniteUtils;
@@ -99,7 +100,8 @@ public class TestNode implements LifecycleAware {
MappingService mappingService
) {
this.nodeName = nodeName;
- var ps = new PrepareServiceImpl(nodeName, 0, mock(DdlSqlToCommandConverter.class), PLANNING_TIMEOUT, mock(MetricManager.class));
+ var ps = new PrepareServiceImpl(nodeName, 0, CaffeineCacheFactory.INSTANCE,
+ mock(DdlSqlToCommandConverter.class), PLANNING_TIMEOUT, mock(MetricManager.class));
this.prepareService = registerService(ps);
this.schemaManager = schemaManager;
@@ -117,7 +119,9 @@ public class TestNode implements LifecycleAware {
mailboxRegistry, messageService
));
NoOpExecutableTableRegistry executableTableRegistry = new NoOpExecutableTableRegistry();
- ExecutionDependencyResolver dependencyResolver = new ExecutionDependencyResolverImpl(executableTableRegistry, null);
+ ExecutionDependencyResolver dependencyResolver = new ExecutionDependencyResolverImpl(
+ executableTableRegistry, null
+ );
executionService = registerService(new ExecutionServiceImpl<>(
messageService,
@@ -222,7 +226,7 @@ public class TestNode implements LifecycleAware {
.cancel(new QueryCancel())
.frameworkConfig(
Frameworks.newConfigBuilder(FRAMEWORK_CONFIG)
- .defaultSchema(schemaManager.schema(DEFAULT_SCHEMA_NAME, Long.MAX_VALUE))
+ .defaultSchema(schemaManager.schema(Long.MAX_VALUE).getSubSchema(DEFAULT_SCHEMA_NAME))
.build()
)
.build();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
index 248b818bf2..9d08c0daa3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
@@ -49,6 +49,7 @@ import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.SqlTestUtils;
+import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.type.NativeTypes;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@@ -64,7 +65,8 @@ public class PlannerTimeoutTest extends AbstractPlannerTest {
IgniteSchema schema = createSchema(createTestTable("T1"));
BaseQueryContext ctx = baseQueryContext(Collections.singletonList(schema), null);
- PrepareService prepareService = new PrepareServiceImpl("test", 0, null, plannerTimeout, new MetricManager());
+ PrepareService prepareService = new PrepareServiceImpl("test", 0,
+ CaffeineCacheFactory.INSTANCE, null, plannerTimeout, new MetricManager());
prepareService.start();
try {
ParserService parserService = new ParserServiceImpl(0, EmptyCacheFactory.INSTANCE);
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
deleted file mode 100644
index c5582a2e69..0000000000
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/CatalogSqlSchemaManagerTest.java
+++ /dev/null
@@ -1,756 +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 org.apache.ignite.internal.catalog.CatalogManagerImpl.INITIAL_CAUSALITY_TOKEN;
-import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_SCHEMA_NAME;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsString;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-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.schema.SchemaPlus;
-import org.apache.calcite.schema.Table;
-import org.apache.ignite.internal.catalog.CatalogManager;
-import org.apache.ignite.internal.catalog.commands.DefaultValue;
-import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
-import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexColumnDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogSortedIndexDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor.SystemViewType;
-import org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
-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.apache.ignite.internal.sql.engine.util.TypeUtils;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.type.NativeType;
-import org.apache.ignite.sql.ColumnType;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-
-/**
- * Tests for {@link CatalogSqlSchemaManager}.
- */
-@ExtendWith(MockitoExtension.class)
-public class CatalogSqlSchemaManagerTest extends BaseIgniteAbstractTest {
-
- private static final AtomicInteger ID = new AtomicInteger();
-
- @Mock
- private CatalogManager catalogManager;
-
- /** Basic schema with several tables. */
- @Test
- public void testBasicSchema() {
- TestSchema testSchema = new TestSchema("TEST");
- testSchema.version = 2000;
- testSchema.timestamp = 111111L;
-
- TestTable t1 = new TestTable("T1");
- t1.addColumn("c1", ColumnType.INT32);
-
- TestTable t2 = new TestTable("T2");
- t2.addColumn("c1", ColumnType.STRING);
-
- testSchema.tables.add(t1);
- testSchema.tables.add(t2);
-
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- assertEquals(testSchema.name, schema.getName());
- assertEquals(testSchema.version, schema.version());
-
- assertNotNull(schema.getTable(t1.name));
- assertNotNull(schema.getTable(t2.name));
- }
-
- /**
- * Table column types.
- */
- @ParameterizedTest
- @MethodSource("columnTypes")
- public void testTableColumns(ColumnType columnType, int precision, int scale, boolean hasNativeType) {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1_nullable", columnType, precision, scale);
- testTable.addColumn("c1_not_nullable", columnType, precision, scale);
-
- testTable.notNull("c1_not_nullable");
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = getTable(schema, testTable);
-
- assertEquals(testTable.id, table.id());
-
- TableDescriptor descriptor = table.descriptor();
- assertEquals(testTable.columns.size(), descriptor.columnsCount(), "column count");
-
- checkColumns(columnType, precision, scale, hasNativeType, descriptor);
- }
-
- private static void checkColumns(ColumnType columnType, int precision, int scale, boolean hasNativeType, TableDescriptor descriptor) {
- ColumnDescriptor c1 = descriptor.columnDescriptor(0);
- assertEquals(0, c1.logicalIndex());
- assertTrue(c1.nullable());
-
- ColumnDescriptor c2 = descriptor.columnDescriptor(1);
- assertEquals(1, c2.logicalIndex());
- assertFalse(c2.nullable());
-
- if (hasNativeType) {
- NativeType nativeType = c1.physicalType();
- NativeType expectedNativeType = TypeUtils.columnType2NativeType(columnType, precision, scale, precision);
-
- assertEquals(expectedNativeType, nativeType);
- } else {
- IllegalArgumentException t = assertThrows(IllegalArgumentException.class, c1::physicalType);
- assertThat(t.getMessage(), containsString("No NativeType for type"));
- }
- }
-
- /** Empty default schema. */
- @Test
- public void testDefaultSchema() {
- TestSchema testSchema = new TestSchema();
-
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
-
- {
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- assertEquals(DEFAULT_SCHEMA_NAME, schema.getName());
- assertEquals(testSchema.version, schema.version());
- }
-
- {
- SchemaPlus schemaPlus = sqlSchemaManager.schema(null, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- assertEquals(DEFAULT_SCHEMA_NAME, schema.getName());
- assertEquals(testSchema.version, schema.version());
- }
- }
-
- /**
- * Column default value constraint.
- */
- @Test
- public void testTableDefaultValue() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT32);
- testTable.addColumn("c2", ColumnType.INT32);
- testTable.addColumn("c3", ColumnType.INT32);
- testTable.addColumn("c4", ColumnType.STRING);
-
- testTable.defaultValueMap.put("c2", DefaultValue.constant(null));
- testTable.defaultValueMap.put("c3", DefaultValue.constant(1));
- testTable.defaultValueMap.put("c4", DefaultValue.functionCall(DefaultValueGenerator.GEN_RANDOM_UUID.name()));
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = getTable(schema, testTable);
-
- assertEquals(testTable.id, table.id());
-
- ColumnDescriptor c1 = table.descriptor().columnDescriptor("c1");
- assertNull(c1.defaultValue());
- assertEquals(DefaultValueStrategy.DEFAULT_NULL, c1.defaultStrategy());
-
- ColumnDescriptor c2 = table.descriptor().columnDescriptor("c2");
- assertNull(c2.defaultValue());
- assertEquals(DefaultValueStrategy.DEFAULT_NULL, c2.defaultStrategy());
-
- ColumnDescriptor c3 = table.descriptor().columnDescriptor("c3");
- assertEquals(1, c3.defaultValue());
- assertEquals(DefaultValueStrategy.DEFAULT_CONSTANT, c3.defaultStrategy());
-
- ColumnDescriptor c4 = table.descriptor().columnDescriptor("c4");
- assertNotNull(c4.defaultValue());
- assertEquals(DefaultValueStrategy.DEFAULT_COMPUTED, c4.defaultStrategy());
- }
-
- /**
- * Table with neither primary key nor colocation key columns are legal.
- */
- @Test
- public void testTableWithoutPrimaryKeyColoKey() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT8);
- testTable.addColumn("c2", ColumnType.INT16);
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
-
- assertFalse(table.descriptor().columnDescriptor(0).key());
- assertFalse(table.descriptor().columnDescriptor(1).key());
- }
-
- /**
- * Table with primary key but w/o colocation key columns are legal.
- */
- @Test
- public void testTableWithPrimaryKeyWithoutColoKey() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT8);
- testTable.addColumn("c2", ColumnType.INT16);
-
- testTable.primaryKey("c1");
- testTable.notNull("c1");
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
-
- assertTrue(table.descriptor().columnDescriptor(0).key());
- assertFalse(table.descriptor().columnDescriptor(0).nullable());
- assertFalse(table.descriptor().columnDescriptor(1).key());
- }
-
- /**
- * Table distribution.
- */
- @Test
- public void testTableDistribution() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT8);
- testTable.addColumn("c2", ColumnType.INT16);
- testTable.addColumn("c3", ColumnType.INT32);
- testTable.addColumn("c4", ColumnType.INT64);
- testTable.addColumn("c5", ColumnType.FLOAT);
- testTable.addColumn("c6", ColumnType.DOUBLE);
-
- testTable.primaryKey("c2", "c3", "c4");
- testTable.notNull("c2", "c3", "c4");
- testTable.colocationKey("c2", "c3");
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
-
- IgniteDistribution distribution = table.descriptor().distribution();
- // TODO Use the actual zone ID after implementing https://issues.apache.org/jira/browse/IGNITE-18426.
- int tableId = table.id();
- assertEquals(IgniteDistributions.affinity(List.of(1, 2), tableId, tableId), distribution);
- assertEquals(distribution, table.getStatistic().getDistribution());
- }
-
- /**
- * Hash index.
- */
- @Test
- public void testHashIndex() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT32);
- testTable.addColumn("c2", ColumnType.INT32);
- testTable.addColumn("c3", ColumnType.INT32);
-
- TestIndex testIndex = new TestIndex("TEST_IDX");
- testIndex.table = testTable.name;
- testIndex.hashColumns = Arrays.asList("c1", "c2");
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.indexes.add(testIndex);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
-
- IgniteIndex testIdx = table.indexes().get(testIndex.name);
-
- assertEquals(testIndex.name, testIdx.name());
- assertEquals(Type.HASH, testIdx.type());
- assertEquals(RelCollations.of(
- new RelFieldCollation(0, Direction.CLUSTERED, NullDirection.UNSPECIFIED),
- new RelFieldCollation(1, Direction.CLUSTERED, NullDirection.UNSPECIFIED)
- ), testIdx.collation());
- }
-
- /**
- * Sorted index.
- */
- @Test
- public void testSortedIndex() {
- TestTable testTable = new TestTable("TEST");
-
- testTable.addColumn("c1", ColumnType.INT32);
- testTable.addColumn("c2", ColumnType.INT32);
- testTable.addColumn("c3", ColumnType.INT32);
- testTable.addColumn("c4", ColumnType.INT32);
- testTable.addColumn("c5", ColumnType.INT32);
-
- TestIndex testIndex = new TestIndex("TEST_IDX");
- testIndex.table = testTable.name;
- testIndex.sortedColumns = Arrays.asList(
- Map.entry("c1", CatalogColumnCollation.ASC_NULLS_LAST),
- Map.entry("c2", CatalogColumnCollation.ASC_NULLS_FIRST),
- Map.entry("c3", CatalogColumnCollation.DESC_NULLS_LAST),
- Map.entry("c4", CatalogColumnCollation.DESC_NULLS_FIRST)
- );
-
- TestSchema testSchema = new TestSchema();
- testSchema.tables.add(testTable);
- testSchema.indexes.add(testIndex);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema schema = unwrapSchema(schemaPlus);
-
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
-
- IgniteIndex testIdx = table.indexes().get(testIndex.name);
-
- assertEquals(testIndex.name, testIdx.name());
- assertEquals(Type.SORTED, testIdx.type());
- assertEquals(RelCollations.of(
- new RelFieldCollation(0, Direction.ASCENDING, NullDirection.LAST),
- new RelFieldCollation(1, Direction.ASCENDING, NullDirection.FIRST),
- new RelFieldCollation(2, Direction.DESCENDING, NullDirection.LAST),
- new RelFieldCollation(3, Direction.DESCENDING, NullDirection.FIRST)
- ), testIdx.collation());
- }
-
- /**
- * Tests basic properties of a system view.
- */
- @ParameterizedTest
- @MethodSource("systemViewDistributions")
- public void testBasicView(SystemViewType viewType, IgniteDistribution distribution) {
- TestSystemView testSystemView = new TestSystemView("TEST", viewType);
- testSystemView.addColumn("c1", ColumnType.STRING);
-
- TestSchema testSchema = new TestSchema();
- testSchema.systemViews.add(testSystemView);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema igniteSchema = unwrapSchema(schemaPlus);
- IgniteSystemView systemView = getSystemView(igniteSchema, testSystemView);
-
- assertEquals(testSystemView.id, systemView.id());
- assertEquals(distribution, systemView.distribution());
- assertEquals(distribution, systemView.getStatistic().getDistribution());
- }
-
- /**
- * Tests system view column types.
- */
- @ParameterizedTest
- @MethodSource("systemViewColumnTypes")
- public void testViewColumns(SystemViewType viewType, ColumnType columnType, int precision, int scale, boolean hasNativeType) {
- TestSystemView testSystemView = new TestSystemView("TEST", viewType);
-
- testSystemView.addColumn("c1_nullable", columnType, precision, scale);
- testSystemView.addColumn("c1_not_nullable", columnType, precision, scale);
-
- testSystemView.notNull("c1_not_nullable");
-
- TestSchema testSchema = new TestSchema();
- testSchema.systemViews.add(testSystemView);
- testSchema.init(catalogManager);
-
- SqlSchemaManager sqlSchemaManager = newSchemaManager();
- SchemaPlus schemaPlus = sqlSchemaManager.schema(testSchema.name, testSchema.timestamp);
- IgniteSchema igniteSchema = unwrapSchema(schemaPlus);
- IgniteSystemView systemView = getSystemView(igniteSchema, testSystemView);
-
- TableDescriptor descriptor = systemView.descriptor();
- assertEquals(testSystemView.columns.size(), descriptor.columnsCount(), "column count");
- }
-
- private static Stream<Arguments> systemViewColumnTypes() {
- List<Arguments> allArgs = new ArrayList<>();
-
- for (SystemViewType type : SystemViewType.values()) {
- columnTypes().map(args -> {
- Object[] vals = args.get();
-
- Object[] newVals = new Object[vals.length + 1];
- newVals[0] = type;
-
- System.arraycopy(vals, 0, newVals, 1, vals.length);
-
- return Arguments.of(newVals);
- }).forEach(allArgs::add);
- }
-
- return allArgs.stream();
- }
-
- private static Stream<Arguments> systemViewDistributions() {
- return Stream.of(
- Arguments.of(SystemViewType.NODE, IgniteDistributions.identity(0)),
- Arguments.of(SystemViewType.CLUSTER, IgniteDistributions.single())
- );
- }
-
-
- private static IgniteSystemView getSystemView(IgniteSchema schema, TestSystemView testSystemView) {
- Table systemViewTable = schema.getTable(testSystemView.name);
- assertNotNull(systemViewTable);
-
- IgniteSystemView systemView = assertInstanceOf(IgniteSystemView.class, systemViewTable);
- assertEquals(systemView.name(), testSystemView.name);
-
- return systemView;
- }
-
- private CatalogSqlSchemaManager newSchemaManager() {
- return new CatalogSqlSchemaManager(catalogManager, 200);
- }
-
- private IgniteSchema unwrapSchema(SchemaPlus schemaPlus) {
- IgniteSchema igniteSchema = schemaPlus.unwrap(IgniteSchema.class);
- assertNotNull(igniteSchema);
- return igniteSchema;
- }
-
- private static IgniteTable getTable(IgniteSchema schema, TestTable testTable) {
- IgniteTable table = (IgniteTable) schema.getTable(testTable.name);
- assertNotNull(table);
- return table;
- }
-
- private static Stream<Arguments> columnTypes() {
- return Stream.of(
- Arguments.of(ColumnType.BOOLEAN, -1, -1, true),
- Arguments.of(ColumnType.INT8, -1, -1, true),
- Arguments.of(ColumnType.INT16, -1, -1, true),
- Arguments.of(ColumnType.INT32, -1, -1, true),
- Arguments.of(ColumnType.INT64, -1, -1, true),
- Arguments.of(ColumnType.FLOAT, -1, -1, true),
- Arguments.of(ColumnType.DOUBLE, -1, -1, true),
- Arguments.of(ColumnType.DECIMAL, 4, -1, true),
- Arguments.of(ColumnType.DECIMAL, 4, 2, true),
- Arguments.of(ColumnType.NUMBER, 4, -1, true),
- Arguments.of(ColumnType.STRING, 40, -1, true),
- Arguments.of(ColumnType.BYTE_ARRAY, 40, -1, true),
- Arguments.of(ColumnType.DATE, -1, -1, true),
- Arguments.of(ColumnType.TIME, 2, -1, true),
- Arguments.of(ColumnType.DATETIME, 2, -1, true),
- Arguments.of(ColumnType.TIMESTAMP, 2, -1, true),
- Arguments.of(ColumnType.PERIOD, 2, -1, false),
- Arguments.of(ColumnType.DURATION, 2, -1, false),
- Arguments.of(ColumnType.UUID, 2, -1, true),
- Arguments.of(ColumnType.BITMASK, 2, -1, true)
- );
- }
-
- private static final class TestSchema {
-
- final String name;
-
- final Set<TestTable> tables = new LinkedHashSet<>();
-
- final Set<TestIndex> indexes = new LinkedHashSet<>();
-
- final Set<TestSystemView> systemViews = new LinkedHashSet<>();
-
- TestSchema() {
- this(DEFAULT_SCHEMA_NAME);
- }
-
- TestSchema(String name) {
- this.name = name;
- }
-
- long timestamp = System.nanoTime();
-
- int version = 1111;
-
- void init(CatalogManager catalogManager) {
- CatalogSchemaDescriptor schemaDescriptor = newSchemaDescriptor(version);
- when(catalogManager.activeCatalogVersion(timestamp)).thenReturn(version);
- when(catalogManager.schema(name != null ? name : DEFAULT_SCHEMA_NAME, version)).thenReturn(schemaDescriptor);
- }
-
- CatalogSchemaDescriptor newSchemaDescriptor(int version) {
- LinkedHashMap<String, CatalogTableDescriptor> tableDescriptors = new LinkedHashMap<>();
-
- for (TestTable testTable : tables) {
- testTable.version = version;
-
- CatalogTableDescriptor descriptor = testTable.newDescriptor();
- tableDescriptors.put(testTable.name, descriptor);
- }
-
- LinkedHashMap<String, CatalogIndexDescriptor> indexDescriptorMap = new LinkedHashMap<>();
-
- for (TestIndex testIndex : indexes) {
- CatalogTableDescriptor tableDescriptor = tableDescriptors.get(testIndex.table);
- testIndex.version = version;
-
- int tableId = tableDescriptor.id();
- String name = testIndex.name;
-
- indexDescriptorMap.put(name, testIndex.newDescriptor(tableId));
- }
-
- LinkedHashMap<String, CatalogSystemViewDescriptor> systemViewDescriptorMap = new LinkedHashMap<>();
-
- for (TestSystemView testSystemView : systemViews) {
- CatalogSystemViewDescriptor descriptor = testSystemView.newDescriptor(testSystemView.id);
-
- systemViewDescriptorMap.put(testSystemView.name, descriptor);
- }
-
- CatalogTableDescriptor[] tablesArray = tableDescriptors.values().toArray(new CatalogTableDescriptor[0]);
- CatalogIndexDescriptor[] indexesArray = indexDescriptorMap.values().toArray(new CatalogIndexDescriptor[0]);
- CatalogSystemViewDescriptor[] systemViewsArray = systemViewDescriptorMap.values().toArray(new CatalogSystemViewDescriptor[0]);
-
- return new CatalogSchemaDescriptor(
- ID.incrementAndGet(),
- name,
- tablesArray,
- indexesArray,
- systemViewsArray,
- INITIAL_CAUSALITY_TOKEN
- );
- }
- }
-
- private abstract static class TestDataSource {
-
- final List<CatalogTableColumnDescriptor> columns = new ArrayList<>();
-
- int id = ID.incrementAndGet();
-
- final String name;
-
- final Set<String> notNull = new HashSet<>();
-
- final Map<String, DefaultValue> defaultValueMap = new HashMap<>();
-
- TestDataSource(String name) {
- this.name = name;
- }
-
- void addColumn(String name, ColumnType columnType) {
- DefaultValue defaultValue = DefaultValue.constant(null);
- columns.add(new CatalogTableColumnDescriptor(name, columnType, true, 0, 0, 0, defaultValue));
- setDefault(name, defaultValue);
- }
-
- void addColumn(String name, ColumnType columnType, int precision, int scale) {
- DefaultValue defaultValue = DefaultValue.constant(null);
- columns.add(new CatalogTableColumnDescriptor(name, columnType, true, precision, scale, precision, defaultValue));
- setDefault(name, defaultValue);
- }
-
- void setDefault(String name, DefaultValue defaultValue) {
- defaultValueMap.put(name, defaultValue);
- }
-
- // Adds NOT NULL constraint, Set nullable = false for the given columns.
- void notNull(String... names) {
- notNull.clear();
- notNull.addAll(Arrays.asList(names));
- }
- }
-
- private static final class TestTable extends TestDataSource {
-
- private final int zoneId = ID.incrementAndGet();
-
- private List<String> primaryKey = Collections.emptyList();
-
- private List<String> colocationKey;
-
- private int version;
-
- private TestTable(String name) {
- super(name);
- }
-
- // Sets primary key columns
- void primaryKey(String... names) {
- primaryKey = Arrays.asList(names);
- }
-
- // Set colocation key columns
- void colocationKey(String... names) {
- colocationKey = Arrays.asList(names);
- }
-
- CatalogTableDescriptor newDescriptor() {
- List<CatalogTableColumnDescriptor> columnDescriptors = new ArrayList<>();
-
- for (CatalogTableColumnDescriptor col : columns) {
- String colName = col.name();
- DefaultValue defaultValue = defaultValueMap.get(colName);
- boolean nullable = !notNull.contains(colName);
- int precision = col.precision();
- int scale = col.scale();
- int length = col.length();
-
- CatalogTableColumnDescriptor newCol = new CatalogTableColumnDescriptor(colName, col.type(), nullable,
- precision, scale, length, defaultValue);
- columnDescriptors.add(newCol);
- }
-
- return new CatalogTableDescriptor(
- id,
- -1,
- -1,
- name,
- zoneId,
- CatalogTableDescriptor.INITIAL_TABLE_VERSION,
- columnDescriptors,
- primaryKey,
- colocationKey,
- INITIAL_CAUSALITY_TOKEN
- );
- }
- }
-
- private static final class TestIndex {
-
- private final String name;
-
- private int id = ID.incrementAndGet();
-
- private int version;
-
- private List<String> hashColumns;
-
- private List<Map.Entry<String, CatalogColumnCollation>> sortedColumns;
-
- private String table;
-
- private TestIndex(String name) {
- this.name = name;
- }
-
- CatalogIndexDescriptor newDescriptor(int tableId) {
- if (hashColumns != null) {
- return new CatalogHashIndexDescriptor(id, name, tableId, false, hashColumns, true);
- } else if (sortedColumns != null) {
- List<CatalogIndexColumnDescriptor> indexColumns = sortedColumns.stream()
- .map((e) -> new CatalogIndexColumnDescriptor(e.getKey(), e.getValue()))
- .collect(Collectors.toList());
-
- return new CatalogSortedIndexDescriptor(id, name, tableId, false, indexColumns, true);
- } else {
- throw new IllegalStateException("Unable to create index");
- }
- }
- }
-
- private static class TestSystemView extends TestDataSource {
-
- final SystemViewType systemViewType;
-
- private TestSystemView(String name, SystemViewType systemViewType) {
- super(name);
- this.systemViewType = systemViewType;
- }
-
- CatalogSystemViewDescriptor newDescriptor(int id) {
- return new CatalogSystemViewDescriptor(id, name, List.copyOf(columns), systemViewType);
- }
- }
-}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java
new file mode 100644
index 0000000000..10078b4e99
--- /dev/null
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImplTest.java
@@ -0,0 +1,708 @@
+/*
+ * 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 org.apache.ignite.internal.sql.engine.util.TypeUtils.columnType2NativeType;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Stream;
+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.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.ignite.internal.catalog.CatalogCommand;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.CatalogTestUtils;
+import org.apache.ignite.internal.catalog.commands.ColumnParams;
+import org.apache.ignite.internal.catalog.commands.CreateHashIndexCommand;
+import org.apache.ignite.internal.catalog.commands.CreateSortedIndexCommand;
+import org.apache.ignite.internal.catalog.commands.CreateSystemViewCommand;
+import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
+import org.apache.ignite.internal.catalog.commands.CreateTableCommandBuilder;
+import org.apache.ignite.internal.catalog.commands.DefaultValue;
+import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor.SystemViewType;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableColumnDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.lang.IgniteInternalException;
+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.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.sql.ColumnType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+
+/**
+ * Tests for {@link SqlSchemaManagerImpl}.
+ */
+@ExtendWith(MockitoExtension.class)
+public class SqlSchemaManagerImplTest extends BaseIgniteAbstractTest {
+ private static final String PUBLIC_SCHEMA_NAME = "PUBLIC";
+ private static final String SYSTEM_SCHEMA_NAME = "SYSTEM";
+
+ private CatalogManager catalogManager;
+ private SqlSchemaManagerImpl sqlSchemaManager;
+
+ @BeforeEach
+ void init() {
+ catalogManager = CatalogTestUtils.createCatalogManagerWithTestUpdateLog("test", new HybridClockImpl());
+ sqlSchemaManager = new SqlSchemaManagerImpl(catalogManager, CaffeineCacheFactory.INSTANCE, 200);
+
+ catalogManager.start();
+ }
+
+ @AfterEach
+ void cleanup() throws Exception {
+ catalogManager.stop();
+ }
+
+ @Test
+ void exceptionWillBeThrownIfTableNotFoundById() {
+ await(catalogManager.execute(List.of(
+ createDummyTable("T1")
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+
+ // just to fill up cache
+ sqlSchemaManager.schema(versionAfter);
+
+ int nonExistingTableId = Integer.MAX_VALUE;
+
+ assertThat(catalogManager.table(nonExistingTableId, versionAfter), nullValue());
+
+ assertThrowsWithCause(
+ () -> sqlSchemaManager.table(versionAfter, nonExistingTableId),
+ IgniteInternalException.class,
+ "Table with given id not found"
+ );
+ }
+
+ @Test
+ void testMultipleSchemas() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ createDummyTable("T1"),
+ createDummySystemView("V1", SystemViewType.NODE)
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ assertNotNull(rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME));
+ assertNotNull(rootSchema.getSubSchema(SYSTEM_SCHEMA_NAME));
+ }
+
+ /** Basic schema with several tables. */
+ @Test
+ public void testBasicSchema() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ createDummyTable("T1"),
+ createDummyTable("T2")
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteSchema schema = unwrapSchema(schemaPlus);
+
+ assertThat(schema.version(), equalTo(versionAfter));
+
+ CatalogSchemaDescriptor schemaDescriptor = catalogManager.schema(PUBLIC_SCHEMA_NAME, versionAfter);
+
+ assertThat(schemaDescriptor, notNullValue());
+ assertThat(schema.getName(), equalTo(schemaDescriptor.name()));
+
+ for (CatalogTableDescriptor tableDescriptor : schemaDescriptor.tables()) {
+ assertThat(schema.getTable(tableDescriptor.name()), notNullValue());
+ }
+ }
+
+ /**
+ * Table column types.
+ */
+ @ParameterizedTest
+ @MethodSource("columnTypes")
+ public void testTableColumns(ColumnType columnType, int precision, int scale) {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ CreateTableCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName("TEST")
+ .columns(List.of(
+ ColumnParams.builder().name("ID").type(ColumnType.INT32).nullable(false).build(),
+ column("VAL_NULLABLE", columnType, precision, scale, true),
+ column("VAL_NOT_NULLABLE", columnType, precision, scale, false)
+ ))
+ .primaryKeyColumns(List.of("ID"))
+ .zone("Default")
+ .build()
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "TEST");
+
+ CatalogTableDescriptor tableDescriptor = catalogManager.table(table.id(), versionAfter);
+
+ assertThat(tableDescriptor, notNullValue());
+
+ TableDescriptor descriptor = table.descriptor();
+ assertEquals(tableDescriptor.columns().size(), descriptor.columnsCount(), "column count");
+
+ for (int i = 0; i < descriptor.columnsCount(); i++) {
+ CatalogTableColumnDescriptor expectedColumnDescriptor = tableDescriptor.columns().get(i);
+ ColumnDescriptor actualColumnDescriptor = descriptor.columnDescriptor(i);
+
+ checkColumns(actualColumnDescriptor, i, expectedColumnDescriptor);
+ }
+ }
+
+ private static void checkColumns(
+ ColumnDescriptor actualColumnDescriptor,
+ int expectedIndex,
+ CatalogTableColumnDescriptor expectedColumnDescriptor
+ ) {
+ assertThat(actualColumnDescriptor.logicalIndex(), equalTo(expectedIndex));
+ assertThat(actualColumnDescriptor.name(), equalTo(expectedColumnDescriptor.name()));
+ assertThat(actualColumnDescriptor.nullable(), equalTo(expectedColumnDescriptor.nullable()));
+ assertThat(
+ actualColumnDescriptor.physicalType(),
+ equalTo(columnType2NativeType(
+ expectedColumnDescriptor.type(),
+ expectedColumnDescriptor.precision(),
+ expectedColumnDescriptor.scale(),
+ expectedColumnDescriptor.length()
+ ))
+ );
+ }
+
+ /**
+ * Column default value constraint.
+ */
+ @Test
+ public void testTableDefaultValue() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ CreateTableCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName("TEST")
+ .columns(List.of(
+ ColumnParams.builder().name("C1").type(ColumnType.INT32).nullable(false).build(),
+ ColumnParams.builder().name("C2").type(ColumnType.INT32)
+ .defaultValue(DefaultValue.constant(null)).build(),
+ ColumnParams.builder().name("C3").type(ColumnType.INT32)
+ .defaultValue(DefaultValue.constant(1)).build(),
+ ColumnParams.builder().name("C4").type(ColumnType.STRING).length(256)
+ .defaultValue(DefaultValue.functionCall(DefaultValueGenerator.GEN_RANDOM_UUID.name())).build()
+ ))
+ .primaryKeyColumns(List.of("C1"))
+ .zone("Default")
+ .build()
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "TEST");
+
+ ColumnDescriptor c1 = table.descriptor().columnDescriptor("C1");
+ assertNull(c1.defaultValue());
+ assertEquals(DefaultValueStrategy.DEFAULT_NULL, c1.defaultStrategy());
+
+ ColumnDescriptor c2 = table.descriptor().columnDescriptor("C2");
+ assertNull(c2.defaultValue());
+ assertEquals(DefaultValueStrategy.DEFAULT_NULL, c2.defaultStrategy());
+
+ ColumnDescriptor c3 = table.descriptor().columnDescriptor("C3");
+ assertEquals(1, c3.defaultValue());
+ assertEquals(DefaultValueStrategy.DEFAULT_CONSTANT, c3.defaultStrategy());
+
+ ColumnDescriptor c4 = table.descriptor().columnDescriptor("C4");
+ assertNotNull(c4.defaultValue());
+ assertEquals(DefaultValueStrategy.DEFAULT_COMPUTED, c4.defaultStrategy());
+ }
+
+ /**
+ * Table with primary key but w/o colocation key columns are legal.
+ */
+ @Test
+ public void testTableWithPrimaryKey() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ CreateTableCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName("TEST")
+ .columns(List.of(
+ ColumnParams.builder().name("C1").type(ColumnType.INT32).nullable(false).build(),
+ ColumnParams.builder().name("C2").type(ColumnType.INT32)
+ .defaultValue(DefaultValue.constant(null)).build(),
+ ColumnParams.builder().name("C3").type(ColumnType.INT32)
+ .defaultValue(DefaultValue.constant(1)).build(),
+ ColumnParams.builder().name("C4").type(ColumnType.STRING).length(256)
+ .defaultValue(DefaultValue.functionCall(DefaultValueGenerator.GEN_RANDOM_UUID.name())).build()
+ ))
+ .primaryKeyColumns(List.of("C1", "C3"))
+ .zone("Default")
+ .build()
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "TEST");
+
+ assertTrue(table.descriptor().columnDescriptor(0).key());
+ assertFalse(table.descriptor().columnDescriptor(1).key());
+ assertTrue(table.descriptor().columnDescriptor(2).key());
+ assertFalse(table.descriptor().columnDescriptor(3).key());
+ }
+
+ /**
+ * Table distribution.
+ */
+ @Test
+ public void testTableDistribution() {
+ Function<CreateTableCommandBuilder, CreateTableCommandBuilder> tableBase = builder -> builder
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .columns(List.of(
+ ColumnParams.builder().name("C1").type(ColumnType.INT32).build(),
+ ColumnParams.builder().name("C2").type(ColumnType.INT32).build(),
+ ColumnParams.builder().name("C3").type(ColumnType.INT32).build(),
+ ColumnParams.builder().name("C4").type(ColumnType.INT32).build()
+ ))
+ .primaryKeyColumns(List.of("C1", "C2", "C3", "C4"))
+ .zone("Default");
+
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ tableBase.apply(CreateTableCommand.builder())
+ .tableName("T1")
+ .colocationColumns(List.of("C2"))
+ .build(),
+ tableBase.apply(CreateTableCommand.builder())
+ .tableName("T2")
+ .colocationColumns(List.of("C4", "C2"))
+ .build(),
+ tableBase.apply(CreateTableCommand.builder())
+ .tableName("T3")
+ .colocationColumns(List.of("C3", "C2", "C1"))
+ .build()
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ {
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "T1");
+ IgniteDistribution distribution = table.descriptor().distribution();
+
+ assertThat(distribution, equalTo(IgniteDistributions.affinity(List.of(1), table.id(), table.id())));
+ }
+
+ {
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "T2");
+ IgniteDistribution distribution = table.descriptor().distribution();
+
+ assertThat(distribution, equalTo(IgniteDistributions.affinity(List.of(3, 1), table.id(), table.id())));
+ }
+
+ {
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "T3");
+ IgniteDistribution distribution = table.descriptor().distribution();
+
+ assertThat(distribution, equalTo(IgniteDistributions.affinity(List.of(2, 1, 0), table.id(), table.id())));
+ }
+ }
+
+ @Test
+ public void testHashIndex() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ createDummyTable("T1"),
+ createHashIndex("T1", "VAL1_IDX", "VAL1")
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "T1");
+ assertNotNull(table);
+
+ IgniteIndex index = table.indexes().get("VAL1_IDX");
+
+ assertThat(index.name(), equalTo("VAL1_IDX"));
+ assertThat(index.type(), equalTo(Type.HASH));
+ assertThat(index.collation(), equalTo(RelCollations.of(
+ new RelFieldCollation(1, Direction.CLUSTERED, NullDirection.UNSPECIFIED)
+ )));
+ }
+
+ @Test
+ public void testSortedIndex() {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ createDummyTable("T1"),
+ createSortedIndex("T1", "IDX1", List.of("VAL1", "VAL2"),
+ List.of(CatalogColumnCollation.ASC_NULLS_FIRST, CatalogColumnCollation.ASC_NULLS_LAST)),
+ createSortedIndex("T1", "IDX2", List.of("VAL1", "VAL2"),
+ List.of(CatalogColumnCollation.DESC_NULLS_FIRST, CatalogColumnCollation.DESC_NULLS_LAST))
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(PUBLIC_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ IgniteTable table = getTable(unwrapSchema(schemaPlus), "T1");
+ assertNotNull(table);
+
+ {
+ IgniteIndex index = table.indexes().get("IDX1");
+
+ assertThat(index.name(), equalTo("IDX1"));
+ assertThat(index.type(), equalTo(Type.SORTED));
+ assertThat(index.collation(), equalTo(RelCollations.of(
+ new RelFieldCollation(1, Direction.ASCENDING, NullDirection.FIRST),
+ new RelFieldCollation(2, Direction.ASCENDING, NullDirection.LAST)
+ )));
+ }
+
+ {
+ IgniteIndex index = table.indexes().get("IDX2");
+
+ assertThat(index.name(), equalTo("IDX2"));
+ assertThat(index.type(), equalTo(Type.SORTED));
+ assertThat(index.collation(), equalTo(RelCollations.of(
+ new RelFieldCollation(1, Direction.DESCENDING, NullDirection.FIRST),
+ new RelFieldCollation(2, Direction.DESCENDING, NullDirection.LAST)
+ )));
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("systemViewDistributions")
+ public void testBasicView(SystemViewType viewType, IgniteDistribution distribution) {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ createDummySystemView("V1", SystemViewType.NODE),
+ createDummySystemView("V2", SystemViewType.CLUSTER)
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(SYSTEM_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ {
+ IgniteSystemView systemView = getSystemView(unwrapSchema(schemaPlus), "V1");
+
+ assertThat(systemView.name(), equalTo("V1"));
+ assertThat(systemView.distribution(), equalTo(IgniteDistributions.identity(0)));
+ }
+
+ {
+ IgniteSystemView systemView = getSystemView(unwrapSchema(schemaPlus), "V2");
+
+ assertThat(systemView.name(), equalTo("V2"));
+ assertThat(systemView.distribution(), equalTo(IgniteDistributions.single()));
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("systemViewColumnTypes")
+ public void testViewColumns(SystemViewType viewType, ColumnType columnType, int precision, int scale) {
+ int versionBefore = catalogManager.latestCatalogVersion();
+ await(catalogManager.execute(List.of(
+ CreateSystemViewCommand.builder()
+ .name("V1")
+ .columns(List.of(
+ ColumnParams.builder().name("ID").type(ColumnType.INT32).nullable(false).build(),
+ column("VAL_NULLABLE", columnType, precision, scale, true),
+ column("VAL_NOT_NULLABLE", columnType, precision, scale, false)
+ ))
+ .type(SystemViewType.CLUSTER)
+ .build()
+ )));
+
+ int versionAfter = catalogManager.latestCatalogVersion();
+ assertThat(versionAfter, equalTo(versionBefore + 1));
+
+ SchemaPlus rootSchema = sqlSchemaManager.schema(versionAfter);
+ assertNotNull(rootSchema);
+
+ SchemaPlus schemaPlus = rootSchema.getSubSchema(SYSTEM_SCHEMA_NAME);
+ assertNotNull(schemaPlus);
+
+ CatalogSchemaDescriptor schemaDescriptor = catalogManager.schema(SYSTEM_SCHEMA_NAME, versionAfter);
+ assertNotNull(schemaDescriptor);
+
+ CatalogSystemViewDescriptor viewDescriptor = schemaDescriptor.systemView("V1");
+ assertNotNull(viewDescriptor);
+
+ IgniteSystemView systemView = getSystemView(unwrapSchema(schemaPlus), "V1");
+
+ TableDescriptor descriptor = systemView.descriptor();
+ assertEquals(viewDescriptor.columns().size(), descriptor.columnsCount(), "column count");
+
+ for (int i = 0; i < descriptor.columnsCount(); i++) {
+ CatalogTableColumnDescriptor expectedColumnDescriptor = viewDescriptor.columns().get(i);
+ ColumnDescriptor actualColumnDescriptor = descriptor.columnDescriptor(i);
+
+ checkColumns(actualColumnDescriptor, i, expectedColumnDescriptor);
+ }
+ }
+
+ private static Stream<Arguments> systemViewColumnTypes() {
+ List<Arguments> allArgs = new ArrayList<>();
+
+ for (SystemViewType type : SystemViewType.values()) {
+ columnTypes().map(args -> {
+ Object[] vals = args.get();
+
+ Object[] newVals = new Object[vals.length + 1];
+ newVals[0] = type;
+
+ System.arraycopy(vals, 0, newVals, 1, vals.length);
+
+ return Arguments.of(newVals);
+ }).forEach(allArgs::add);
+ }
+
+ return allArgs.stream();
+ }
+
+ private static Stream<Arguments> systemViewDistributions() {
+ return Stream.of(
+ Arguments.of(SystemViewType.NODE, IgniteDistributions.identity(0)),
+ Arguments.of(SystemViewType.CLUSTER, IgniteDistributions.single())
+ );
+ }
+
+
+ private static IgniteSystemView getSystemView(IgniteSchema schema, String name) {
+ Table systemViewTable = schema.getTable(name);
+ assertNotNull(systemViewTable);
+
+ IgniteSystemView systemView = assertInstanceOf(IgniteSystemView.class, systemViewTable);
+ assertEquals(systemView.name(), name);
+
+ return systemView;
+ }
+
+ private static IgniteSchema unwrapSchema(SchemaPlus schemaPlus) {
+ IgniteSchema igniteSchema = schemaPlus.unwrap(IgniteSchema.class);
+ assertNotNull(igniteSchema);
+ return igniteSchema;
+ }
+
+ private static IgniteTable getTable(IgniteSchema schema, String name) {
+ IgniteTable table = (IgniteTable) schema.getTable(name);
+ assertNotNull(table);
+ return table;
+ }
+
+ private static Stream<Arguments> columnTypes() {
+ return Stream.of(
+ Arguments.of(ColumnType.BOOLEAN, -1, -1),
+ Arguments.of(ColumnType.INT8, -1, -1),
+ Arguments.of(ColumnType.INT16, -1, -1),
+ Arguments.of(ColumnType.INT32, -1, -1),
+ Arguments.of(ColumnType.INT64, -1, -1),
+ Arguments.of(ColumnType.FLOAT, -1, -1),
+ Arguments.of(ColumnType.DOUBLE, -1, -1),
+ Arguments.of(ColumnType.DECIMAL, 4, 0),
+ Arguments.of(ColumnType.DECIMAL, 4, 2),
+ Arguments.of(ColumnType.NUMBER, 4, -1),
+ Arguments.of(ColumnType.STRING, 40, -1),
+ Arguments.of(ColumnType.BYTE_ARRAY, 40, -1),
+ Arguments.of(ColumnType.DATE, -1, -1),
+ Arguments.of(ColumnType.TIME, 2, -1),
+ Arguments.of(ColumnType.DATETIME, 2, -1),
+ Arguments.of(ColumnType.TIMESTAMP, 2, -1),
+ Arguments.of(ColumnType.UUID, -1, -1),
+ Arguments.of(ColumnType.BITMASK, 2, -1)
+ );
+ }
+
+ private static CatalogCommand createDummyTable(String name) {
+ return CreateTableCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName(name)
+ .columns(List.of(
+ ColumnParams.builder().name("ID").type(ColumnType.INT32).nullable(false).build(),
+ ColumnParams.builder().name("VAL1").type(ColumnType.INT32).nullable(false).build(),
+ ColumnParams.builder().name("VAL2").type(ColumnType.INT32).nullable(false).build()
+ ))
+ .primaryKeyColumns(List.of("ID"))
+ .zone("Default")
+ .build();
+ }
+
+ private static CatalogCommand createDummySystemView(String name, SystemViewType type) {
+ return CreateSystemViewCommand.builder()
+ .name(name)
+ .columns(List.of(
+ ColumnParams.builder().name("NODE").type(ColumnType.STRING).length(256).build(),
+ ColumnParams.builder().name("C1").type(ColumnType.INT32).nullable(false).build(),
+ ColumnParams.builder().name("C2").type(ColumnType.INT32).nullable(false).build()
+ ))
+ .type(type)
+ .build();
+ }
+
+ private static CatalogCommand createHashIndex(String tableName, String indexName, String... cols) {
+ return CreateHashIndexCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName(tableName)
+ .indexName(indexName)
+ .unique(false)
+ .columns(Arrays.asList(cols))
+ .build();
+ }
+
+ private static CatalogCommand createSortedIndex(String tableName, String indexName,
+ List<String> cols, List<CatalogColumnCollation> collations) {
+ return CreateSortedIndexCommand.builder()
+ .schemaName(PUBLIC_SCHEMA_NAME)
+ .tableName(tableName)
+ .indexName(indexName)
+ .unique(false)
+ .columns(cols)
+ .collations(collations)
+ .build();
+ }
+
+ private static ColumnParams column(String name, ColumnType columnType, int precision, int scale, boolean nullable) {
+ ColumnParams.Builder builder = ColumnParams.builder()
+ .name(name)
+ .type(columnType)
+ .nullable(nullable);
+
+ switch (columnType) {
+ case INT8:
+ case INT16:
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case DATE:
+ case UUID:
+ case BOOLEAN:
+ break;
+ case NUMBER:
+ case TIME:
+ case DATETIME:
+ case TIMESTAMP:
+ builder.precision(precision);
+ break;
+ case DECIMAL:
+ builder.precision(precision);
+ builder.scale(scale);
+ break;
+ case STRING:
+ case BYTE_ARRAY:
+ case BITMASK:
+ builder.length(precision);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported native type: " + columnType);
+ }
+
+ return builder.build();
+ }
+}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
index e3ec0af37a..163ed1a531 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/metrics/PlanningCacheMetricsTest.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.util.BaseQueryContext;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
+import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.type.NativeTypes;
import org.junit.jupiter.api.Test;
@@ -47,7 +48,8 @@ public class PlanningCacheMetricsTest extends AbstractPlannerTest {
@Test
public void plannerCacheStatisticsTest() throws Exception {
MetricManager metricManager = new MetricManager();
- PrepareService prepareService = new PrepareServiceImpl("test", 2, null, DEFAULT_PLANNER_TIMEOUT, metricManager);
+ PrepareService prepareService = new PrepareServiceImpl("test", 2, CaffeineCacheFactory.INSTANCE,
+ null, DEFAULT_PLANNER_TIMEOUT, metricManager);
prepareService.start();
diff --git a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
index 44adbece25..5fb1965bbf 100644
--- a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
+++ b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryChecker.java
@@ -299,6 +299,8 @@ public interface QueryChecker {
QueryChecker returnNothing();
+ QueryChecker returnSomething();
+
QueryChecker columnNames(String... columns);
QueryChecker columnTypes(Type... columns);
diff --git a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
index ea49ce9f1e..84746901cd 100644
--- a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
+++ b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
@@ -22,6 +22,8 @@ import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.util.ArrayUtils.OBJECT_EMPTY_ARRAY;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.not;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.fail;
@@ -36,6 +38,8 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
import org.apache.ignite.internal.sql.engine.QueryContext;
import org.apache.ignite.internal.sql.engine.QueryProcessor;
@@ -55,14 +59,15 @@ import org.hamcrest.Matcher;
* Query checker base class.
*/
abstract class QueryCheckerImpl implements QueryChecker {
+ private static final IgniteLogger LOG = Loggers.forClass(QueryCheckerImpl.class);
- final QueryTemplate queryTemplate;
+ private final QueryTemplate queryTemplate;
private final ArrayList<Matcher<String>> planMatchers = new ArrayList<>();
private final ArrayList<String> disabledRules = new ArrayList<>();
- private List<List<?>> expectedResult;
+ private ResultChecker resultChecker;
private List<ColumnMatcher> metadataMatchers;
@@ -146,11 +151,13 @@ abstract class QueryCheckerImpl implements QueryChecker {
*/
@Override
public QueryChecker returns(Object... res) {
- assert expectedResult != QueryChecker.EMPTY_RES
- : "Erroneous awaiting results mixing, impossible to simultaneously wait something and nothing";
+ assert resultChecker == null || resultChecker instanceof RowByRowResultChecker
+ : "Result checker already set to " + resultChecker.getClass().getSimpleName();
- if (expectedResult == null) {
- expectedResult = new ArrayList<>();
+ RowByRowResultChecker rowByRowResultChecker = (RowByRowResultChecker) resultChecker;
+
+ if (rowByRowResultChecker == null) {
+ rowByRowResultChecker = new RowByRowResultChecker();
}
// let's interpret null array as simple single null.
@@ -158,7 +165,7 @@ abstract class QueryCheckerImpl implements QueryChecker {
res = QueryChecker.NULL_AS_VARARG;
}
- expectedResult.add(Arrays.asList(res));
+ rowByRowResultChecker.expectedResult.add(Arrays.asList(res));
return this;
}
@@ -170,9 +177,18 @@ abstract class QueryCheckerImpl implements QueryChecker {
*/
@Override
public QueryChecker returnNothing() {
- assert expectedResult == null : "Erroneous awaiting results mixing, impossible to simultaneously wait nothing and something";
+ assert resultChecker == null : "Result checker already set to " + resultChecker.getClass().getSimpleName();
+
+ resultChecker = new EmptyResultChecker();
+
+ return this;
+ }
+
+ @Override
+ public QueryChecker returnSomething() {
+ assert resultChecker == null : "Result checker already set to " + resultChecker.getClass().getSimpleName();
- expectedResult = QueryChecker.EMPTY_RES;
+ resultChecker = new NotEmptyResultChecker();
return this;
}
@@ -252,6 +268,8 @@ abstract class QueryCheckerImpl implements QueryChecker {
String qry = queryTemplate.createQuery();
+ LOG.info("Executing query: {}", qry);
+
try {
if (!CollectionUtils.nullOrEmpty(planMatchers)) {
@@ -293,22 +311,10 @@ abstract class QueryCheckerImpl implements QueryChecker {
assertEquals(metadataMatchers.size(), columnMetadata.size(), "Column metadata doesn't match");
}
- var res = getAllFromCursor(cur);
+ List<List<?>> res = Commons.cast(getAllFromCursor(cur));
- if (expectedResult != null) {
- if (Objects.equals(expectedResult, QueryChecker.EMPTY_RES)) {
- assertEquals(0, res.size(), "Empty result expected");
-
- return;
- }
-
- if (!ordered) {
- // Avoid arbitrary order.
- res.sort(new ListComparator());
- expectedResult.sort(new ListComparator());
- }
-
- QueryChecker.assertEqualsCollections(expectedResult, res);
+ if (resultChecker != null) {
+ resultChecker.check(res, ordered);
}
} finally {
await(qryProc.closeSession(sessionId));
@@ -414,4 +420,38 @@ abstract class QueryCheckerImpl implements QueryChecker {
}
}
}
+
+ @FunctionalInterface
+ interface ResultChecker {
+ void check(List<List<?>> rows, boolean ordered);
+ }
+
+ private static class EmptyResultChecker implements ResultChecker {
+ @Override
+ public void check(List<List<?>> rows, boolean ordered) {
+ assertThat(rows, empty());
+ }
+ }
+
+ private static class NotEmptyResultChecker implements ResultChecker {
+ @Override
+ public void check(List<List<?>> rows, boolean ordered) {
+ assertThat(rows, not(empty()));
+ }
+ }
+
+ private static class RowByRowResultChecker implements ResultChecker {
+ private final List<List<?>> expectedResult = new ArrayList<>();
+
+ @Override
+ public void check(List<List<?>> rows, boolean ordered) {
+ if (!ordered) {
+ // Avoid arbitrary order.
+ rows.sort(new ListComparator());
+ expectedResult.sort(new ListComparator());
+ }
+
+ QueryChecker.assertEqualsCollections(expectedResult, rows);
+ }
+ }
}