[Improve][API] Unified tables_configs and table_list (#8100)

apache · Nov 25, 2024 · 84c0b8d · 84c0b8d
1 parent 3d3f7ba
commit 84c0b8d
Show file tree

Hide file tree

Showing 30 changed files with 295 additions and 116 deletions.
diff --git a/docs/en/concept/schema-feature.md b/docs/en/concept/schema-feature.md
@@ -172,6 +172,46 @@ constraintKeys = [
 | INDEX_KEY      | key         |
 | UNIQUE_KEY     | unique key  |
 
+## Multi table schemas
+
+```
+tables_configs = [
+  {
+    schema {
+      table = "database.schema.table1"
+      schema_first = false
+      comment = "comment"
+      columns = [
+        ...
+      ]
+      primaryKey {
+        ...
+      }
+      constraintKeys {
+        ...
+      }
+    }
+  },
+  {
+    schema = {
+      table = "database.schema.table2"
+      schema_first = false
+      comment = "comment"
+      columns = [
+        ...
+      ]
+      primaryKey {
+        ...
+      }
+      constraintKeys {
+        ...
+      }
+    }
+  }
+]
+
+```
+
 ## How to use schema
 
 ### Recommended

diff --git a/docs/en/connector-v2/source/Hive.md b/docs/en/connector-v2/source/Hive.md
@@ -120,6 +120,24 @@ Source plugin common parameters, please refer to [Source Common Options](../sour
 ```
 
 ### Example 2: Multiple tables
+> Note: Hive is a structured data source and should be use 'table_list', and 'tables_configs' will be removed in the future.
+
+```bash
+
+  Hive {
+    table_list = [
+        {
+          table_name = "default.seatunnel_orc_1"
+          metastore_uri = "thrift://namenode001:9083"
+        },
+        {
+          table_name = "default.seatunnel_orc_2"
+          metastore_uri = "thrift://namenode001:9083"
+        }
+    ]
+  }
+
+```
 
 ```bash
 

diff --git a/docs/en/connector-v2/source/kafka.md b/docs/en/connector-v2/source/kafka.md
@@ -189,6 +189,65 @@ source {
 
 > This is written to the same pg table according to different formats and topics of parsing kafka Perform upsert operations based on the id
 
+> Note: Kafka is an unstructured data source and should be use 'tables_configs', and 'table_list' will be removed in the future.
+
+```hocon
+
+env {
+  execution.parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  Kafka {
+    bootstrap.servers = "kafka_e2e:9092"
+    tables_configs = [
+      {
+        topic = "^test-ogg-sou.*"
+        pattern = "true"
+        consumer.group = "ogg_multi_group"
+        start_mode = earliest
+        schema = {
+          fields {
+            id = "int"
+            name = "string"
+            description = "string"
+            weight = "string"
+          }
+        },
+        format = ogg_json
+      },
+      {
+        topic = "test-cdc_mds"
+        start_mode = earliest
+        schema = {
+          fields {
+            id = "int"
+            name = "string"
+            description = "string"
+            weight = "string"
+          }
+        },
+        format = canal_json
+      }
+    ]
+  }
+}
+
+sink {
+  Jdbc {
+    driver = org.postgresql.Driver
+    url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
+    user = test
+    password = test
+    generate_sink_sql = true
+    database = test
+    table = public.sink
+    primary_keys = ["id"]
+  }
+}
+```
+
 ```hocon
 
 env {

diff --git a/docs/zh/concept/schema-feature.md b/docs/zh/concept/schema-feature.md
@@ -172,6 +172,46 @@ constraintKeys = [
 | INDEX_KEY  | 键   |
 | UNIQUE_KEY | 唯一键 |
 
+## 多表Schema
+
+```
+tables_configs = [
+  {
+    schema {
+      table = "database.schema.table1"
+      schema_first = false
+      comment = "comment"
+      columns = [
+        ...
+      ]
+      primaryKey {
+        ...
+      }
+      constraintKeys {
+        ...
+      }
+    }
+  },
+  {
+    schema = {
+      table = "database.schema.table2"
+      schema_first = false
+      comment = "comment"
+      columns = [
+        ...
+      ]
+      primaryKey {
+        ...
+      }
+      constraintKeys {
+        ...
+      }
+    }
+  }
+]
+
+```
+
 ## 如何使用schema
 
 ### 推荐

diff --git a/docs/zh/connector-v2/source/Kafka.md b/docs/zh/connector-v2/source/Kafka.md
@@ -181,6 +181,65 @@ source {
 
 > 根据不同的 Kafka 主题和格式解析数据，并基于 ID 执行 upsert 操作。
 
+> 注意: Kafka是一个非结构化数据源，应该使用`tables_configs`，将来会删除`table_list`
+
+```hocon
+
+env {
+  execution.parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  Kafka {
+    bootstrap.servers = "kafka_e2e:9092"
+    tables_configs = [
+      {
+        topic = "^test-ogg-sou.*"
+        pattern = "true"
+        consumer.group = "ogg_multi_group"
+        start_mode = earliest
+        schema = {
+          fields {
+            id = "int"
+            name = "string"
+            description = "string"
+            weight = "string"
+          }
+        },
+        format = ogg_json
+      },
+      {
+        topic = "test-cdc_mds"
+        start_mode = earliest
+        schema = {
+          fields {
+            id = "int"
+            name = "string"
+            description = "string"
+            weight = "string"
+          }
+        },
+        format = canal_json
+      }
+    ]
+  }
+}
+
+sink {
+  Jdbc {
+    driver = org.postgresql.Driver
+    url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
+    user = test
+    password = test
+    generate_sink_sql = true
+    database = test
+    table = public.sink
+    primary_keys = ["id"]
+  }
+}
+```
+
 ```hocon
 env {
   execution.parallelism = 1

diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogOptions.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogOptions.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.api.table.catalog;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
+
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
 
@@ -56,4 +58,12 @@ public interface CatalogOptions {
                     .withDescription(
                             "The table names RegEx of the database to capture."
                                     + "The table name needs to include the database name, for example: database_.*\\.table_.*");
+
+    Option<List<Map<String, Object>>> TABLE_LIST =
+            Options.key("table_list")
+                    .type(new TypeReference<List<Map<String, Object>>>() {})
+                    .noDefaultValue()
+                    .withDescription(
+                            "SeaTunnel Multi Table Schema, acts on structed data sources. "
+                                    + "such as jdbc, paimon, doris, etc");
 }
diff --git a/...l-api/src/main/java/org/apache/seatunnel/api/table/catalog/schema/TableSchemaOptions.java b/...l-api/src/main/java/org/apache/seatunnel/api/table/catalog/schema/TableSchemaOptions.java
@@ -55,6 +55,14 @@ public static class TableIdentifierOptions {
                     .noDefaultValue()
                     .withDescription("SeaTunnel Schema");
 
+    public static final Option<List<Map<String, Object>>> TABLE_CONFIGS =
+            Options.key("tables_configs")
+                    .type(new TypeReference<List<Map<String, Object>>>() {})
+                    .noDefaultValue()
+                    .withDescription(
+                            "SeaTunnel Multi Table Schema, acts on unstructed data sources. "
+                                    + "such as file, assert, mongodb, etc");
+
     // We should use ColumnOptions instead of FieldOptions
     @Deprecated
     public static class FieldOptions {

diff --git a/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertConfig.java b/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertConfig.java
@@ -22,7 +22,6 @@
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
 
-import java.util.List;
 import java.util.Map;
 
 public class AssertConfig {
@@ -85,13 +84,6 @@ public static class TableIdentifierRule {
                     .withDescription(
                             "Rule definition of user's available data. Each rule represents one field validation or row num validation.");
 
-    public static final Option<List<Map<String, Object>>> TABLE_CONFIGS =
-            Options.key("tables_configs")
-                    .type(new TypeReference<List<Map<String, Object>>>() {})
-                    .noDefaultValue()
-                    .withDescription(
-                            "Table configuration for the sink. Each table configuration contains the table name and the rules for the table.");
-
     public static final Option<String> TABLE_PATH =
             Options.key("table_path")
                     .stringType()

diff --git a/...rt/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSink.java b/...rt/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSink.java
@@ -42,11 +42,11 @@
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
+import static org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions.TABLE_CONFIGS;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.CATALOG_TABLE_RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.FIELD_RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.ROW_RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.RULES;
-import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TABLE_CONFIGS;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TABLE_PATH;
 
 public class AssertSink extends AbstractSimpleSink<SeaTunnelRow, Void>

diff --git a/...-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeOption.java b/...-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeOption.java
@@ -27,12 +27,6 @@
 
 public class FakeOption {
 
-    public static final Option<List<Map<String, Object>>> TABLES_CONFIGS =
-            Options.key("tables_configs")
-                    .type(new TypeReference<List<Map<String, Object>>>() {})
-                    .noDefaultValue()
-                    .withDescription("The multiple table config list of fake source");
-
     public static final Option<List<Map<String, Object>>> ROWS =
             Options.key("rows")
                     .type(new TypeReference<List<Map<String, Object>>>() {})

diff --git a/.../org/apache/seatunnel/connectors/seatunnel/fake/config/MultipleTableFakeSourceConfig.java b/.../org/apache/seatunnel/connectors/seatunnel/fake/config/MultipleTableFakeSourceConfig.java
@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.fake.config;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
 
 import org.apache.commons.collections4.CollectionUtils;
 
@@ -36,7 +37,7 @@ public class MultipleTableFakeSourceConfig implements Serializable {
     @Getter private List<FakeConfig> fakeConfigs;
 
     public MultipleTableFakeSourceConfig(ReadonlyConfig fakeSourceRootConfig) {
-        if (fakeSourceRootConfig.getOptional(FakeOption.TABLES_CONFIGS).isPresent()) {
+        if (fakeSourceRootConfig.getOptional(TableSchemaOptions.TABLE_CONFIGS).isPresent()) {
             parseFromConfigs(fakeSourceRootConfig);
         } else {
             parseFromConfig(fakeSourceRootConfig);
@@ -56,7 +57,7 @@ public MultipleTableFakeSourceConfig(ReadonlyConfig fakeSourceRootConfig) {
 
     private void parseFromConfigs(ReadonlyConfig readonlyConfig) {
         List<ReadonlyConfig> readonlyConfigs =
-                readonlyConfig.getOptional(FakeOption.TABLES_CONFIGS).get().stream()
+                readonlyConfig.getOptional(TableSchemaOptions.TABLE_CONFIGS).get().stream()
                         .map(ReadonlyConfig::fromMap)
                         .collect(Collectors.toList());
         // Use the config outside if it's not set in sub config

diff --git a/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java b/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java
@@ -54,7 +54,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.SPLIT_READ_INTERVAL;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.STRING_FAKE_MODE;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.STRING_TEMPLATE;
-import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.TABLES_CONFIGS;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.TIME_HOUR_TEMPLATE;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.TIME_MINUTE_TEMPLATE;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.TIME_SECOND_TEMPLATE;
@@ -72,7 +71,7 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .optional(TABLES_CONFIGS)
+                .optional(TableSchemaOptions.TABLE_CONFIGS)
                 .optional(TableSchemaOptions.SCHEMA)
                 .optional(STRING_FAKE_MODE)
                 .conditional(STRING_FAKE_MODE, FakeOption.FakeMode.TEMPLATE, STRING_TEMPLATE)