IGNITE-20439 Sql. Support multiple schemas in CatalogSqlSchemaManager

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java

@@ -296,7 +296,8 @@ public CompletableFuture<Void> catalogReadyFuture(int version) {
         return versionTracker.waitFor(version);
     }
 
-    private Catalog catalog(int version) {
+    @Override
+    public Catalog catalog(int version) {
         return catalogByVer.get(version);
     }
 

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);

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));
+            }
+        };
+    }
+}

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 void stop() throws Exception {
         };
     }
 
+    /**
+     * 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 @@ static CatalogCommand dropColumnParams(String tableName, String... columns) {
     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 {
+
+        }
+    }
 }

modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItCommonApiTest.java

@@ -35,6 +35,7 @@
 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 @@ private static class ErroneousSchemaManager implements SqlSchemaManager {
 
         /** {@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 @@ private static class ErroneousSchemaManager implements SqlSchemaManager {
         public CompletableFuture<Void> schemaReadyFuture(int version) {
             throw new UnsupportedOperationException();
         }
+
+        /** {@inheritDoc} */
+        @Override
+        public @Nullable IgniteTable table(int schemaVersion, int tableId) {
+            return null;
+        }
     }
 }