IGNITE-24022 Sql. Fix CreateSchemaCommand catalog command validation

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

@@ -283,7 +283,7 @@ private CompletableFuture<Void> initCatalog(Catalog emptyCatalog) {
                         .build(),
                 // Add schemas
                 CreateSchemaCommand.builder().name(SqlCommon.DEFAULT_SCHEMA_NAME).build(),
-                CreateSchemaCommand.builder().name(SYSTEM_SCHEMA_NAME).build()
+                CreateSchemaCommand.systemSchemaBuilder().name(SYSTEM_SCHEMA_NAME).build()
         );
 
         List<UpdateEntry> entries = new BulkUpdateProducer(initCommands).get(new UpdateContext(emptyCatalog));

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

@@ -24,6 +24,7 @@
 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.SchemaExistsException;
 import org.apache.ignite.internal.catalog.UpdateContext;
 import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
@@ -43,11 +44,18 @@ public class CreateSchemaCommand implements CatalogCommand {
 
     private final boolean ifNotExists;
 
-    private CreateSchemaCommand(String schemaName, boolean ifNotExists) {
+    private final boolean systemSchemaCommand;
+
+    private CreateSchemaCommand(String schemaName, boolean ifNotExists, boolean systemSchemaCommand) {
         validateIdentifier(schemaName, "Name of the schema");
 
+        if (systemSchemaCommand && !CatalogUtils.isSystemSchema(schemaName)) {
+            throw new CatalogValidationException(format("Not a system schema, schema: '{}'", schemaName));
+        }
+
         this.schemaName = schemaName;
         this.ifNotExists = ifNotExists;
+        this.systemSchemaCommand = systemSchemaCommand;
     }
 
     public boolean ifNotExists() {
@@ -57,6 +65,10 @@ public boolean ifNotExists() {
     /** {@inheritDoc} */
     @Override
     public List<UpdateEntry> get(UpdateContext updateContext) {
+        if (!systemSchemaCommand && CatalogUtils.isSystemSchema(schemaName)) {
+            throw new CatalogValidationException("Reserved system schema with name '{}' can't be created.", schemaName);
+        }
+
         Catalog catalog = updateContext.catalog();
         int id = catalog.objectIdGenState();
 
@@ -112,7 +124,31 @@ public CreateSchemaCommandBuilder ifNotExists(boolean value) {
         /** {@inheritDoc} */
         @Override
         public CatalogCommand build() {
-            return new CreateSchemaCommand(name, ifNotExists);
+            return new CreateSchemaCommand(name, ifNotExists, false);
+        }
+    }
+
+    /** Returns builder to create a command to create a system schema. */
+    public static SystemSchemaBuilder systemSchemaBuilder() {
+        return new SystemSchemaBuilder();
+    }
+
+    /** Implementation of {@link CreateSystemSchemaCommandBuilder}. */
+    public static class SystemSchemaBuilder implements CreateSystemSchemaCommandBuilder {
+
+        private String name;
+
+        /** {@inheritDoc} */
+        @Override
+        public CreateSystemSchemaCommandBuilder name(String name) {
+            this.name = name;
+            return this;
+        }
+
+        /** {@inheritDoc} */
+        @Override
+        public CatalogCommand build() {
+            return new CreateSchemaCommand(name, false, true);
         }
     }
 }

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

@@ -17,19 +17,10 @@
 
 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);
-
+public interface CreateSchemaCommandBuilder extends CreateSystemSchemaCommandBuilder {
     /** Sets a flag indicating whether {@code IF NOT EXISTS} option was specified. */
     CreateSchemaCommandBuilder ifNotExists(boolean value);
-
-    /** Creates new schema command. */
-    CatalogCommand build();
 }

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

@@ -0,0 +1,31 @@
+/*
+ * 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} with system schemas usage.
+ */
+public interface CreateSystemSchemaCommandBuilder {
+    /** Sets schema name. Should not be null or blank. */
+    CreateSystemSchemaCommandBuilder name(String name);
+
+    /** Creates new schema command. */
+    CatalogCommand build();
+}

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

@@ -17,18 +17,22 @@
 
 package org.apache.ignite.internal.catalog;
 
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.SYSTEM_SCHEMAS;
 import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
 import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
 import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
 import static org.hamcrest.MatcherAssert.assertThat;
 
+import java.util.Locale;
+import java.util.stream.Stream;
 import org.apache.ignite.internal.catalog.commands.CreateSchemaCommand;
 import org.apache.ignite.internal.catalog.commands.DropSchemaCommand;
 import org.apache.ignite.internal.sql.SqlCommon;
 import org.apache.ignite.internal.testframework.IgniteTestUtils;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 /** Validation tests for schema related commands. */
 public class CatalogSchemaValidationTest extends BaseCatalogManagerTest {
@@ -86,11 +90,18 @@ public void testCreateSchemaWithNullOrEmptyNameIsRejected() {
     }
 
     @ParameterizedTest
-    @ValueSource(strings = {
-            CatalogService.SYSTEM_SCHEMA_NAME,
-            CatalogService.INFORMATION_SCHEMA,
-            CatalogService.DEFINITION_SCHEMA
-    })
+    @MethodSource("reservedSchemaNames")
+    public void testCreateSystemSchemaIsRejected(String schemaName) {
+        CatalogCommand createCmd = CreateSchemaCommand.builder().name(schemaName).build();
+
+        assertThat(manager.execute(createCmd),
+                willThrowFast(CatalogValidationException.class, format("Reserved system schema with name '{}' can't be created.",
+                        schemaName))
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("reservedSchemaNames")
     public void testDropSystemSchemaIsForbidden(String schemaName) {
         CatalogCommand dropCmd = DropSchemaCommand.builder().name(schemaName).build();
 
@@ -100,6 +111,23 @@ public void testDropSystemSchemaIsForbidden(String schemaName) {
         );
     }
 
+    private static Stream<Arguments> reservedSchemaNames() {
+        return SYSTEM_SCHEMAS.stream().map(Arguments::of);
+    }
+
+    @ParameterizedTest
+    @MethodSource("reservedSchemaNames")
+    public void testDropNotExistSchemas(String schemaName) {
+        schemaName = schemaName.toLowerCase(Locale.ROOT);
+
+        CatalogCommand dropCmd = DropSchemaCommand.builder().name(schemaName).build();
+
+        assertThat(
+                manager.execute(dropCmd),
+                willThrowFast(CatalogValidationException.class, format("Schema with name '{}' not found", schemaName))
+        );
+    }
+
     @Test
     @SuppressWarnings("ThrowableNotThrown")
     public void testDropSchemaWithNullOrEmptyNameIsRejected() {

modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CreateSchemaCommandValidationTest.java

@@ -29,6 +29,7 @@
 /**
  * Tests to verify validation of {@link CreateSchemaCommand}.
  */
+@SuppressWarnings({"ThrowableNotThrown"})
 public class CreateSchemaCommandValidationTest extends AbstractCommandValidationTest {
 
     @ParameterizedTest(name = "[{index}] ''{argumentsWithNames}''")
@@ -56,6 +57,8 @@ void commandFailsWhenSchemaAlreadyExists() {
                 () -> builder.build().get(new UpdateContext(catalog)),
                 "Schema with name 'TEST' already exists"
         );
+
+        builder.ifNotExists(true).build().get(catalog);
     }
 
     private static Catalog catalogWithSchema(String schemaName) {

modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CreateSystemSchemaCommandValidationTest.java

@@ -0,0 +1,78 @@
+/*
+ * 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.CatalogService.INFORMATION_SCHEMA;
+import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrows;
+
+import java.util.Locale;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.CatalogValidationException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Tests to verify validation of {@link CreateSchemaCommand} for system schemas.
+ */
+@SuppressWarnings({"ThrowableNotThrown"})
+public class CreateSystemSchemaCommandValidationTest extends AbstractCommandValidationTest {
+
+    @ParameterizedTest(name = "[{index}] ''{argumentsWithNames}''")
+    @MethodSource("nullAndBlankStrings")
+    void schemaNameMustNotBeNullOrBlank(String name) {
+        CreateSystemSchemaCommandBuilder builder = CreateSchemaCommand.systemSchemaBuilder().name(name);
+
+        assertThrows(
+                CatalogValidationException.class,
+                builder::build,
+                "Name of the schema can't be null or blank"
+        );
+    }
+
+    @Test
+    void commandFailsWithNonSystemSchema() {
+        String schemaName = INFORMATION_SCHEMA.toLowerCase(Locale.ROOT);
+
+        Catalog catalog = catalogWithSchema(INFORMATION_SCHEMA);
+
+        assertThrows(
+                CatalogValidationException.class,
+                () -> CreateSchemaCommand.systemSchemaBuilder().name(schemaName).build().get(catalog),
+                format("Not a system schema, schema: '{}'", schemaName)
+        );
+    }
+
+    @Test
+    void commandFailsWhenSchemaAlreadyExists() {
+        CreateSystemSchemaCommandBuilder builder = CreateSchemaCommand.systemSchemaBuilder().name(INFORMATION_SCHEMA);
+
+        Catalog catalog = catalogWithSchema(INFORMATION_SCHEMA);
+
+        assertThrows(
+                CatalogValidationException.class,
+                () -> builder.build().get(catalog),
+                format("Schema with name '{}' already exists", INFORMATION_SCHEMA)
+        );
+    }
+
+    private static Catalog catalogWithSchema(String schemaName) {
+        return catalog(CreateSchemaCommand.systemSchemaBuilder().name(schemaName).build());
+    }
+}

modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItCreateTableDdlTest.java

@@ -323,7 +323,6 @@ public void literalAsColumDefault() {
     }
 
     @Test
-    @SuppressWarnings("ThrowableNotThrown")
     public void doNotAllowFunctionsInNonPkColumns() {
         // SQL Standard 2016 feature E141-07 - Basic integrity constraints. Column defaults
         assertThrowsSqlException(
@@ -339,7 +338,7 @@ public void testItIsNotPossibleToCreateTablesInSystemSchema(String schema) {
         if (DEFINITION_SCHEMA.equals(schema) || INFORMATION_SCHEMA.equals(schema)) {
             IgniteImpl igniteImpl = unwrapIgniteImpl(CLUSTER.aliveNode());
 
-            IgniteTestUtils.await(igniteImpl.catalogManager().execute(CreateSchemaCommand.builder().name(schema).build()));
+            IgniteTestUtils.await(igniteImpl.catalogManager().execute(CreateSchemaCommand.systemSchemaBuilder().name(schema).build()));
         }
 
         assertThrowsSqlException(
@@ -348,6 +347,44 @@ public void testItIsNotPossibleToCreateTablesInSystemSchema(String schema) {
                 () -> sql(format("CREATE TABLE {}.SYS_TABLE (NAME VARCHAR PRIMARY KEY, SIZE BIGINT)", schema.toLowerCase())));
     }
 
+    @ParameterizedTest
+    @MethodSource("reservedSchemaNames")
+    public void testCreateSystemSchemas(String schema) {
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("Reserved system schema with name '{}' can't be created.", schema),
+                () -> sql(format("CREATE SCHEMA {}", schema.toLowerCase())));
+
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("Reserved system schema with name '{}' can't be created.", schema),
+                () -> sql(format("CREATE SCHEMA {}", schema)));
+
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("Reserved system schema with name '{}' can't be created.", schema),
+                () -> sql(format("CREATE SCHEMA \"{}\"", schema)));
+    }
+
+    @ParameterizedTest
+    @MethodSource("reservedSchemaNames")
+    public void testDropSystemSchemas(String schema) {
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("System schema can't be dropped [name={}]", schema),
+                () -> sql(format("DROP SCHEMA {}", schema.toLowerCase())));
+
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("System schema can't be dropped [name={}]", schema),
+                () -> sql(format("DROP SCHEMA {}", schema)));
+
+        assertThrowsSqlException(
+                STMT_VALIDATION_ERR,
+                format("System schema can't be dropped [name={}]", schema),
+                () -> sql(format("DROP SCHEMA \"{}\"", schema)));
+    }
+
     private static Stream<Arguments> reservedSchemaNames() {
         return SYSTEM_SCHEMAS.stream().map(Arguments::of);
     }