Support config tableIdentifier for schema

ruanwenjun · ruanwenjun · commit 5fb8b01b7c45 · 2023-10-16T12:12:56.000+08:00
diff --git a/docs/en/concept/schema-feature.md b/docs/en/concept/schema-feature.md
@@ -11,6 +11,8 @@ We can use SchemaOptions to define schema, the SchemaOptions contains some confi
 
 ```
 schema = {
+    table = "database.schema.table"
+    comment = "comment"
     columns = [
     ...
     ]
@@ -24,6 +26,14 @@ 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`.
+
+### 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 +141,7 @@ source {
     result_table_name = "fake"
     row.num = 16
     schema {
+        table = "FakeDatabase.FakeTable"
         columns = [
            {
               name = id
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,39 @@ 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));
+        } 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,6 +34,8 @@ 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) {
         String[] paths = fullName.split("\\.");
 
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,21 @@
 
 public class TableSchemaOptions {
 
+    public static class TableIdentifierOptions {
+
+        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
@@ -30,6 +30,7 @@
 import org.apache.seatunnel.api.source.SupportParallelism;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
 import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -41,6 +42,8 @@
 import org.apache.seatunnel.connectors.seatunnel.fake.exception.FakeConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.fake.state.FakeSourceState;
 
+import org.apache.commons.collections4.CollectionUtils;
+
 import com.google.auto.service.AutoService;
 import com.google.common.collect.Lists;
 
@@ -61,7 +64,7 @@ public class FakeSource
     public FakeSource() {}
 
     public FakeSource(ReadonlyConfig readonlyConfig) {
-        this.catalogTable = CatalogTableUtil.buildWithConfig(readonlyConfig);
+        this.catalogTable = CatalogTableUtil.buildWithConfig(getPluginName(), readonlyConfig);
         this.fakeConfig = FakeConfig.buildWithConfig(readonlyConfig.toConfig());
     }
 
@@ -74,20 +77,23 @@ public Boundedness getBoundedness() {
 
     @Override
     public List<CatalogTable> getProducedCatalogTables() {
-        if (fakeConfig.getTableIdentifiers().isEmpty()) {
+        // If tableNames is empty, means this is only one catalogTable, return the original
+        // catalogTable
+        if (CollectionUtils.isEmpty(fakeConfig.getTableIdentifiers())) {
             return Lists.newArrayList(catalogTable);
-        } else {
-            return fakeConfig.getTableIdentifiers().stream()
-                    .map(
-                            tableIdentifier ->
-                                    CatalogTable.of(
-                                            tableIdentifier,
-                                            catalogTable.getTableSchema(),
-                                            catalogTable.getOptions(),
-                                            catalogTable.getPartitionKeys(),
-                                            catalogTable.getComment()))
-                    .collect(Collectors.toList());
         }
+        // Otherwise, return the catalogTables with the tableNames
+        return fakeConfig.getTableIdentifiers().stream()
+                .map(
+                        tableIdentifier ->
+                                CatalogTable.of(
+                                        TableIdentifier.of(
+                                                getPluginName(), tableIdentifier.toTablePath()),
+                                        catalogTable.getTableSchema(),
+                                        catalogTable.getOptions(),
+                                        catalogTable.getPartitionKeys(),
+                                        catalogTable.getComment()))
+                .collect(Collectors.toList());
     }
 
     @Override
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
@@ -24,6 +24,7 @@ source {
   FakeSource {
     row.num = 100
     schema = {
+      table = "test.fakeTable"
       columns = [
         {
             name = id
@@ -63,6 +64,11 @@ sink{
   Assert {
       rules {
         catalog_table_rule {
+            table_identifier_rule = {
+                catalog_name = "FakeSource"
+                table = "test.fakeTable"
+            }
+
             primary_key_rule = {
                 primary_key_name = "primary key"
                 primary_key_columns = ["id"]

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`	`}`