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,9 +44,19 @@ public class CreateSchemaCommand implements CatalogCommand {
 
     private final boolean ifNotExists;
 
-    private CreateSchemaCommand(String schemaName, boolean ifNotExists) {
+    private CreateSchemaCommand(String schemaName, boolean ifNotExists, boolean systemSchemaCommand) {
         validateIdentifier(schemaName, "Name of the schema");
 
+        if (systemSchemaCommand) {
+            if (!CatalogUtils.isSystemSchema(schemaName)) {
+                throw new CatalogValidationException(format("Not a system schema, schema: '{}'", schemaName));
+            }
+        } else {
+            if (CatalogUtils.isSystemSchema(schemaName)) {
+                throw new CatalogValidationException("Reserved system schema with name '{}' can't be created.", schemaName);
+            }
+        }
+
         this.schemaName = schemaName;
         this.ifNotExists = ifNotExists;
     }
@@ -112,7 +123,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,23 @@
 
 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.CreateSchemaCommandBuilder;
 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 {
@@ -85,12 +90,21 @@ public void testCreateSchemaWithNullOrEmptyNameIsRejected() {
         );
     }
 
+    @SuppressWarnings("ThrowableNotThrown")
+    @ParameterizedTest
+    @MethodSource("reservedSchemaNames")
+    public void testCreateSystemSchemaIsRejected(String schemaName) {
+        CreateSchemaCommandBuilder createCmd = CreateSchemaCommand.builder().name(schemaName);
+
+        IgniteTestUtils.assertThrows(
+                CatalogValidationException.class,
+                () -> manager.execute(createCmd.build()),
+                format("Reserved system schema with name '{}' can't be created.", schemaName)
+        );
+    }
+
     @ParameterizedTest
-    @ValueSource(strings = {
-            CatalogService.SYSTEM_SCHEMA_NAME,
-            CatalogService.INFORMATION_SCHEMA,
-            CatalogService.DEFINITION_SCHEMA
-    })
+    @MethodSource("reservedSchemaNames")
     public void testDropSystemSchemaIsForbidden(String schemaName) {
         CatalogCommand dropCmd = DropSchemaCommand.builder().name(schemaName).build();
 
@@ -100,6 +114,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(new UpdateContext(catalog));
     }
 
     private static Catalog catalogWithSchema(String schemaName) {

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

@@ -0,0 +1,79 @@
+/*
+ * 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.apache.ignite.internal.catalog.UpdateContext;
+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 CreateSystemSchemaValidationTest 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(new UpdateContext(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(new UpdateContext(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);
     }

modules/sql-engine/src/integrationTest/sql/group1/identifiers/test_long_identifiers.test

@@ -2,8 +2,6 @@
 # description: SQL feature F391 (Long identifiers)
 # group: [identifiers]
 
-# TODO: IGNITE-19703 Add cases for long identifiers for schema names.
-
 statement ok
 PRAGMA enable_verification
 
@@ -13,6 +11,21 @@ tableName_veryLooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 statement error: Non-query expression encountered in illegal context.
 SELECT 1; tableName_veryLooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf129Characters
 
+# Create schema with long identifiers
+statement ok
+CREATE SCHEMA identifier_veryLooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters;
+
+statement error: Length of identifier
+CREATE SCHEMA identifier_veryLoooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf129Characters;
+
+# Create schema and table with long identifiers
+statement ok
+CREATE TABLE identifier_veryLooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters.tableName_veryLoooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters (keyColumnName_veryLoooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters INTEGER, valueColumnName_veryLoooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters INTEGER, PRIMARY KEY (keyColumnName_veryLoooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters));
+
+statement ok
+DROP SCHEMA identifier_veryLooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongIdentifierOf128Characters CASCADE;
+
+
 # Create table with short identifiers for test simplicity purpose
 statement ok
 CREATE TABLE t (id INTEGER, val INTEGER, PRIMARY KEY (id))