IGNITE-19082: Catalog. Cleanup dead code

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

@@ -47,6 +47,7 @@
 import java.util.concurrent.Flow.Publisher;
 import java.util.function.LongSupplier;
 import org.apache.ignite.internal.catalog.commands.AlterZoneSetDefaultCatalogCommand;
+import org.apache.ignite.internal.catalog.commands.CreateSchemaCommand;
 import org.apache.ignite.internal.catalog.commands.CreateZoneCommand;
 import org.apache.ignite.internal.catalog.commands.StorageProfileParams;
 import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
@@ -119,6 +120,9 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
     /** Versioned catalog descriptors sorted in chronological order. */
     private final NavigableMap<Long, Catalog> catalogByTs = new ConcurrentSkipListMap<>();
 
+    /** A future that completes when an empty catalog is initialised. If catalog is not empty this future when this completes starts. */
+    private final CompletableFuture<Void> catalogInitializationFuture = new CompletableFuture<>();
+
     private final UpdateLog updateLog;
 
     private final PendingComparableValuesTracker<Integer, Void> versionTracker = new PendingComparableValuesTracker<>(0);
@@ -158,27 +162,7 @@ public CatalogManagerImpl(
     public CompletableFuture<Void> startAsync() {
         int objectIdGen = 0;
 
-        // TODO: IGNITE-19082 Move default schema objects initialization to cluster init procedure.
-        CatalogSchemaDescriptor publicSchema = new CatalogSchemaDescriptor(
-                objectIdGen++,
-                DEFAULT_SCHEMA_NAME,
-                new CatalogTableDescriptor[0],
-                new CatalogIndexDescriptor[0],
-                new CatalogSystemViewDescriptor[0],
-                INITIAL_CAUSALITY_TOKEN
-        );
-
-        // TODO: IGNITE-19082 Move system schema objects initialization to cluster init procedure.
-        CatalogSchemaDescriptor systemSchema = new CatalogSchemaDescriptor(
-                objectIdGen++,
-                SYSTEM_SCHEMA_NAME,
-                new CatalogTableDescriptor[0],
-                new CatalogIndexDescriptor[0],
-                new CatalogSystemViewDescriptor[0],
-                INITIAL_CAUSALITY_TOKEN
-        );
-
-        Catalog emptyCatalog = new Catalog(0, 0L, objectIdGen, List.of(), List.of(publicSchema, systemSchema), null);
+        Catalog emptyCatalog = new Catalog(0, 0L, objectIdGen, List.of(), List.of(), null);
 
         registerCatalog(emptyCatalog);
 
@@ -187,12 +171,17 @@ public CompletableFuture<Void> startAsync() {
         return updateLog.startAsync()
                 .thenCompose(none -> {
                     if (latestCatalogVersion() == emptyCatalog.version()) {
-                        // node has not seen any updates yet, let's try to initialise
-                        // catalog with default zone
-                        return createDefaultZone(emptyCatalog);
-                    }
+                        int initializedCatalogVersion = emptyCatalog.version() + 1;
+
+                        this.catalogReadyFuture(initializedCatalogVersion)
+                                .thenCompose(ignored -> awaitVersionActivation(initializedCatalogVersion))
+                                .handle((r, e) -> catalogInitializationFuture.complete(null));
 
-                    return nullCompletedFuture();
+                        return initCatalog(emptyCatalog);
+                    } else {
+                        catalogInitializationFuture.complete(null);
+                        return nullCompletedFuture();
+                    }
                 });
     }
 
@@ -205,7 +194,11 @@ public CompletableFuture<Void> stopAsync() {
 
     @Override
     public @Nullable CatalogTableDescriptor table(String tableName, long timestamp) {
-        return catalogAt(timestamp).schema(DEFAULT_SCHEMA_NAME).table(tableName);
+        CatalogSchemaDescriptor schema = catalogAt(timestamp).schema(DEFAULT_SCHEMA_NAME);
+        if (schema == null) {
+            return null;
+        }
+        return schema.table(tableName);
     }
 
     @Override
@@ -225,7 +218,11 @@ public Collection<CatalogTableDescriptor> tables(int catalogVersion) {
 
     @Override
     public @Nullable CatalogIndexDescriptor aliveIndex(String indexName, long timestamp) {
-        return catalogAt(timestamp).schema(DEFAULT_SCHEMA_NAME).aliveIndex(indexName);
+        CatalogSchemaDescriptor schema = catalogAt(timestamp).schema(DEFAULT_SCHEMA_NAME);
+        if (schema == null) {
+            return null;
+        }
+        return schema.aliveIndex(indexName);
     }
 
     @Override
@@ -321,6 +318,11 @@ public CompletableFuture<Void> catalogReadyFuture(int version) {
         return versionTracker.waitFor(version);
     }
 
+    @Override
+    public CompletableFuture<Void> catalogInitializationFuture() {
+        return catalogInitializationFuture;
+    }
+
     @Override
     public @Nullable Catalog catalog(int catalogVersion) {
         return catalogByVer.get(catalogVersion);
@@ -363,8 +365,9 @@ public CompletableFuture<Boolean> compactCatalog(long timestamp) {
         return updateLog.saveSnapshot(new SnapshotEntry(catalog));
     }
 
-    private CompletableFuture<Void> createDefaultZone(Catalog emptyCatalog) {
-        List<UpdateEntry> createZoneEntries = new BulkUpdateProducer(List.of(
+    private CompletableFuture<Void> initCatalog(Catalog emptyCatalog) {
+        List<CatalogCommand> initCommands = List.of(
+                // Init default zone
                 CreateZoneCommand.builder()
                         .zoneName(DEFAULT_ZONE_NAME)
                         .partitions(DEFAULT_PARTITION_COUNT)
@@ -378,10 +381,15 @@ private CompletableFuture<Void> createDefaultZone(Catalog emptyCatalog) {
                         .build(),
                 AlterZoneSetDefaultCatalogCommand.builder()
                         .zoneName(DEFAULT_ZONE_NAME)
-                        .build()
-        )).get(emptyCatalog);
+                        .build(),
+                // Add schemas
+                CreateSchemaCommand.builder().name(DEFAULT_SCHEMA_NAME).build(),
+                CreateSchemaCommand.builder().name(SYSTEM_SCHEMA_NAME).build()
+        );
+
+        List<UpdateEntry> entries = new BulkUpdateProducer(initCommands).get(emptyCatalog);
 
-        return updateLog.append(new VersionedUpdate(emptyCatalog.version() + 1, 0L, createZoneEntries))
+        return updateLog.append(new VersionedUpdate(emptyCatalog.version() + 1, 0L, entries))
                 .handle((result, error) -> {
                     if (error != null) {
                         LOG.warn("Unable to create default zone.", error);
@@ -407,13 +415,7 @@ private CompletableFuture<Integer> saveUpdateAndWaitForActivation(UpdateProducer
         CompletableFuture<Integer> resultFuture = new CompletableFuture<>();
 
         saveUpdate(updateProducer, 0)
-                .thenCompose(newVersion -> {
-                    Catalog catalog = catalogByVer.get(newVersion);
-
-                    HybridTimestamp tsSafeForRoReadingInPastOptimization = calcClusterWideEnsureActivationTime(catalog);
-
-                    return clockService.waitFor(tsSafeForRoReadingInPastOptimization).thenApply(unused -> newVersion);
-                })
+                .thenCompose(this::awaitVersionActivation)
                 .whenComplete((newVersion, err) -> {
                     if (err != null) {
                         Throwable errUnwrapped = ExceptionUtils.unwrapCause(err);
@@ -448,6 +450,14 @@ private CompletableFuture<Integer> saveUpdateAndWaitForActivation(UpdateProducer
         return resultFuture;
     }
 
+    private CompletableFuture<Integer> awaitVersionActivation(int version) {
+        Catalog catalog = catalogByVer.get(version);
+
+        HybridTimestamp tsSafeForRoReadingInPastOptimization = calcClusterWideEnsureActivationTime(catalog);
+
+        return clockService.waitFor(tsSafeForRoReadingInPastOptimization).thenApply(unused -> version);
+    }
+
     private HybridTimestamp calcClusterWideEnsureActivationTime(Catalog catalog) {
         return clusterWideEnsuredActivationTsSafeForRoReads(
                 catalog,

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

@@ -45,8 +45,10 @@
  * <p>TBD: events
  */
 public interface CatalogService extends EventProducer<CatalogEvent, CatalogEventParameters> {
+    /** Default schema name. */
     String DEFAULT_SCHEMA_NAME = "PUBLIC";
 
+    /** System schema name. */
     String SYSTEM_SCHEMA_NAME = "SYSTEM";
 
     /** Default storage profile. */
@@ -110,4 +112,9 @@ public interface CatalogService extends EventProducer<CatalogEvent, CatalogEvent
      * @param version Catalog version to wait for.
      */
     CompletableFuture<Void> catalogReadyFuture(int version);
+
+    /**
+     * Returns a future, which completes when empty catalog is initialised. Otherwise this future completes upon startup.
+     */
+    CompletableFuture<Void> catalogInitializationFuture();
 }

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AbstractCreateIndexCommand.java

@@ -84,7 +84,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         }
 
         return List.of(
-                new NewIndexEntry(createDescriptor(catalog.objectIdGenState(), table.id(), catalog.version() + 1), schemaName),
+                new NewIndexEntry(createDescriptor(catalog.objectIdGenState(), table.id(), catalog.version() + 1), schema.id()),
                 new ObjectIdGenUpdateEntry(1)
         );
     }

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAddColumnCommand.java

@@ -87,7 +87,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         }
 
         return List.of(
-                new NewColumnsEntry(table.id(), columnDescriptors, schemaName)
+                new NewColumnsEntry(table.id(), columnDescriptors)
         );
     }
 

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableAlterColumnCommand.java

@@ -113,7 +113,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         }
 
         return List.of(
-                new AlterColumnEntry(table.id(), target, schemaName)
+                new AlterColumnEntry(table.id(), target)
         );
     }
 

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/AlterTableDropColumnCommand.java

@@ -105,7 +105,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         });
 
         return List.of(
-                new DropColumnsEntry(table.id(), columns, schemaName)
+                new DropColumnsEntry(table.id(), columns)
         );
     }
 

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateSchemaCommand.java

@@ -0,0 +1,95 @@
+/*
+ * 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.commands;
+
+import static org.apache.ignite.internal.catalog.CatalogParamsValidationUtils.validateIdentifier;
+import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+
+import java.util.List;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.CatalogCommand;
+import org.apache.ignite.internal.catalog.CatalogValidationException;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogSystemViewDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.storage.NewSchemaEntry;
+import org.apache.ignite.internal.catalog.storage.ObjectIdGenUpdateEntry;
+import org.apache.ignite.internal.catalog.storage.UpdateEntry;
+
+/**
+ * Command to create a new schema.
+ */
+public class CreateSchemaCommand implements CatalogCommand {
+
+    private final String schemaName;
+
+    private CreateSchemaCommand(String schemaName) {
+        validateIdentifier(schemaName, "Name of the schema");
+
+        this.schemaName = schemaName;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public List<UpdateEntry> get(Catalog catalog) {
+        int id = catalog.objectIdGenState();
+
+        if (catalog.schema(schemaName) != null) {
+            throw new CatalogValidationException(format("Schema with name '{}' already exists", schemaName));
+        }
+
+        CatalogSchemaDescriptor schema = new CatalogSchemaDescriptor(
+                id,
+                schemaName,
+                new CatalogTableDescriptor[0],
+                new CatalogIndexDescriptor[0],
+                new CatalogSystemViewDescriptor[0],
+                0
+        );
+
+        return List.of(
+                new NewSchemaEntry(schema),
+                new ObjectIdGenUpdateEntry(1)
+        );
+    }
+
+    /** Returns builder to create a command to create a new schema. */
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    /** Implementation of {@link CreateSchemaCommandBuilder}. */
+    public static class Builder implements CreateSchemaCommandBuilder {
+
+        private String name;
+
+        /** {@inheritDoc} */
+        @Override
+        public CreateSchemaCommandBuilder name(String name) {
+            this.name = name;
+            return this;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public CatalogCommand build() {
+            return new CreateSchemaCommand(name);
+        }
+    }
+}

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateSchemaCommandBuilder.java

@@ -0,0 +1,32 @@
+/*
+ * 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.commands;
+
+import org.apache.ignite.internal.catalog.CatalogCommand;
+
+/**
+ * Builder for a {@link CreateSchemaCommand}.
+ */
+public interface CreateSchemaCommandBuilder {
+
+    /** Sets schema name. Should not be null or blank. */
+    CreateSchemaCommandBuilder name(String name);
+
+    /** Creates new schema command. */
+    CatalogCommand build();
+}

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateSystemViewCommand.java

@@ -108,7 +108,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         CatalogSchemaDescriptor systemSchema = schemaOrThrow(catalog, CatalogManager.SYSTEM_SCHEMA_NAME);
 
         List<CatalogTableColumnDescriptor> viewColumns = columns.stream().map(CatalogUtils::fromParams).collect(toList());
-        CatalogSystemViewDescriptor descriptor = new CatalogSystemViewDescriptor(id, name, viewColumns, systemViewType);
+        CatalogSystemViewDescriptor descriptor = new CatalogSystemViewDescriptor(id, systemSchema.id(), name, viewColumns, systemViewType);
 
         CatalogSystemViewDescriptor existingSystemView = systemSchema.systemView(name);
 
@@ -121,7 +121,7 @@ public List<UpdateEntry> get(Catalog catalog) {
         }
 
         return List.of(
-                new NewSystemViewEntry(descriptor, systemSchema.name()),
+                new NewSystemViewEntry(descriptor),
                 new ObjectIdGenUpdateEntry(1)
         );
     }

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CreateTableCommand.java

@@ -150,8 +150,8 @@ public List<UpdateEntry> get(Catalog catalog) {
         CatalogIndexDescriptor pkIndex = createIndexDescriptor(txWaitCatalogVersion, indexName, pkIndexId, tableId);
 
         return List.of(
-                new NewTableEntry(table, schemaName),
-                new NewIndexEntry(pkIndex, schemaName),
+                new NewTableEntry(table),
+                new NewIndexEntry(pkIndex, schema.id()),
                 new MakeIndexAvailableEntry(pkIndexId),
                 new ObjectIdGenUpdateEntry(id - catalog.objectIdGenState())
         );

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/DropTableCommand.java

@@ -62,7 +62,7 @@ public List<UpdateEntry> get(Catalog catalog) {
                     updateEntries.add(new RemoveIndexEntry(index.id()));
                 });
 
-        updateEntries.add(new DropTableEntry(table.id(), schemaName));
+        updateEntries.add(new DropTableEntry(table.id()));
 
         return updateEntries;
     }