[BitSail][MultiTableSInk]Bitsail enabled write multi sink in single task.

bitsail-base/src/main/java/com/bytedance/bitsail/base/catalog/TableCatalogFactory.java → bitsail-base/src/main/java/com/bytedance/bitsail/base/catalog/CatalogFactory.java

@@ -18,25 +18,22 @@
 
 import com.bytedance.bitsail.base.component.ComponentBuilder;
 import com.bytedance.bitsail.base.connector.BuilderGroup;
-import com.bytedance.bitsail.base.execution.ExecutionEnviron;
-import com.bytedance.bitsail.common.catalog.table.TableCatalog;
+import com.bytedance.bitsail.common.catalog.table.Catalog;
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 
 import java.io.Serializable;
 
 /**
  * Created 2022/5/23
  */
-public interface TableCatalogFactory extends Serializable, ComponentBuilder {
+public interface CatalogFactory extends Serializable, ComponentBuilder {
 
   /**
    * Create a table catalog.
    *
-   * @param executionEnviron       execution environment
    * @param connectorConfiguration configuration for the reader/writer
    */
-  TableCatalog createTableCatalog(BuilderGroup builderGroup,
-                                  ExecutionEnviron executionEnviron,
-                                  BitSailConfiguration connectorConfiguration);
+  Catalog createTableCatalog(BuilderGroup builderGroup,
+                             BitSailConfiguration connectorConfiguration);
 
 }

bitsail-base/src/main/java/com/bytedance/bitsail/base/catalog/TableCatalogFactoryHelper.java → bitsail-base/src/main/java/com/bytedance/bitsail/base/catalog/CatalogFactoryHelper.java

@@ -20,11 +20,11 @@
 
 import org.apache.commons.lang3.StringUtils;
 
-public class TableCatalogFactoryHelper {
+public class CatalogFactoryHelper {
 
-  public static TableCatalogFactory getTableCatalogFactory(String connectorName) {
-    DefaultComponentBuilderLoader<TableCatalogFactory> loader =
-        new DefaultComponentBuilderLoader<>(TableCatalogFactory.class);
+  public static CatalogFactory getTableCatalogFactory(String connectorName) {
+    DefaultComponentBuilderLoader<CatalogFactory> loader =
+        new DefaultComponentBuilderLoader<>(CatalogFactory.class);
 
     return loader.loadComponent(StringUtils.lowerCase(connectorName), false);
   }

bitsail-base/src/main/java/com/bytedance/bitsail/base/extension/SupportMultipleSinkTable.java

@@ -0,0 +1,37 @@
+/*
+ *       Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ *       Licensed under the Apache License, Version 2.0 (the "License");
+ *       you may not use this file except in compliance with the License.
+ *       You may obtain a copy of the License at
+ *
+ *           http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *       Unless required by applicable law or agreed to in writing, software
+ *       distributed under the License is distributed on an "AS IS" BASIS,
+ *       WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *       See the License for the specific language governing permissions and
+ *       limitations under the License.
+ */
+
+package com.bytedance.bitsail.base.extension;
+
+import com.bytedance.bitsail.base.connector.writer.v1.Writer;
+import com.bytedance.bitsail.base.connector.writer.v1.WriterCommitter;
+import com.bytedance.bitsail.common.catalog.table.TableId;
+import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
+
+import java.io.Serializable;
+import java.util.Optional;
+
+public interface SupportMultipleSinkTable<InputT, CommitT extends Serializable, WriterStateT extends Serializable> {
+
+  Writer<InputT, CommitT, WriterStateT> createWriter(Writer.Context<WriterStateT> context,
+                                                     BitSailConfiguration templateConfiguration);
+
+  Optional<WriterCommitter<CommitT>> createCommitter(BitSailConfiguration templateConfiguration);
+
+  BitSailConfiguration applyTableId(BitSailConfiguration template,
+                                    TableId tableId);
+
+}

bitsail-base/src/main/java/com/bytedance/bitsail/base/extension/SupportProducedType.java

@@ -0,0 +1,26 @@
+/*
+ *       Copyright 2022-2023 Bytedance Ltd. and/or its affiliates.
+ *
+ *       Licensed under the Apache License, Version 2.0 (the "License");
+ *       you may not use this file except in compliance with the License.
+ *       You may obtain a copy of the License at
+ *
+ *           http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *       Unless required by applicable law or agreed to in writing, software
+ *       distributed under the License is distributed on an "AS IS" BASIS,
+ *       WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *       See the License for the specific language governing permissions and
+ *       limitations under the License.
+ */
+
+package com.bytedance.bitsail.base.extension;
+
+import com.bytedance.bitsail.common.typeinfo.RowTypeInfo;
+
+import java.io.Serializable;
+
+public interface SupportProducedType extends Serializable {
+
+  RowTypeInfo getProducedType();
+}

bitsail-common/pom.xml

@@ -31,7 +31,6 @@
 
     <properties>
         <maven.deploy.skip>false</maven.deploy.skip>
-        <main.basedir>${project.parent.basedir}</main.basedir>
     </properties>
 
     <dependencies>

bitsail-common/src/main/java/com/bytedance/bitsail/common/catalog/TableCatalogManager.java

@@ -17,12 +17,12 @@
 package com.bytedance.bitsail.common.catalog;
 
 import com.bytedance.bitsail.common.BitSailException;
+import com.bytedance.bitsail.common.catalog.table.Catalog;
 import com.bytedance.bitsail.common.catalog.table.CatalogTable;
 import com.bytedance.bitsail.common.catalog.table.CatalogTableAlterDefinition;
 import com.bytedance.bitsail.common.catalog.table.CatalogTableColumn;
-import com.bytedance.bitsail.common.catalog.table.CatalogTableDefinition;
 import com.bytedance.bitsail.common.catalog.table.CatalogTableSchema;
-import com.bytedance.bitsail.common.catalog.table.TableCatalog;
+import com.bytedance.bitsail.common.catalog.table.TableId;
 import com.bytedance.bitsail.common.catalog.table.TableOperation;
 import com.bytedance.bitsail.common.configuration.BitSailConfiguration;
 import com.bytedance.bitsail.common.model.ColumnInfo;
@@ -50,8 +50,8 @@ public class TableCatalogManager {
   private final TypeInfoConverter readerTypeInfoConverter;
   private final TypeInfoConverter writerTypeInfoConverter;
 
-  private final TableCatalog readerTableCatalog;
-  private final TableCatalog writerTableCatalog;
+  private final Catalog readerCatalog;
+  private final Catalog writerCatalog;
 
   private final BitSailConfiguration commonConfiguration;
   private final BitSailConfiguration readerConfiguration;
@@ -73,15 +73,15 @@ public class TableCatalogManager {
   @Builder
   public TableCatalogManager(TypeInfoConverter readerTypeInfoConverter,
                              TypeInfoConverter writerTypeInfoConverter,
-                             TableCatalog readerTableCatalog,
-                             TableCatalog writerTableCatalog,
+                             Catalog readerCatalog,
+                             Catalog writerCatalog,
                              BitSailConfiguration commonConfiguration,
                              BitSailConfiguration readerConfiguration,
                              BitSailConfiguration writerConfiguration) {
     this.readerTypeInfoConverter = readerTypeInfoConverter;
     this.writerTypeInfoConverter = writerTypeInfoConverter;
-    this.readerTableCatalog = readerTableCatalog;
-    this.writerTableCatalog = writerTableCatalog;
+    this.readerCatalog = readerCatalog;
+    this.writerCatalog = writerCatalog;
     this.commonConfiguration = commonConfiguration;
     this.readerConfiguration = readerConfiguration;
     this.writerConfiguration = writerConfiguration;
@@ -102,7 +102,7 @@ private void prepareCatalogManager() {
   }
 
   public void alignmentCatalogTable() throws Exception {
-    if (Objects.isNull(readerTableCatalog) || Objects.isNull(writerTableCatalog)) {
+    if (Objects.isNull(readerCatalog) || Objects.isNull(writerCatalog)) {
       return;
     }
 
@@ -114,28 +114,28 @@ public void alignmentCatalogTable() throws Exception {
     startTableCatalog();
 
     try {
-      CatalogTableDefinition readerTableDefinition = readerTableCatalog.createCatalogTableDefinition();
-      CatalogTableDefinition writerTableDefinition = readerTableCatalog.createCatalogTableDefinition();
-      if (!readerTableCatalog.tableExists(readerTableDefinition)) {
+      TableId readerTableDefinition = readerCatalog.createCatalogTableDefinition();
+      TableId writerTableDefinition = readerCatalog.createCatalogTableDefinition();
+      if (!readerCatalog.tableExists(readerTableDefinition)) {
         throw BitSailException.asBitSailException(TableCatalogErrorCode.TABLE_CATALOG_TABLE_NOT_EXISTS,
             String.format("Reader table definition %s not exists.", readerTableDefinition));
       }
 
       // get reader catalog table.
-      readerCatalogTable = readerTableCatalog.getCatalogTable(readerTableDefinition);
+      readerCatalogTable = readerCatalog.getCatalogTable(readerTableDefinition);
 
-      if (!writerTableCatalog.tableExists(writerTableDefinition)) {
+      if (!writerCatalog.tableExists(writerTableDefinition)) {
 
         if (!tableCatalogCreateTableNotExists) {
           throw BitSailException.asBitSailException(TableCatalogErrorCode.TABLE_CATALOG_TABLE_NOT_EXISTS,
               String.format("Writer table definition %s not exists.", writerTableDefinition));
         }
         // try to create table when not exists.
-        writerTableCatalog.createTable(writerTableDefinition, readerCatalogTable);
+        writerCatalog.createTable(writerTableDefinition, readerCatalogTable);
       }
 
       // get writer catalog table.
-      writerCatalogTable = writerTableCatalog.getCatalogTable(writerTableDefinition);
+      writerCatalogTable = writerCatalog.getCatalogTable(writerTableDefinition);
 
       // get base table schema.
       CatalogTableSchema catalogTableSchema = tableCatalogStrategy
@@ -190,7 +190,7 @@ private void alterCatalogSchema(CatalogTableAlterDefinition catalogTableAlterDef
         CollectionUtils.isNotEmpty(catalogTableAlterDefinition.getPendingAddColumns())) {
       LOG.info("Writer catalog table {} try to add column: {}.", writerCatalogTable,
           catalogTableAlterDefinition.getPendingAddColumns());
-      writerTableCatalog.alterTableColumns(
+      writerCatalog.alterTableColumns(
           TableOperation.ALTER_COLUMNS_ADD,
           catalogTableAlterDefinition.getPendingAddColumns()
       );
@@ -200,7 +200,7 @@ private void alterCatalogSchema(CatalogTableAlterDefinition catalogTableAlterDef
         CollectionUtils.isNotEmpty(catalogTableAlterDefinition.getPendingUpdateColumns())) {
       LOG.info("Writer catalog table {} try to update column: {}.", writerCatalogTable,
           catalogTableAlterDefinition.getPendingUpdateColumns());
-      writerTableCatalog.alterTableColumns(
+      writerCatalog.alterTableColumns(
           TableOperation.ALTER_COLUMNS_UPDATE,
           catalogTableAlterDefinition.getPendingUpdateColumns()
       );
@@ -210,7 +210,7 @@ private void alterCatalogSchema(CatalogTableAlterDefinition catalogTableAlterDef
         CollectionUtils.isNotEmpty(catalogTableAlterDefinition.getPendingDeleteColumns())) {
       LOG.info("Writer catalog table {} try to delete column: {}.", writerCatalogTable,
           catalogTableAlterDefinition.getPendingDeleteColumns());
-      writerTableCatalog.alterTableColumns(
+      writerCatalog.alterTableColumns(
           TableOperation.ALTER_COLUMNS_DELETE,
           catalogTableAlterDefinition.getPendingDeleteColumns()
       );
@@ -262,7 +262,7 @@ private CatalogTableAlterDefinition calNecessaryCatalogSchema(CatalogTableSchema
         TypeInfo<?> baseTypeInfo = catalogTableColumn.getType();
 
         finalCatalogColumns.add(catalogTableColumn);
-        if (!writerTableCatalog.compareTypeCompatible(writerTypeInfo, baseTypeInfo)) {
+        if (!writerCatalog.compareTypeCompatible(writerTypeInfo, baseTypeInfo)) {
           pendingUpdateTableColumns.add(catalogTableColumn);
         }
       } else {
@@ -279,13 +279,13 @@ private CatalogTableAlterDefinition calNecessaryCatalogSchema(CatalogTableSchema
   }
 
   private void startTableCatalog() {
-    readerTableCatalog.open(readerTypeInfoConverter);
-    writerTableCatalog.open(writerTypeInfoConverter);
+    readerCatalog.open(readerTypeInfoConverter);
+    writerCatalog.open(writerTypeInfoConverter);
   }
 
   private void closeTableCatalog() {
-    readerTableCatalog.close();
-    writerTableCatalog.close();
+    readerCatalog.close();
+    writerCatalog.close();
   }
 
 }

bitsail-common/src/main/java/com/bytedance/bitsail/common/catalog/table/TableCatalog.java → bitsail-common/src/main/java/com/bytedance/bitsail/common/catalog/table/Catalog.java

@@ -28,7 +28,7 @@
 /**
  * Table catalog only for the signal table for now.
  */
-public interface TableCatalog extends Serializable {
+public interface Catalog extends Serializable {
 
   /**
    * Open Table catalog
@@ -43,22 +43,27 @@ public interface TableCatalog extends Serializable {
   /**
    * Get the reference table for the table catalog.
    */
-  CatalogTableDefinition createCatalogTableDefinition();
+  TableId createCatalogTableDefinition();
+
+  /**
+   * List table under the database
+   */
+  List<TableId> listTables();
 
   /**
    * Check the table exits or not.
    */
-  boolean tableExists(CatalogTableDefinition catalogTableDefinition);
+  boolean tableExists(TableId catalogTableDefinition);
 
   /**
    * Acquire catalog table by the table definition.
    */
-  CatalogTable getCatalogTable(CatalogTableDefinition catalogTableDefinition);
+  CatalogTable getCatalogTable(TableId catalogTableDefinition);
 
   /**
    * Create table
    */
-  void createTable(CatalogTableDefinition catalogTableDefinition,
+  void createTable(TableId catalogTableDefinition,
                    CatalogTable catalogTable);
 
   /**

bitsail-common/src/main/java/com/bytedance/bitsail/common/catalog/table/CatalogTable.java

@@ -25,29 +25,29 @@
 @Builder
 public class CatalogTable implements Serializable {
 
-  private CatalogTableDefinition catalogTableDefinition;
+  private TableId tableId;
 
   private final CatalogTableSchema catalogTableSchema;
 
   private final String comment;
 
-  public CatalogTable(CatalogTableDefinition catalogTableDefinition,
+  public CatalogTable(TableId tableId,
                       CatalogTableSchema catalogTableSchema) {
-    this(catalogTableDefinition, catalogTableSchema, null);
+    this(tableId, catalogTableSchema, null);
   }
 
-  public CatalogTable(CatalogTableDefinition catalogTableDefinition,
+  public CatalogTable(TableId tableId,
                       CatalogTableSchema catalogTableSchema,
                       String comment) {
-    this.catalogTableDefinition = catalogTableDefinition;
+    this.tableId = tableId;
     this.catalogTableSchema = catalogTableSchema;
     this.comment = comment;
   }
 
   @Override
   public String toString() {
     return "CatalogTable{" +
-        "catalogTableDefinition=" + catalogTableDefinition +
+        "catalogTableDefinition=" + tableId +
         ", catalogTableSchema=" + catalogTableSchema +
         ", comment='" + comment + '\'' +
         '}';

bitsail-common/src/main/java/com/bytedance/bitsail/common/catalog/table/CatalogTableColumn.java

@@ -37,6 +37,8 @@ public class CatalogTableColumn implements Serializable {
 
   private String comment;
 
+  private Object defaultValue;
+
   public CatalogTableColumn(String name, TypeInfo<?> type) {
     this.name = name;
     this.type = type;