[AMORO-2423] [Flink]: Using 'mixed_iceberg' and 'mixed_hive' indentif…

…ier to CREATE CATALOG and deprecate 'arctic' identifier (apache#2424)

* [AMORO-2423] [Flink]: Using 'mixed_iceberg' and 'mixed_hive' identifiers to CREATE CATALOG and deprecate 'arctic' identifier

mixed/flink/flink-common/src/main/java/com/netease/arctic/flink/catalog/factories/ArcticCatalogFactoryOptions.java

@@ -30,7 +30,9 @@
 /** {@link ConfigOption}s for {@link ArcticCatalog}. */
 @Internal
 public class ArcticCatalogFactoryOptions {
-  public static final String IDENTIFIER = "arctic";
+  public static final String MIXED_ICEBERG_IDENTIFIER = "mixed_iceberg";
+  public static final String MIXED_HIVE_IDENTIFIER = "mixed_hive";
+  @Deprecated public static final String IDENTIFIER = "arctic";
   public static final String UNIFIED_IDENTIFIER = "unified";
 
   public static final ConfigOption<String> DEFAULT_DATABASE =

mixed/flink/flink-common/src/main/java/com/netease/arctic/flink/catalog/factories/MixedHiveCatalogFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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 com.netease.arctic.flink.catalog.factories;
+
+import com.netease.arctic.flink.catalog.ArcticCatalog;
+
+/**
+ * The factory to create {@link ArcticCatalog} with {@link
+ * ArcticCatalogFactoryOptions#MIXED_HIVE_IDENTIFIER} identifier.
+ */
+public class MixedHiveCatalogFactory extends ArcticCatalogFactory {
+
+  @Override
+  public String factoryIdentifier() {
+    return ArcticCatalogFactoryOptions.MIXED_HIVE_IDENTIFIER;
+  }
+}

mixed/flink/flink-common/src/main/java/com/netease/arctic/flink/catalog/factories/MixedIcebergCatalogFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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 com.netease.arctic.flink.catalog.factories;
+
+import com.netease.arctic.flink.catalog.ArcticCatalog;
+
+/**
+ * The factory to create {@link ArcticCatalog} with {@link
+ * ArcticCatalogFactoryOptions#MIXED_ICEBERG_IDENTIFIER} identifier.
+ */
+public class MixedIcebergCatalogFactory extends ArcticCatalogFactory {
+
+  @Override
+  public String factoryIdentifier() {
+    return ArcticCatalogFactoryOptions.MIXED_ICEBERG_IDENTIFIER;
+  }
+}

mixed/flink/flink-common/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -16,6 +16,8 @@
 # limitations under the License.
 #
 
+com.netease.arctic.flink.catalog.factories.MixedIcebergCatalogFactory
+com.netease.arctic.flink.catalog.factories.MixedHiveCatalogFactory
 com.netease.arctic.flink.catalog.factories.ArcticCatalogFactory
 com.netease.arctic.flink.catalog.factories.FlinkCatalogFactory
 com.netease.arctic.flink.table.DynamicTableFactory

mixed/flink/flink-common/src/test/java/com/netease/arctic/flink/catalog/TestCatalog.java

@@ -98,13 +98,30 @@ public void before() throws Exception {
 
   @After
   public void after() {
-    sql("DROP TABLE " + CATALOG + "." + DB + "." + TABLE);
-    sql("DROP DATABASE " + CATALOG + "." + DB);
+    sql("DROP TABLE IF EXISTS " + CATALOG + "." + DB + "." + TABLE);
+    sql("DROP DATABASE IF EXISTS " + CATALOG + "." + DB);
     Assert.assertTrue(CollectionUtil.isNullOrEmpty(getMixedFormatCatalog().listDatabases()));
     sql("USE CATALOG default_catalog");
     sql("DROP CATALOG " + CATALOG);
   }
 
+  @Test
+  public void testCreateIcebergHiveCatalog() {
+    sql(
+        "CREATE CATALOG mixed_iceberg_catalog WITH ('type'='mixed_iceberg', 'metastore.url'='%s')",
+        getCatalogUrl());
+    sql(
+        "CREATE CATALOG mixed_hive_catalog WITH ('type'='mixed_hive', 'metastore.url'='%s')",
+        getCatalogUrl());
+
+    String[] catalogs = getTableEnv().listCatalogs();
+    Assert.assertArrayEquals(
+        Arrays.stream(catalogs).sorted().toArray(),
+        Stream.of("default_catalog", "arcticCatalog", "mixed_iceberg_catalog", "mixed_hive_catalog")
+            .sorted()
+            .toArray());
+  }
+
   @Test
   public void testDDL() throws IOException {
     sql(