From 3ac63f2e9379e27a390771faf67a01111a29add7 Mon Sep 17 00:00:00 2001
From: North Lin <37775475+qg-lin@users.noreply.github.com>
Date: Mon, 25 Nov 2024 10:47:52 +0800
Subject: [PATCH] [fix] Fix auto create database failed when it's name contains
 special symbols (#517)

---
 .../flink/sink/schema/SchemaChangeHelper.java   |  2 +-
 .../flink/sink/schema/SchemaManagerITCase.java  | 17 ++++++++++++++++-
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java
index d0630b03e..b772ec3d2 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeHelper.java
@@ -150,7 +150,7 @@ public static String buildDatabaseExistsQuery(String database) {
     }
 
     public static String buildCreateDatabaseDDL(String database) {
-        return String.format(CREATE_DATABASE_DDL, database);
+        return String.format(CREATE_DATABASE_DDL, DorisSchemaFactory.identifier(database));
     }
 
     public static String buildModifyColumnCommentDDL(
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java
index c4f7f282e..b3cc0abcb 100644
--- a/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java
+++ b/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java
@@ -226,8 +226,22 @@ public void testModifyColumnTypeAndComment()
     public void testCreateTableWhenDatabaseNotExists()
             throws IOException, IllegalArgumentException, InterruptedException {
         String databaseName = DATABASE + "_" + Integer.toUnsignedString(new Random().nextInt(), 36);
-        String tableName = "auto_create_database";
+        createTableWhenDatabaseNotExists(databaseName);
+    }
 
+    @Test
+    public void testCreateTableWhenDatabaseNotExistsAndContainsSpecialSymbol()
+            throws IOException, IllegalArgumentException, InterruptedException {
+        String databaseName =
+                DATABASE.replace("_", "-")
+                        + "_"
+                        + Integer.toUnsignedString(new Random().nextInt(), 36);
+        createTableWhenDatabaseNotExists(databaseName);
+    }
+
+    public void createTableWhenDatabaseNotExists(String databaseName)
+            throws IOException, IllegalArgumentException, InterruptedException {
+        String tableName = "auto_create_database";
         TableSchema tableSchema = new TableSchema();
         tableSchema.setDatabase(databaseName);
         tableSchema.setTable(tableName);
@@ -240,6 +254,7 @@ public void testCreateTableWhenDatabaseNotExists()
         Map<String, String> tableProperties = new HashMap<>();
         tableProperties.put("replication_num", "1");
         tableSchema.setProperties(tableProperties);
+
         schemaChangeManager.createTable(tableSchema);
 
         Thread.sleep(3_000);