This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 784a6a17 [improve]Improve the way of dorisSystem create table (#452)
784a6a17 is described below
commit 784a6a171b234b4a9f64e0d2b310995d172b1bf6
Author: wudongliang <46414265+donglian...@users.noreply.github.com>
AuthorDate: Tue Aug 6 19:17:52 2024 +0800
[improve]Improve the way of dorisSystem create table (#452)
---
.../flink/catalog/doris/DorisSchemaFactory.java | 132 +++++++++++++++++++++
.../doris/flink/catalog/doris/DorisSystem.java | 125 ++-----------------
.../doris/flink/sink/copy/CopySQLBuilder.java | 4 +-
.../flink/sink/schema/SchemaChangeHelper.java | 30 ++---
.../catalog/doris/DorisSchemaFactoryTest.java | 111 ++++++++++++-----
5 files changed, 237 insertions(+), 165 deletions(-)
diff --git
a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactory.java
b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactory.java
index 3f824a1f..9693d433 100644
---
a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactory.java
+++
b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactory.java
@@ -18,9 +18,12 @@
package org.apache.doris.flink.catalog.doris;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
+import org.apache.doris.flink.exception.CreateTableException;
import org.apache.doris.flink.tools.cdc.DorisTableConfig;
import java.util.ArrayList;
@@ -30,6 +33,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
/**
* Factory that creates doris schema.
@@ -103,4 +107,132 @@ public class DorisSchemaFactory {
}
return null;
}
+
+ public static String generateCreateTableDDL(TableSchema schema) {
+ StringBuilder sb = new StringBuilder("CREATE TABLE IF NOT EXISTS ");
+ sb.append(identifier(schema.getDatabase()))
+ .append(".")
+ .append(identifier(schema.getTable()))
+ .append("(");
+
+ Map<String, FieldSchema> fields = schema.getFields();
+ List<String> keys = schema.getKeys();
+ // append keys
+ for (String key : keys) {
+ if (!fields.containsKey(key)) {
+ throw new CreateTableException("key " + key + " not found in
column list");
+ }
+ FieldSchema field = fields.get(key);
+ buildColumn(sb, field, true);
+ }
+
+ // append values
+ for (Map.Entry<String, FieldSchema> entry : fields.entrySet()) {
+ if (keys.contains(entry.getKey())) {
+ continue;
+ }
+ FieldSchema field = entry.getValue();
+ buildColumn(sb, field, false);
+ }
+ sb = sb.deleteCharAt(sb.length() - 1);
+ sb.append(" ) ");
+ // append uniq model
+ if (DataModel.UNIQUE.equals(schema.getModel())) {
+ sb.append(schema.getModel().name())
+ .append(" KEY(")
+ .append(String.join(",", identifier(schema.getKeys())))
+ .append(")");
+ }
+
+ // append table comment
+ if (!StringUtils.isNullOrWhitespaceOnly(schema.getTableComment())) {
+ sb.append(" COMMENT
'").append(quoteComment(schema.getTableComment())).append("' ");
+ }
+
+ // append distribute key
+ sb.append(" DISTRIBUTED BY HASH(")
+ .append(String.join(",",
identifier(schema.getDistributeKeys())))
+ .append(")");
+
+ Map<String, String> properties = schema.getProperties();
+ if (schema.getTableBuckets() != null) {
+
+ int bucketsNum = schema.getTableBuckets();
+ if (bucketsNum <= 0) {
+ throw new CreateTableException("The number of buckets must be
positive.");
+ }
+ sb.append(" BUCKETS ").append(bucketsNum);
+ } else {
+ sb.append(" BUCKETS AUTO ");
+ }
+ // append properties
+ int index = 0;
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ if (index == 0) {
+ sb.append(" PROPERTIES (");
+ }
+ if (index > 0) {
+ sb.append(",");
+ }
+ sb.append(quoteProperties(entry.getKey()))
+ .append("=")
+ .append(quoteProperties(entry.getValue()));
+ index++;
+
+ if (index == (schema.getProperties().size())) {
+ sb.append(")");
+ }
+ }
+ return sb.toString();
+ }
+
+ private static void buildColumn(StringBuilder sql, FieldSchema field,
boolean isKey) {
+ String fieldType = field.getTypeString();
+ if (isKey && DorisType.STRING.equals(fieldType)) {
+ fieldType = String.format("%s(%s)", DorisType.VARCHAR, 65533);
+ }
+ sql.append(identifier(field.getName())).append(" ").append(fieldType);
+
+ if (field.getDefaultValue() != null) {
+ sql.append(" DEFAULT " +
quoteDefaultValue(field.getDefaultValue()));
+ }
+ sql.append(" COMMENT
'").append(quoteComment(field.getComment())).append("',");
+ }
+
+ private static String quoteProperties(String name) {
+ return "'" + name + "'";
+ }
+
+ private static List<String> identifier(List<String> names) {
+ return
names.stream().map(DorisSchemaFactory::identifier).collect(Collectors.toList());
+ }
+
+ public static String identifier(String name) {
+ if (name.startsWith("`") && name.endsWith("`")) {
+ return name;
+ }
+ return "`" + name + "`";
+ }
+
+ public static String quoteDefaultValue(String defaultValue) {
+ // DEFAULT current_timestamp not need quote
+ if (defaultValue.equalsIgnoreCase("current_timestamp")) {
+ return defaultValue;
+ }
+ return "'" + defaultValue + "'";
+ }
+
+ public static String quoteComment(String comment) {
+ if (comment == null) {
+ return "";
+ } else {
+ return comment.replaceAll("'", "\\\\'");
+ }
+ }
+
+ public static String quoteTableIdentifier(String tableIdentifier) {
+ String[] dbTable = tableIdentifier.split("\\.");
+ Preconditions.checkArgument(dbTable.length == 2);
+ return identifier(dbTable[0]) + "." + identifier(dbTable[1]);
+ }
}
diff --git
a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
index be6572d3..427eb8b3 100644
---
a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
+++
b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/doris/DorisSystem.java
@@ -18,14 +18,12 @@
package org.apache.doris.flink.catalog.doris;
import org.apache.flink.annotation.Public;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;
import org.apache.commons.compress.utils.Lists;
import org.apache.doris.flink.cfg.DorisConnectionOptions;
import org.apache.doris.flink.connection.JdbcConnectionProvider;
import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
-import org.apache.doris.flink.exception.CreateTableException;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.exception.DorisSystemException;
import org.slf4j.Logger;
@@ -41,7 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Predicate;
-import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -141,81 +138,7 @@ public class DorisSystem implements Serializable {
}
public static String buildCreateTableDDL(TableSchema schema) {
- StringBuilder sb = new StringBuilder("CREATE TABLE IF NOT EXISTS ");
- sb.append(identifier(schema.getDatabase()))
- .append(".")
- .append(identifier(schema.getTable()))
- .append("(");
-
- Map<String, FieldSchema> fields = schema.getFields();
- List<String> keys = schema.getKeys();
- // append keys
- for (String key : keys) {
- if (!fields.containsKey(key)) {
- throw new CreateTableException("key " + key + " not found in
column list");
- }
- FieldSchema field = fields.get(key);
- buildColumn(sb, field, true);
- }
-
- // append values
- for (Map.Entry<String, FieldSchema> entry : fields.entrySet()) {
- if (keys.contains(entry.getKey())) {
- continue;
- }
- FieldSchema field = entry.getValue();
- buildColumn(sb, field, false);
- }
- sb = sb.deleteCharAt(sb.length() - 1);
- sb.append(" ) ");
- // append uniq model
- if (DataModel.UNIQUE.equals(schema.getModel())) {
- sb.append(schema.getModel().name())
- .append(" KEY(")
- .append(String.join(",", identifier(schema.getKeys())))
- .append(")");
- }
-
- // append table comment
- if (!StringUtils.isNullOrWhitespaceOnly(schema.getTableComment())) {
- sb.append(" COMMENT
'").append(quoteComment(schema.getTableComment())).append("' ");
- }
-
- // append distribute key
- sb.append(" DISTRIBUTED BY HASH(")
- .append(String.join(",",
identifier(schema.getDistributeKeys())))
- .append(")");
-
- Map<String, String> properties = schema.getProperties();
- if (schema.getTableBuckets() != null) {
-
- int bucketsNum = schema.getTableBuckets();
- if (bucketsNum <= 0) {
- throw new CreateTableException("The number of buckets must be
positive.");
- }
- sb.append(" BUCKETS ").append(bucketsNum);
- } else {
- sb.append(" BUCKETS AUTO ");
- }
- // append properties
- int index = 0;
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- if (index == 0) {
- sb.append(" PROPERTIES (");
- }
- if (index > 0) {
- sb.append(",");
- }
- sb.append(quoteProperties(entry.getKey()))
- .append("=")
- .append(quoteProperties(entry.getValue()));
- index++;
-
- if (index == (schema.getProperties().size())) {
- sb.append(")");
- }
- }
- return sb.toString();
+ return DorisSchemaFactory.generateCreateTableDDL(schema);
}
public Map<String, String> getTableFieldNames(String databaseName, String
tableName) {
@@ -244,53 +167,23 @@ public class DorisSystem implements Serializable {
}
}
- private static void buildColumn(StringBuilder sql, FieldSchema field,
boolean isKey) {
- String fieldType = field.getTypeString();
- if (isKey && DorisType.STRING.equals(fieldType)) {
- fieldType = String.format("%s(%s)", DorisType.VARCHAR, 65533);
- }
- sql.append(identifier(field.getName())).append(" ").append(fieldType);
-
- if (field.getDefaultValue() != null) {
- sql.append(" DEFAULT " +
quoteDefaultValue(field.getDefaultValue()));
- }
- sql.append(" COMMENT
'").append(quoteComment(field.getComment())).append("',");
- }
-
+ @Deprecated
public static String quoteDefaultValue(String defaultValue) {
- // DEFAULT current_timestamp not need quote
- if (defaultValue.equalsIgnoreCase("current_timestamp")) {
- return defaultValue;
- }
- return "'" + defaultValue + "'";
+ return DorisSchemaFactory.quoteDefaultValue(defaultValue);
}
+ @Deprecated
public static String quoteComment(String comment) {
- if (comment == null) {
- return "";
- } else {
- return comment.replaceAll("'", "\\\\'");
- }
- }
-
- private static List<String> identifier(List<String> name) {
- return
name.stream().map(DorisSystem::identifier).collect(Collectors.toList());
+ return DorisSchemaFactory.quoteComment(comment);
}
+ @Deprecated
public static String identifier(String name) {
- if (name.startsWith("`") && name.endsWith("`")) {
- return name;
- }
- return "`" + name + "`";
+ return DorisSchemaFactory.identifier(name);
}
+ @Deprecated
public static String quoteTableIdentifier(String tableIdentifier) {
- String[] dbTable = tableIdentifier.split("\\.");
- Preconditions.checkArgument(dbTable.length == 2);
- return identifier(dbTable[0]) + "." + identifier(dbTable[1]);
- }
-
- private static String quoteProperties(String name) {
- return "'" + name + "'";
+ return DorisSchemaFactory.quoteTableIdentifier(tableIdentifier);
}
}
diff --git
a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/copy/CopySQLBuilder.java
b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/copy/CopySQLBuilder.java
index 33843a06..8edfba3f 100644
---
a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/copy/CopySQLBuilder.java
+++
b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/copy/CopySQLBuilder.java
@@ -17,7 +17,7 @@
package org.apache.doris.flink.sink.copy;
-import org.apache.doris.flink.catalog.doris.DorisSystem;
+import org.apache.doris.flink.catalog.doris.DorisSchemaFactory;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import java.util.Arrays;
@@ -53,7 +53,7 @@ public class CopySQLBuilder {
public String buildCopySQL() {
StringBuilder sb = new StringBuilder();
sb.append("COPY INTO ")
- .append(DorisSystem.quoteTableIdentifier(tableIdentifier))
+
.append(DorisSchemaFactory.quoteTableIdentifier(tableIdentifier))
.append(" FROM @~('{")
.append(String.join(",", fileList))
.append("}') ")
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 06546877..74b57417 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
@@ -20,7 +20,7 @@ package org.apache.doris.flink.sink.schema;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.compress.utils.Lists;
import org.apache.commons.lang3.StringUtils;
-import org.apache.doris.flink.catalog.doris.DorisSystem;
+import org.apache.doris.flink.catalog.doris.DorisSchemaFactory;
import org.apache.doris.flink.catalog.doris.FieldSchema;
import java.util.List;
@@ -114,11 +114,11 @@ public class SchemaChangeHelper {
new StringBuilder(
String.format(
ADD_DDL,
-
DorisSystem.quoteTableIdentifier(tableIdentifier),
- DorisSystem.identifier(name),
+
DorisSchemaFactory.quoteTableIdentifier(tableIdentifier),
+ DorisSchemaFactory.identifier(name),
type));
if (defaultValue != null) {
- addDDL.append(" DEFAULT
").append(DorisSystem.quoteDefaultValue(defaultValue));
+ addDDL.append(" DEFAULT
").append(DorisSchemaFactory.quoteDefaultValue(defaultValue));
}
commentColumn(addDDL, comment);
return addDDL.toString();
@@ -127,17 +127,17 @@ public class SchemaChangeHelper {
public static String buildDropColumnDDL(String tableIdentifier, String
columName) {
return String.format(
DROP_DDL,
- DorisSystem.quoteTableIdentifier(tableIdentifier),
- DorisSystem.identifier(columName));
+ DorisSchemaFactory.quoteTableIdentifier(tableIdentifier),
+ DorisSchemaFactory.identifier(columName));
}
public static String buildRenameColumnDDL(
String tableIdentifier, String oldColumnName, String
newColumnName) {
return String.format(
RENAME_DDL,
- DorisSystem.quoteTableIdentifier(tableIdentifier),
- DorisSystem.identifier(oldColumnName),
- DorisSystem.identifier(newColumnName));
+ DorisSchemaFactory.quoteTableIdentifier(tableIdentifier),
+ DorisSchemaFactory.identifier(oldColumnName),
+ DorisSchemaFactory.identifier(newColumnName));
}
public static String buildColumnExistsQuery(String database, String table,
String column) {
@@ -156,9 +156,9 @@ public class SchemaChangeHelper {
String tableIdentifier, String columnName, String newComment) {
return String.format(
MODIFY_COMMENT_DDL,
- DorisSystem.quoteTableIdentifier(tableIdentifier),
- DorisSystem.identifier(columnName),
- DorisSystem.quoteComment(newComment));
+ DorisSchemaFactory.quoteTableIdentifier(tableIdentifier),
+ DorisSchemaFactory.identifier(columnName),
+ DorisSchemaFactory.quoteComment(newComment));
}
public static String buildModifyColumnDataTypeDDL(
@@ -170,8 +170,8 @@ public class SchemaChangeHelper {
new StringBuilder(
String.format(
MODIFY_TYPE_DDL,
-
DorisSystem.quoteTableIdentifier(tableIdentifier),
- DorisSystem.identifier(columnName),
+
DorisSchemaFactory.quoteTableIdentifier(tableIdentifier),
+ DorisSchemaFactory.identifier(columnName),
dataType));
commentColumn(modifyDDL, comment);
return modifyDDL.toString();
@@ -179,7 +179,7 @@ public class SchemaChangeHelper {
private static void commentColumn(StringBuilder ddl, String comment) {
if (StringUtils.isNotEmpty(comment)) {
- ddl.append(" COMMENT
'").append(DorisSystem.quoteComment(comment)).append("'");
+ ddl.append(" COMMENT
'").append(DorisSchemaFactory.quoteComment(comment)).append("'");
}
}
diff --git
a/flink-doris-connector/src/test/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactoryTest.java
b/flink-doris-connector/src/test/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactoryTest.java
index 1bc1f115..a0f3aaf0 100644
---
a/flink-doris-connector/src/test/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactoryTest.java
+++
b/flink-doris-connector/src/test/java/org/apache/doris/flink/catalog/doris/DorisSchemaFactoryTest.java
@@ -46,6 +46,13 @@ public class DorisSchemaFactoryTest {
@Test
public void testCreateTableSchema() {
+ TableSchema tableSchema = buildCreateTableSchema();
+ Assert.assertEquals(
+ "TableSchema{database='doris', table='create_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=emai [...]
+ tableSchema.toString());
+ }
+
+ private TableSchema buildCreateTableSchema() {
String dorisTable = "doris.create_tab";
String[] dbTable = dorisTable.split("\\.");
Preconditions.checkArgument(dbTable.length == 2);
@@ -58,21 +65,24 @@ public class DorisSchemaFactoryTest {
List<String> pkKeys = Collections.singletonList("email");
Map<String, String> tableProperties = new HashMap<>();
String tableComment = "auto_tab_comment";
- TableSchema tableSchema =
- DorisSchemaFactory.createTableSchema(
- dbTable[0],
- dbTable[1],
- columnFields,
- pkKeys,
- new DorisTableConfig(tableProperties),
- tableComment);
- Assert.assertEquals(
- "TableSchema{database='doris', table='create_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=emai [...]
- tableSchema.toString());
+ return DorisSchemaFactory.createTableSchema(
+ dbTable[0],
+ dbTable[1],
+ columnFields,
+ pkKeys,
+ new DorisTableConfig(tableProperties),
+ tableComment);
}
@Test
public void testCreateTableSchemaTableBuckets() {
+ TableSchema tableSchema = buildCreateTableSchemaTableBuckets();
+ Assert.assertEquals(
+ "TableSchema{database='doris', table='create_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=emai [...]
+ tableSchema.toString());
+ }
+
+ private TableSchema buildCreateTableSchemaTableBuckets() {
String dorisTable = "doris.create_tab";
String[] dbTable = dorisTable.split("\\.");
Preconditions.checkArgument(dbTable.length == 2);
@@ -87,21 +97,24 @@ public class DorisSchemaFactoryTest {
tableProperties.put("table-buckets", "create_tab:40, create_taba:10,
tabs:12");
tableProperties.put("replication_num", "2");
String tableComment = "auto_tab_comment";
- TableSchema tableSchema =
- DorisSchemaFactory.createTableSchema(
- dbTable[0],
- dbTable[1],
- columnFields,
- pkKeys,
- new DorisTableConfig(tableProperties),
- tableComment);
- Assert.assertEquals(
- "TableSchema{database='doris', table='create_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=emai [...]
- tableSchema.toString());
+ return DorisSchemaFactory.createTableSchema(
+ dbTable[0],
+ dbTable[1],
+ columnFields,
+ pkKeys,
+ new DorisTableConfig(tableProperties),
+ tableComment);
}
@Test
public void testCreateDuplicateTableSchema() {
+ TableSchema tableSchema = buildCreateDuplicateTableSchema();
+ Assert.assertEquals(
+ "TableSchema{database='doris', table='dup_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=name, m [...]
+ tableSchema.toString());
+ }
+
+ private TableSchema buildCreateDuplicateTableSchema() {
String dorisTable = "doris.dup_tab";
String[] dbTable = dorisTable.split("\\.");
Preconditions.checkArgument(dbTable.length == 2);
@@ -114,16 +127,50 @@ public class DorisSchemaFactoryTest {
Map<String, String> tableProperties = new HashMap<>();
tableProperties.put("replication_num", "1");
String tableComment = "auto_tab_comment";
- TableSchema tableSchema =
- DorisSchemaFactory.createTableSchema(
- dbTable[0],
- dbTable[1],
- columnFields,
- new ArrayList<>(),
- new DorisTableConfig(tableProperties),
- tableComment);
+ return DorisSchemaFactory.createTableSchema(
+ dbTable[0],
+ dbTable[1],
+ columnFields,
+ new ArrayList<>(),
+ new DorisTableConfig(tableProperties),
+ tableComment);
+ }
+
+ @Test
+ public void testGenerateCreateTableDDL() {
+ TableSchema tableSchema = buildCreateTableSchema();
+ String ddl = DorisSchemaFactory.generateCreateTableDDL(tableSchema);
Assert.assertEquals(
- "TableSchema{database='doris', table='dup_tab',
tableComment='auto_tab_comment', fields={name=FieldSchema{name='name',
typeString='VARVHAR(100)', defaultValue='null', comment='Name_test'},
id=FieldSchema{name='id', typeString='INT', defaultValue='100',
comment='int_test'}, age=FieldSchema{name='age', typeString='INT',
defaultValue='null', comment='null'}, email=FieldSchema{name='email',
typeString='VARCHAR(100)', defaultValue='em...@doris.com', comment='e'}},
keys=name, m [...]
- tableSchema.toString());
+ "CREATE TABLE IF NOT EXISTS `doris`.`create_tab`(`email`
VARCHAR(100) DEFAULT 'em...@doris.com' COMMENT 'e',`name` VARVHAR(100) COMMENT
'Name_test',`id` INT DEFAULT '100' COMMENT 'int_test',`age` INT COMMENT '' )
UNIQUE KEY(`email`) COMMENT 'auto_tab_comment' DISTRIBUTED BY HASH(`email`)
BUCKETS AUTO PROPERTIES ('light_schema_change'='true')",
+ ddl);
+ }
+
+ @Test
+ public void testGenerateCreateTableDDLBuckets() {
+ TableSchema tableSchema = buildCreateTableSchemaTableBuckets();
+ String ddl = DorisSchemaFactory.generateCreateTableDDL(tableSchema);
+ Assert.assertEquals(
+ "CREATE TABLE IF NOT EXISTS `doris`.`create_tab`(`email`
VARCHAR(100) DEFAULT 'em...@doris.com' COMMENT 'e',`name` VARVHAR(100) COMMENT
'Name_test',`id` INT DEFAULT '100' COMMENT 'int_test',`age` INT COMMENT '' )
UNIQUE KEY(`email`) COMMENT 'auto_tab_comment' DISTRIBUTED BY HASH(`email`)
BUCKETS 40 PROPERTIES ('replication_num'='2','light_schema_change'='true')",
+ ddl);
+ }
+
+ @Test
+ public void testGenerateCreateTableDDLDuplicate() {
+ TableSchema tableSchema = buildCreateDuplicateTableSchema();
+ String ddl = DorisSchemaFactory.generateCreateTableDDL(tableSchema);
+ Assert.assertEquals(
+ "CREATE TABLE IF NOT EXISTS `doris`.`dup_tab`(`name`
VARVHAR(100) COMMENT 'Name_test',`id` INT DEFAULT '100' COMMENT
'int_test',`age` INT COMMENT '',`email` VARCHAR(100) DEFAULT 'em...@doris.com'
COMMENT 'e' ) COMMENT 'auto_tab_comment' DISTRIBUTED BY HASH(`name`) BUCKETS
AUTO PROPERTIES ('replication_num'='1','light_schema_change'='true')",
+ ddl);
+ }
+
+ @Test
+ public void quoteTableIdentifier() {
+ String quoted = DorisSchemaFactory.quoteTableIdentifier("db.tbl");
+ Assert.assertEquals("`db`.`tbl`", quoted);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void quoteTableIdentifierException() {
+ DorisSchemaFactory.quoteTableIdentifier("db.tbl.sc");
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
