Support config tableIdentifier for schema

ruanwenjun · ruanwenjun · commit 156cc3353b78 · 2023-10-17T12:56:04.000+08:00
diff --git a/docs/en/concept/schema-feature.md b/docs/en/concept/schema-feature.md
@@ -11,6 +11,9 @@ We can use SchemaOptions to define schema, the SchemaOptions contains some confi
 
 ```
 schema = {
+    table = "database.schema.table"
+    schema_first = false
+    comment = "comment"
     columns = [
     ...
     ]
@@ -24,6 +27,20 @@ schema = {
 }
 ```
 
+### table
+
+The table full name of the table identifier which the schema belongs to, it contains database, schema, table name. e.g. `database.schema.table`, `database.table`, `table`.
+
+### schema_first
+
+Default is false.
+
+If the schema_first is true, the schema will be used first, this means if we set `table = "a.b"`, `a` will be parsed as schema rather than database, then we can support write `table = "schema.table"`.
+
+### comment
+
+The comment of the CatalogTable which the schema belongs to.
+
 ### Columns
 
 Columns is a list of config used to define the column in schema, each column can contains name, type, nullable, defaultValue, comment field.
@@ -131,6 +148,7 @@ source {
     result_table_name = "fake"
     row.num = 16
     schema {
+        table = "FakeDatabase.FakeTable"
         columns = [
            {
               name = id
diff --git a/docs/en/connector-v2/source/FakeSource.md b/docs/en/connector-v2/source/FakeSource.md
@@ -265,7 +265,21 @@ The template list of double type that connector generated, if user configured it
 
 ### table-names
 
-The table list that connector generated, used to simulate multi-table scenarios
+The table list that connector generated, used to simulate multi-table scenarios.
+
+This option will override the `table` option in the `schema` option.
+For example, if you configure the `table-names` option as follows, the connector will generate data for the `test.table1` and `test.table2` tables, the `database.schema.table` will be drop.
+
+```agsl
+FakeSource {
+    table-names = ["test.table1", "test.table2"]
+    schema = {
+        table = "database.schema.table"
+        ...
+    }
+    ...
+}
+```
 
 ### common options
 
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java
@@ -48,7 +48,8 @@ public static CatalogTable of(
             Map<String, String> options,
             List<String> partitionKeys,
             String comment) {
-        return new CatalogTable(tableId, tableSchema, options, partitionKeys, comment);
+        return new CatalogTable(
+                tableId, tableSchema, options, partitionKeys, comment, tableId.getCatalogName());
     }
 
     public static CatalogTable of(
@@ -67,7 +68,7 @@ private CatalogTable(
             Map<String, String> options,
             List<String> partitionKeys,
             String comment) {
-        this(tableId, tableSchema, options, partitionKeys, comment, "");
+        this(tableId, tableSchema, options, partitionKeys, comment, tableId.getCatalogName());
     }
 
     private CatalogTable(
@@ -127,6 +128,9 @@ public String toString() {
                 + ", comment='"
                 + comment
                 + '\''
+                + ", catalogName='"
+                + catalogName
+                + '\''
                 + '}';
     }
 }
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java
@@ -31,6 +31,8 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
 
+import org.apache.commons.lang3.StringUtils;
+
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.Serializable;
@@ -135,7 +137,7 @@ public static List<CatalogTable> getCatalogTablesFromConfig(
             if (schemaMap.isEmpty()) {
                 throw new SeaTunnelException("Schema config can not be empty");
             }
-            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(readonlyConfig);
+            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(factoryId, readonlyConfig);
             return Collections.singletonList(catalogTable);
         }
 
@@ -190,6 +192,9 @@ public static SeaTunnelDataType<SeaTunnelRow> convertToDataType(
         }
     }
 
+    // We need to use buildWithConfig(String catalogName, ReadonlyConfig readonlyConfig);
+    // Since this method will not inject the correct catalogName into CatalogTable
+    @Deprecated
     public static List<CatalogTable> convertDataTypeToCatalogTables(
             SeaTunnelDataType<?> seaTunnelDataType, String tableId) {
         List<CatalogTable> catalogTables;
@@ -210,18 +215,42 @@ public static List<CatalogTable> convertDataTypeToCatalogTables(
     }
 
     public static CatalogTable buildWithConfig(ReadonlyConfig readonlyConfig) {
+        return buildWithConfig("", readonlyConfig);
+    }
+
+    public static CatalogTable buildWithConfig(String catalogName, ReadonlyConfig readonlyConfig) {
         if (readonlyConfig.get(TableSchemaOptions.SCHEMA) == null) {
             throw new RuntimeException(
                     "Schema config need option [schema], please correct your config first");
         }
         TableSchema tableSchema = new ReadonlyConfigParser().parse(readonlyConfig);
+
+        ReadonlyConfig schemaConfig =
+                readonlyConfig
+                        .getOptional(TableSchemaOptions.SCHEMA)
+                        .map(ReadonlyConfig::fromMap)
+                        .orElseThrow(
+                                () -> new IllegalArgumentException("Schema config can't be null"));
+
+        TablePath tablePath;
+        if (StringUtils.isNotEmpty(
+                schemaConfig.get(TableSchemaOptions.TableIdentifierOptions.TABLE))) {
+            tablePath =
+                    TablePath.of(
+                            schemaConfig.get(TableSchemaOptions.TableIdentifierOptions.TABLE),
+                            schemaConfig.get(
+                                    TableSchemaOptions.TableIdentifierOptions.SCHEMA_FIRST));
+        } else {
+            tablePath = TablePath.EMPTY;
+        }
+
         return CatalogTable.of(
-                // TODO: other table info
-                TableIdentifier.of("", "", ""),
+                TableIdentifier.of(catalogName, tablePath),
                 tableSchema,
                 new HashMap<>(),
+                // todo: add partitionKeys?
                 new ArrayList<>(),
-                "");
+                readonlyConfig.get(TableSchemaOptions.TableIdentifierOptions.COMMENT));
     }
 
     public static SeaTunnelRowType buildSimpleTextSchema() {
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/TablePath.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/TablePath.java
@@ -34,12 +34,26 @@ public final class TablePath implements Serializable {
     private final String schemaName;
     private final String tableName;
 
+    public static final TablePath EMPTY = TablePath.of(null, null, null);
+
     public static TablePath of(String fullName) {
+        return of(fullName, false);
+    }
+
+    public static TablePath of(String fullName, boolean schemaFirst) {
         String[] paths = fullName.split("\\.");
 
+        if (paths.length == 1) {
+            return of(null, paths[0]);
+        }
+
         if (paths.length == 2) {
-            return of(paths[0], paths[1]);
+            if (schemaFirst) {
+                return of(null, paths[0], paths[1]);
+            }
+            return of(paths[0], null, paths[1]);
         }
+
         if (paths.length == 3) {
             return of(paths[0], paths[1], paths[2]);
         }
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/schema/TableSchemaOptions.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/schema/TableSchemaOptions.java
@@ -28,6 +28,27 @@
 
 public class TableSchemaOptions {
 
+    public static class TableIdentifierOptions {
+
+        public static final Option<Boolean> SCHEMA_FIRST =
+                Options.key("schema_first")
+                        .booleanType()
+                        .defaultValue(false)
+                        .withDescription("Parse Schema First from table");
+
+        public static final Option<String> TABLE =
+                Options.key("table")
+                        .stringType()
+                        .noDefaultValue()
+                        .withDescription("SeaTunnel Schema Full Table Name");
+
+        public static final Option<String> COMMENT =
+                Options.key("comment")
+                        .stringType()
+                        .noDefaultValue()
+                        .withDescription("SeaTunnel Schema Table Comment");
+    }
+
     public static final Option<Map<String, Object>> SCHEMA =
             Options.key("schema")
                     .type(new TypeReference<Map<String, Object>>() {})
diff --git a/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/rule/AssertCatalogTableRule.java b/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/rule/AssertCatalogTableRule.java
@@ -22,6 +22,7 @@
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
 import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
 import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.connectors.seatunnel.assertion.exception.AssertConnectorException;
 
@@ -48,6 +49,9 @@ public class AssertCatalogTableRule implements Serializable {
     @OptionMark(description = "column rule")
     private AssertColumnRule columnRule;
 
+    @OptionMark(description = "tableIdentifier rule")
+    private AssertTableIdentifierRule tableIdentifierRule;
+
     public void checkRule(CatalogTable catalogTable) {
         TableSchema tableSchema = catalogTable.getTableSchema();
         if (tableSchema == null) {
@@ -62,6 +66,9 @@ public void checkRule(CatalogTable catalogTable) {
         if (columnRule != null) {
             columnRule.checkRule(tableSchema.getColumns());
         }
+        if (tableIdentifierRule != null) {
+            tableIdentifierRule.checkRule(catalogTable.getTableId());
+        }
     }
 
     @Data
@@ -138,4 +145,24 @@ public void checkRule(List<Column> check) {
             }
         }
     }
+
+    @Data
+    @AllArgsConstructor
+    public static class AssertTableIdentifierRule implements Serializable {
+
+        private TableIdentifier tableIdentifier;
+
+        public void checkRule(TableIdentifier actiualTableIdentifier) {
+            if (actiualTableIdentifier == null) {
+                throw new AssertConnectorException(CATALOG_TABLE_FAILED, "tableIdentifier is null");
+            }
+            if (!actiualTableIdentifier.equals(tableIdentifier)) {
+                throw new AssertConnectorException(
+                        CATALOG_TABLE_FAILED,
+                        String.format(
+                                "tableIdentifier: %s is not equal to %s",
+                                actiualTableIdentifier, tableIdentifier));
+            }
+        }
+    }
 }
diff --git a/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/rule/AssertCatalogTableRuleParser.java b/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/rule/AssertCatalogTableRuleParser.java
@@ -23,6 +23,8 @@
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
 import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
 import org.apache.seatunnel.api.table.catalog.SeaTunnelDataTypeConvertorUtil;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
 import org.apache.seatunnel.common.config.TypesafeConfigUtils;
 
@@ -46,6 +48,9 @@
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.PRIMARY_KEY_COLUMNS;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.PRIMARY_KEY_NAME;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.PRIMARY_KEY_RULE;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TableIdentifierRule.TABLE_IDENTIFIER_CATALOG_NAME;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TableIdentifierRule.TABLE_IDENTIFIER_RULE;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TableIdentifierRule.TABLE_IDENTIFIER_TABLE_NAME;
 
 public class AssertCatalogTableRuleParser {
 
@@ -55,6 +60,7 @@ public AssertCatalogTableRule parseCatalogTableRule(Config catalogTableRule) {
         parsePrimaryKeyRule(catalogTableRule).ifPresent(tableRule::setPrimaryKeyRule);
         parseConstraintKeyRule(catalogTableRule).ifPresent(tableRule::setConstraintKeyRule);
         parseColumnRule(catalogTableRule).ifPresent(tableRule::setColumnRule);
+        parseTableIdentifierRule(catalogTableRule).ifPresent(tableRule::setTableIdentifierRule);
         return tableRule;
     }
 
@@ -156,4 +162,17 @@ private Optional<AssertCatalogTableRule.AssertConstraintKeyRule> parseConstraint
                         .collect(Collectors.toList());
         return Optional.of(new AssertCatalogTableRule.AssertConstraintKeyRule(constraintKeys));
     }
+
+    private Optional<AssertCatalogTableRule.AssertTableIdentifierRule> parseTableIdentifierRule(
+            Config catalogTableRule) {
+        if (!catalogTableRule.hasPath(TABLE_IDENTIFIER_RULE)) {
+            return Optional.empty();
+        }
+        Config tableIdentifierRule = catalogTableRule.getConfig(TABLE_IDENTIFIER_RULE);
+        TableIdentifier tableIdentifier =
+                TableIdentifier.of(
+                        tableIdentifierRule.getString(TABLE_IDENTIFIER_CATALOG_NAME),
+                        TablePath.of(tableIdentifierRule.getString(TABLE_IDENTIFIER_TABLE_NAME)));
+        return Optional.of(new AssertCatalogTableRule.AssertTableIdentifierRule(tableIdentifier));
+    }
 }
diff --git a/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertConfig.java b/seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertConfig.java
@@ -64,6 +64,13 @@ public class AssertConfig {
     public static final String COLUMN_DEFAULT_VALUE = "default_value";
     public static final String COLUMN_COMMENT = "comment";
 
+    public static class TableIdentifierRule {
+        public static final String TABLE_IDENTIFIER_RULE = "table_identifier_rule";
+
+        public static final String TABLE_IDENTIFIER_CATALOG_NAME = "catalog_name";
+        public static final String TABLE_IDENTIFIER_TABLE_NAME = "table";
+    }
+
     public static final Option<String> COMMENT =
             Options.key("comment")
                     .stringType()
diff --git a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeConfig.java b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeConfig.java
@@ -416,7 +416,7 @@ public static FakeConfig buildWithConfig(Config config) {
             List<String> tableNames = config.getStringList(CatalogOptions.TABLE_NAMES.key());
             List<TableIdentifier> tableIdentifiers = new ArrayList<>(tableNames.size());
             for (String tableName : tableNames) {
-                tableIdentifiers.add(TableIdentifier.of("fake", TablePath.of(tableName)));
+                tableIdentifiers.add(TableIdentifier.of("FakeSource", TablePath.of(tableName)));
             }
             builder.tableIdentifiers(tableIdentifiers);
         }
diff --git a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert_with_catalogtable.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert_with_catalogtable.conf

Original file line number	Diff line number	Diff line change
`@@ -416,7 +416,7 @@ public static FakeConfig buildWithConfig(Config config) {`
`416`	`416`	`List<String> tableNames = config.getStringList(CatalogOptions.TABLE_NAMES.key());`
`417`	`417`	`List<TableIdentifier> tableIdentifiers = new ArrayList<>(tableNames.size());`
`418`	`418`	`for (String tableName : tableNames) {`
`419`		`- tableIdentifiers.add(TableIdentifier.of("fake", TablePath.of(tableName)));`
	`419`	`+ tableIdentifiers.add(TableIdentifier.of("FakeSource", TablePath.of(tableName)));`
`420`	`420`	`}`
`421`	`421`	`builder.tableIdentifiers(tableIdentifiers);`
`422`	`422`	`}`