[fix] Fix garbled of table or column comments contain Chinese charact…

…ers(apache#401) (apache#403)
hyperpaas · Jun 18, 2024 · 0acfbf2 · 0acfbf2
1 parent fb25484
commit 0acfbf2
Show file tree

Hide file tree

Showing 3 changed files with 76 additions and 3 deletions.
diff --git a/...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java b/...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
@@ -56,11 +56,17 @@ public class SchemaChangeManager implements Serializable {
     private static final String SCHEMA_CHANGE_API = "http://%s/api/query/default_cluster/%s";
     private ObjectMapper objectMapper = new ObjectMapper();
     private DorisOptions dorisOptions;
+    private String charsetEncoding = "UTF-8";
 
     public SchemaChangeManager(DorisOptions dorisOptions) {
         this.dorisOptions = dorisOptions;
     }
 
+    public SchemaChangeManager(DorisOptions dorisOptions, String charsetEncoding) {
+        this.dorisOptions = dorisOptions;
+        this.charsetEncoding = charsetEncoding;
+    }
+
     public boolean createTable(TableSchema table) throws IOException, IllegalArgumentException {
         String createTableDDL = DorisSystem.buildCreateTableDDL(table);
         return execute(createTableDDL, table.getDatabase());
@@ -133,7 +139,8 @@ public boolean checkSchemaChange(String database, String table, Map<String, Obje
                         table);
         HttpGetWithEntity httpGet = new HttpGetWithEntity(requestUrl);
         httpGet.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
-        httpGet.setEntity(new StringEntity(objectMapper.writeValueAsString(params)));
+        httpGet.setEntity(
+                new StringEntity(objectMapper.writeValueAsString(params), charsetEncoding));
         String responseEntity = "";
         Map<String, Object> responseMap = handleResponse(httpGet, responseEntity);
         return handleSchemaChange(responseMap, responseEntity);
@@ -173,8 +180,11 @@ public HttpPost buildHttpPost(String ddl, String database)
                         database);
         HttpPost httpPost = new HttpPost(requestUrl);
         httpPost.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
-        httpPost.setHeader(HttpHeaders.CONTENT_TYPE, "application/json");
-        httpPost.setEntity(new StringEntity(objectMapper.writeValueAsString(param)));
+        httpPost.setHeader(
+                HttpHeaders.CONTENT_TYPE,
+                String.format("application/json;charset=%s", charsetEncoding));
+        httpPost.setEntity(
+                new StringEntity(objectMapper.writeValueAsString(param), charsetEncoding));
         return httpPost;
     }
 

diff --git a/...doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java b/...doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java
@@ -28,8 +28,11 @@
 import java.io.IOException;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
 
 public class SchemaManagerITCase extends DorisTestBase {
 
@@ -82,6 +85,60 @@ public void testAddColumn() throws SQLException, IOException, IllegalArgumentExc
         Assert.assertTrue(exists);
     }
 
+    @Test
+    public void testAddColumnWithChineseComment()
+            throws SQLException, IOException, IllegalArgumentException {
+        String addColumnTbls = "add_column";
+        initDorisSchemaChangeTable(addColumnTbls);
+
+        // add a column by UTF-8 encoding
+        String addColumnName = "col_with_comment1";
+        String chineseComment = "中文注释1";
+        addColumnWithChineseCommentAndAssert(addColumnTbls, addColumnName, chineseComment, true);
+
+        // change charset encoding to US-ASCII would cause garbled of Chinese.
+        schemaChangeManager = new SchemaChangeManager(options, "US-ASCII");
+        addColumnName = "col_with_comment2";
+        chineseComment = "中文注释2";
+        addColumnWithChineseCommentAndAssert(addColumnTbls, addColumnName, chineseComment, false);
+    }
+
+    private void addColumnWithChineseCommentAndAssert(
+            String tableName, String addColumnName, String chineseComment, boolean assertFlag)
+            throws SQLException, IOException, IllegalArgumentException {
+        FieldSchema field = new FieldSchema(addColumnName, "string", chineseComment);
+        schemaChangeManager.addColumn(DATABASE, tableName, field);
+        boolean exists = schemaChangeManager.addColumn(DATABASE, tableName, field);
+        Assert.assertTrue(exists);
+
+        exists = schemaChangeManager.checkColumnExists(DATABASE, tableName, addColumnName);
+        Assert.assertTrue(exists);
+
+        // check Chinese comment
+        Map<String, String> columnComments = getColumnComments(tableName);
+        if (assertFlag) {
+            Assert.assertEquals(columnComments.get(addColumnName), chineseComment);
+        } else {
+            Assert.assertNotEquals(columnComments.get(addColumnName), chineseComment);
+        }
+    }
+
+    private Map<String, String> getColumnComments(String table) throws SQLException {
+        Map<String, String> columnCommentsMap = new HashMap<>();
+        try (Connection connection =
+                DriverManager.getConnection(
+                        String.format(URL, DORIS_CONTAINER.getHost()), USERNAME, PASSWORD)) {
+            ResultSet columns = connection.getMetaData().getColumns(null, DATABASE, table, null);
+
+            while (columns.next()) {
+                String columnName = columns.getString("COLUMN_NAME");
+                String comment = columns.getString("REMARKS");
+                columnCommentsMap.put(columnName, comment);
+            }
+        }
+        return columnCommentsMap;
+    }
+
     @Test
     public void testDropColumn() throws SQLException, IOException, IllegalArgumentException {
         String dropColumnTbls = "drop_column";

diff --git a/...k-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerTest.java b/...k-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerTest.java
@@ -132,6 +132,12 @@ public void testAddColumn() {
         Assert.assertEquals(
                 "ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT current_timestamp COMMENT 'comment \"\\'sdf\\''",
                 addColumnDDL);
+
+        field = new FieldSchema("col", "int", "current_timestamp", "中文注释");
+        addColumnDDL = SchemaChangeHelper.buildAddColumnDDL("test.test_flink", field);
+        Assert.assertEquals(
+                "ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT current_timestamp COMMENT '中文注释'",
+                addColumnDDL);
     }
 
     @Test