IGNITE-24025 Introduce catalog command to drop schema

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

@@ -48,6 +48,18 @@ public interface CatalogService extends EventProducer<CatalogEvent, CatalogEvent
     /** System schema name. */
     String SYSTEM_SCHEMA_NAME = "SYSTEM";
 
+    /**
+     * Information schema - a system schema defined by SQL standard.
+     * The schema provides system-views, which describe Catalog objects, and can be read by a user.
+     */
+    String INFORMATION_SCHEMA = "INFORMATION_SCHEMA";
+
+    /**
+     * Definition schema - a system schema defined by SQL standard.
+     * The schema provides tables/sources for Catalog object’s metadata and can’t be accessed by a user directly.
+     */
+    String DEFINITION_SCHEMA = "DEFINITION_SCHEMA";
+
     /** Default storage profile. */
     String DEFAULT_STORAGE_PROFILE = "default";
 

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

@@ -18,6 +18,8 @@
 package org.apache.ignite.internal.catalog.commands;
 
 import static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.catalog.CatalogService.DEFINITION_SCHEMA;
+import static org.apache.ignite.internal.catalog.CatalogService.INFORMATION_SCHEMA;
 import static org.apache.ignite.internal.catalog.CatalogService.SYSTEM_SCHEMA_NAME;
 import static org.apache.ignite.internal.catalog.commands.DefaultValue.Type.FUNCTION_CALL;
 import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
@@ -145,7 +147,11 @@ public class CatalogUtils {
         FUNCTIONAL_DEFAULT_FUNCTIONS.put("RAND_UUID", ColumnType.UUID);
     }
 
-    public static final List<String> SYSTEM_SCHEMAS = List.of(SYSTEM_SCHEMA_NAME);
+    public static final List<String> SYSTEM_SCHEMAS = List.of(
+            SYSTEM_SCHEMA_NAME,
+            DEFINITION_SCHEMA,
+            INFORMATION_SCHEMA
+    );
 
     /** System schema names. */
     public static boolean isSystemSchema(String schemaName) {

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

@@ -0,0 +1,115 @@
+/*
+ * 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.commands.CatalogUtils.schemaOrThrow;
+
+import java.util.ArrayList;
+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.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.storage.DropIndexEntry;
+import org.apache.ignite.internal.catalog.storage.DropSchemaEntry;
+import org.apache.ignite.internal.catalog.storage.DropTableEntry;
+import org.apache.ignite.internal.catalog.storage.UpdateEntry;
+
+/**
+ * A command that drops a schema with specified name.
+ */
+public class DropSchemaCommand implements CatalogCommand {
+    /** Returns builder to create a command to drop schema with specified name. */
+    public static DropSchemaCommandBuilder builder() {
+        return new Builder();
+    }
+
+    private final String schemaName;
+
+    private final boolean cascade;
+
+    /**
+     * Constructor.
+     *
+     * @param schemaName Name of the schema.
+     * @param cascade Flag indicating forced deletion of a non-empty schema.
+     * @throws CatalogValidationException if any of restrictions above is violated.
+     */
+    private DropSchemaCommand(String schemaName, boolean cascade) throws CatalogValidationException {
+        this.schemaName = schemaName;
+        this.cascade = cascade;
+    }
+
+    @Override
+    public List<UpdateEntry> get(Catalog catalog) {
+        if (CatalogUtils.isSystemSchema(schemaName)) {
+            throw new CatalogValidationException("System schema can't be dropped [name={}].", schemaName);
+        }
+
+        CatalogSchemaDescriptor schema = schemaOrThrow(catalog, schemaName);
+
+        if (!cascade && (schema.indexes().length > 0 || schema.tables().length > 0)) {
+            throw new CatalogValidationException("Schema '{}' is not empty. Use CASCADE to drop it anyway.", schemaName);
+        }
+
+        assert schema.systemViews().length == 0 : "name=" + schemaName + ", count=" + schema.systemViews().length;
+
+        List<UpdateEntry> updateEntries = new ArrayList<>();
+
+        for (CatalogIndexDescriptor idx : schema.indexes()) {
+            updateEntries.add(new DropIndexEntry(idx.id()));
+        }
+
+        for (CatalogTableDescriptor tbl : schema.tables()) {
+            updateEntries.add(new DropTableEntry(tbl.id()));
+        }
+
+        updateEntries.add(new DropSchemaEntry(schema.id()));
+
+        return updateEntries;
+    }
+
+    /**
+     * Implementation of {@link DropSchemaCommandBuilder}.
+     */
+    private static class Builder implements DropSchemaCommandBuilder {
+        private String schemaName;
+        private boolean cascade;
+
+        @Override
+        public DropSchemaCommandBuilder name(String schemaName) {
+            this.schemaName = schemaName;
+
+            return this;
+        }
+
+        @Override
+        public DropSchemaCommandBuilder cascade(boolean cascade) {
+            this.cascade = cascade;
+
+            return this;
+        }
+
+        @Override
+        public CatalogCommand build() {
+            return new DropSchemaCommand(schemaName, cascade);
+        }
+    }
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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 of a command that drop specified schema.
+ */
+public interface DropSchemaCommandBuilder {
+    /** Sets schema name. Should not be null or blank. */
+    DropSchemaCommandBuilder name(String schemaName);
+
+    /** Sets flag indicating forced deletion of a non-empty schema. */
+    DropSchemaCommandBuilder cascade(boolean cascade);
+
+    /** Returns a command with specified parameters. */
+    CatalogCommand build();
+}

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/DropSchemaEntry.java

@@ -0,0 +1,91 @@
+/*
+ * 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.storage;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.defaultZoneIdOpt;
+
+import java.io.IOException;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.storage.serialization.CatalogObjectSerializer;
+import org.apache.ignite.internal.catalog.storage.serialization.MarshallableEntryType;
+import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.io.IgniteDataInput;
+import org.apache.ignite.internal.util.io.IgniteDataOutput;
+
+/**
+ * Describes deletion of a schema.
+ */
+public class DropSchemaEntry implements UpdateEntry {
+    public static final CatalogObjectSerializer<DropSchemaEntry> SERIALIZER = new DropSchemaSerializer();
+
+    private final int schemaId;
+
+    /**
+     * Constructs the object.
+     *
+     * @param schemaId An id of a schema to drop.
+     */
+    public DropSchemaEntry(int schemaId) {
+        this.schemaId = schemaId;
+    }
+
+    /** Returns an id of a schema to drop. */
+    public int schemaId() {
+        return schemaId;
+    }
+
+    @Override
+    public int typeId() {
+        return MarshallableEntryType.DROP_SCHEMA.id();
+    }
+
+    @Override
+    public Catalog applyUpdate(Catalog catalog, long causalityToken) {
+        return new Catalog(
+                catalog.version(),
+                catalog.time(),
+                catalog.objectIdGenState(),
+                catalog.zones(),
+                catalog.schemas().stream().filter(s -> s.id() != schemaId).collect(toList()),
+                defaultZoneIdOpt(catalog)
+        );
+    }
+
+    @Override
+    public String toString() {
+        return S.toString(this);
+    }
+
+    /**
+     * Serializer for {@link DropSchemaEntry}.
+     */
+    private static class DropSchemaSerializer implements CatalogObjectSerializer<DropSchemaEntry> {
+        @Override
+        public DropSchemaEntry readFrom(IgniteDataInput input) throws IOException {
+            int schemaId = input.readVarIntAsInt();
+
+            return new DropSchemaEntry(schemaId);
+        }
+
+        @Override
+        public void writeTo(DropSchemaEntry entry, IgniteDataOutput output) throws IOException {
+            output.writeVarInt(entry.schemaId());
+        }
+    }
+}

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/serialization/CatalogEntrySerializerProvider.java

@@ -23,6 +23,7 @@
 import org.apache.ignite.internal.catalog.storage.AlterZoneEntry;
 import org.apache.ignite.internal.catalog.storage.DropColumnsEntry;
 import org.apache.ignite.internal.catalog.storage.DropIndexEntry;
+import org.apache.ignite.internal.catalog.storage.DropSchemaEntry;
 import org.apache.ignite.internal.catalog.storage.DropTableEntry;
 import org.apache.ignite.internal.catalog.storage.DropZoneEntry;
 import org.apache.ignite.internal.catalog.storage.MakeIndexAvailableEntry;
@@ -81,6 +82,7 @@ public interface CatalogEntrySerializerProvider {
             serializers[MarshallableEntryType.RENAME_INDEX.id()] = RenameIndexEntry.SERIALIZER;
             serializers[MarshallableEntryType.SET_DEFAULT_ZONE.id()] = SetDefaultZoneEntry.SERIALIZER;
             serializers[MarshallableEntryType.NEW_SCHEMA.id()] = NewSchemaEntry.SERIALIZER;
+            serializers[MarshallableEntryType.DROP_SCHEMA.id()] = DropSchemaEntry.SERIALIZER;
             //noinspection ThisEscapedInObjectConstruction
             serializers[MarshallableEntryType.VERSIONED_UPDATE.id()] = new VersionedUpdateSerializer(this);
 

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/serialization/MarshallableEntryType.java

@@ -41,7 +41,8 @@ public enum MarshallableEntryType {
     VERSIONED_UPDATE(17),
     RENAME_INDEX(18),
     SET_DEFAULT_ZONE(19),
-    NEW_SCHEMA(20);
+    NEW_SCHEMA(20),
+    DROP_SCHEMA(21);
 
     /** Type ID. */
     private final int id;

modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogIndexTest.java

@@ -848,10 +848,10 @@ private void createSomeIndex(String tableName, String indexName) {
     }
 
     private void createSomeSortedIndex(String tableName, String indexName) {
-        assertThat(
-                manager.execute(createSortedIndexCommand(tableName, indexName, false, List.of("key1"), List.of(ASC_NULLS_LAST))),
-                willCompleteSuccessfully()
-        );
+        CatalogCommand newSortedIndexCommand = createSortedIndexCommand(
+                SqlCommon.DEFAULT_SCHEMA_NAME, tableName, indexName, false, List.of("key1"), List.of(ASC_NULLS_LAST));
+
+        assertThat(manager.execute(newSortedIndexCommand), willCompleteSuccessfully());
     }
 
     private void renameIndex(String indexName, String newIndexName) {