[Fix][Connector-V2] Fixed doris/starrocks create table sql parse error (#6580)

Author: Hisoka-X

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/CreateTableParser.java

@@ -43,6 +43,9 @@ public static Map<String, ColumnInfo> getColumnList(String createTableSql) {
             } else if ((c == ',' || c == ')') && !insideParentheses) {
                 parseColumn(columnBuilder.toString(), columns, startIndex + i + 1);
                 columnBuilder.setLength(0);
+                if (c == ')') {
+                    break;
+                }
             } else if (c == ')') {
                 insideParentheses = false;
                 columnBuilder.append(c);

seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/catalog/DorisCreateTableTest.java

@@ -305,4 +305,45 @@ public void testWithVarchar() {
                         + "    \"dynamic_partition.prefix\" = \"p\"                                                                                                                                                                           \n"
                         + ");");
     }
+
+    @Test
+    public void testWithThreePrimaryKeys() {
+        List<Column> columns = new ArrayList<>();
+
+        columns.add(PhysicalColumn.of("id", BasicType.LONG_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("name", BasicType.STRING_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("age", BasicType.INT_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("comment", BasicType.STRING_TYPE, 500, true, null, ""));
+        columns.add(PhysicalColumn.of("description", BasicType.STRING_TYPE, 70000, true, null, ""));
+
+        String result =
+                DorisCatalogUtil.getCreateTableStatement(
+                        "create table '${database}'.'${table_name}'(\n"
+                                + "     ${rowtype_fields}\n"
+                                + " )\n"
+                                + " partitioned by ${rowtype_primary_key};",
+                        TablePath.of("test1", "test2"),
+                        CatalogTable.of(
+                                TableIdentifier.of("test", "test1", "test2"),
+                                TableSchema.builder()
+                                        .primaryKey(
+                                                PrimaryKey.of(
+                                                        "test", Arrays.asList("id", "age", "name")))
+                                        .columns(columns)
+                                        .build(),
+                                Collections.emptyMap(),
+                                Collections.emptyList(),
+                                ""));
+
+        Assertions.assertEquals(
+                "create table 'test1'.'test2'(\n"
+                        + "     `id` BIGINT(1) NULL ,\n"
+                        + "`name` STRING NULL ,\n"
+                        + "`age` INT(1) NULL ,\n"
+                        + "`comment` VARCHAR(500) NULL ,\n"
+                        + "`description` STRING NULL \n"
+                        + " )\n"
+                        + " partitioned by `id`,`age`,`name`;",
+                result);
+    }
 }

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/CreateTableParser.java

@@ -43,6 +43,9 @@ public static Map<String, ColumnInfo> getColumnList(String createTableSql) {
             } else if ((c == ',' || c == ')') && !insideParentheses) {
                 parseColumn(columnBuilder.toString(), columns, startIndex + i + 1);
                 columnBuilder.setLength(0);
+                if (c == ')') {
+                    break;
+                }
             } else if (c == ')') {
                 insideParentheses = false;
                 columnBuilder.append(c);

seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/catalog/StarRocksCreateTableTest.java

@@ -289,4 +289,40 @@ public void testWithVarchar() {
                         + ");",
                 result);
     }
+
+    @Test
+    public void testWithThreePrimaryKeys() {
+        List<Column> columns = new ArrayList<>();
+
+        columns.add(PhysicalColumn.of("id", BasicType.LONG_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("name", BasicType.STRING_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("age", BasicType.INT_TYPE, (Long) null, true, null, ""));
+        columns.add(PhysicalColumn.of("comment", BasicType.STRING_TYPE, 500, true, null, ""));
+        columns.add(PhysicalColumn.of("description", BasicType.STRING_TYPE, 70000, true, null, ""));
+
+        String result =
+                StarRocksSaveModeUtil.getCreateTableSql(
+                        "create table '${database}'.'${table_name}'(\n"
+                                + "     ${rowtype_fields}\n"
+                                + " )\n"
+                                + " partitioned by ${rowtype_primary_key};",
+                        "test1",
+                        "test2",
+                        TableSchema.builder()
+                                .primaryKey(
+                                        PrimaryKey.of("test", Arrays.asList("id", "age", "name")))
+                                .columns(columns)
+                                .build());
+
+        Assertions.assertEquals(
+                "create table 'test1'.'test2'(\n"
+                        + "     `id` BIGINT NULL ,\n"
+                        + "`name` STRING NULL ,\n"
+                        + "`age` INT NULL ,\n"
+                        + "`comment` VARCHAR(500) NULL ,\n"
+                        + "`description` STRING NULL \n"
+                        + " )\n"
+                        + " partitioned by `id`,`age`,`name`;",
+                result);
+    }
 }