fix Implements SchemaEvolution API

apache · Nov 24, 2024 · 84b59ba · 84b59ba
1 parent c586910
commit 84b59ba
Showing 1 changed file with 15 additions and 1 deletion.
diff --git a/...in/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java b/...in/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java
@@ -18,10 +18,13 @@
 package org.apache.seatunnel.connectors.seatunnel.starrocks.sink;
 
 import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
 import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
 import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
 import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
@@ -52,11 +55,13 @@
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 
 @Slf4j
-public class StarRocksSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
+public class StarRocksSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
+        implements SupportSchemaEvolutionSinkWriter, SupportSchemaEvolutionSink {
 
     private StarRocksISerializer serializer;
     private StarRocksSinkManager manager;
@@ -234,4 +239,13 @@ protected void replaceColumnByIndex(
             }
         }
     }
+
+    @Override
+    public List<SchemaChangeType> supports() {
+        return Arrays.asList(
+                SchemaChangeType.ADD_COLUMN,
+                SchemaChangeType.DROP_COLUMN,
+                SchemaChangeType.RENAME_COLUMN,
+                SchemaChangeType.UPDATE_COLUMN);
+    }
 }