[Improve] update clickhouse connector, use factory to create source/sink

liunaijie · Oct 30, 2024 · da7120e · da7120e
1 parent bdffbcc
commit da7120e
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSink.java b/...java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSink.java
@@ -21,6 +21,7 @@
 import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ReaderOption;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKAggCommitInfo;
@@ -35,9 +36,11 @@ public class ClickhouseSink
         implements SeaTunnelSink<SeaTunnelRow, ClickhouseSinkState, CKCommitInfo, CKAggCommitInfo> {
 
     private ReaderOption option;
+    private CatalogTable catalogTable;
 
-    public ClickhouseSink(ReaderOption option) {
+    public ClickhouseSink(ReaderOption option, CatalogTable catalogTable) {
         this.option = option;
+        this.catalogTable = catalogTable;
     }
 
     @Override
@@ -61,4 +64,9 @@ public SinkWriter<SeaTunnelRow, CKCommitInfo, ClickhouseSinkState> restoreWriter
     public Optional<Serializer<ClickhouseSinkState>> getWriterStateSerializer() {
         return Optional.of(new DefaultSerializer<>());
     }
+
+    @Override
+    public Optional<CatalogTable> getWriteCatalogTable() {
+        return Optional.of(catalogTable);
+    }
 }
diff --git a/...g/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java b/...g/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java
@@ -138,7 +138,7 @@ public TableSink<SeaTunnelRow, ClickhouseSinkState, CKCommitInfo, CKAggCommitInf
                             .supportUpsert(supportUpsert)
                             .allowExperimentalLightweightDelete(allowExperimentalLightweightDelete)
                             .build();
-            return () -> new ClickhouseSink(option);
+            return () -> new ClickhouseSink(option, catalogTable);
         } finally {
             proxy.close();
         }