[Feature][transform-v2] SQL transform support schema change

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractSeaTunnelTransform.java

@@ -59,6 +59,12 @@ public CatalogTable getProducedCatalogTable() {
         return outputCatalogTable;
     }
 
+    public void restProducedCatalogTable() {
+        synchronized (this) {
+            outputCatalogTable = transformCatalogTable();
+        }
+    }
+
     @Override
     public List<CatalogTable> getProducedCatalogTables() {
         return Collections.singletonList(getProducedCatalogTable());

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLEngine.java

@@ -34,4 +34,6 @@ void init(
     List<SeaTunnelRow> transformBySQL(SeaTunnelRow inputRow, SeaTunnelRowType outputRowType);
 
     default void close() {}
+
+    void resetAllColumnsCount();
 }

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java

@@ -206,6 +206,13 @@ public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) {
                         inputCatalogTable.getOptions(),
                         inputCatalogTable.getPartitionKeys(),
                         inputCatalogTable.getComment());
+        sqlEngine.init(
+                inputTableName,
+                inputCatalogTable.getTableId().getTableName(),
+                inputCatalogTable.getSeaTunnelRowType(),
+                query);
+        sqlEngine.resetAllColumnsCount();
+        restProducedCatalogTable();
 
         if (event instanceof AlterTableColumnsEvent) {
             AlterTableColumnsEvent alterTableColumnsEvent = (AlterTableColumnsEvent) event;

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java

@@ -303,4 +303,9 @@ private int countColumnsSize(List<SelectItem<?>> selectItems) {
                         - allColumnsCnt;
         return allColumnsCount;
     }
+
+    @Override
+    public void resetAllColumnsCount() {
+        allColumnsCount = null;
+    }
 }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java

@@ -20,6 +20,7 @@
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
 import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
 import org.apache.seatunnel.api.table.catalog.PrimaryKey;
@@ -39,6 +40,8 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import com.google.common.collect.Lists;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -299,7 +302,7 @@ public void testSchemaChange() {
                         TableSchema.builder()
                                 .column(
                                         PhysicalColumn.of(
-                                                "f1",
+                                                "id",
                                                 BasicType.LONG_TYPE,
                                                 null,
                                                 null,
@@ -308,92 +311,201 @@ public void testSchemaChange() {
                                                 null))
                                 .column(
                                         PhysicalColumn.of(
-                                                "f2",
-                                                BasicType.LONG_TYPE,
+                                                "name",
+                                                BasicType.STRING_TYPE,
                                                 null,
                                                 null,
                                                 true,
                                                 null,
                                                 null))
                                 .column(
                                         PhysicalColumn.of(
-                                                "f3",
+                                                "age",
                                                 BasicType.LONG_TYPE,
                                                 null,
                                                 null,
                                                 true,
                                                 null,
                                                 null))
-                                .primaryKey(PrimaryKey.of("pk1", Arrays.asList("f1")))
+                                .primaryKey(PrimaryKey.of("pk1", Arrays.asList("id")))
                                 .constraintKey(
                                         ConstraintKey.of(
                                                 ConstraintKey.ConstraintType.UNIQUE_KEY,
                                                 "uk1",
                                                 Arrays.asList(
                                                         ConstraintKey.ConstraintKeyColumn.of(
-                                                                "f2",
+                                                                "name",
                                                                 ConstraintKey.ColumnSortType.ASC),
                                                         ConstraintKey.ConstraintKeyColumn.of(
-                                                                "f3",
+                                                                "age",
                                                                 ConstraintKey.ColumnSortType.ASC))))
                                 .build(),
                         Collections.emptyMap(),
-                        Collections.singletonList("f2"),
+                        Collections.singletonList("name"),
                         null);
 
+        ReadonlyConfig config =
+                ReadonlyConfig.fromMap(
+                        new HashMap<String, Object>() {
+                            {
+                                put("query", "select * from dual");
+                            }
+                        });
+        List<SeaTunnelRow> result;
+        SQLTransform transform = new SQLTransform(config, catalogTable);
+        result =
+                transform.transformRow(
+                        new SeaTunnelRow(
+                                new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)}));
+        List<String> columnNames;
+        List<String> columnType;
+        List<String> assertNames;
+        List<String> assertTypes;
+        Object[] columnValues;
+        Object[] assertValue;
+
+        columnNames =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map(Column::getName)
+                        .collect(Collectors.toList());
+        columnType =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map((e) -> e.getDataType().getSqlType().name())
+                        .collect(Collectors.toList());
+        assertNames = Lists.newArrayList("id", "name", "age");
+        assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT");
+
+        columnValues = result.get(0).getFields();
+        assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)};
+        Assertions.assertIterableEquals(columnNames, assertNames);
+        Assertions.assertIterableEquals(columnType, assertTypes);
+        Assertions.assertArrayEquals(columnValues, assertValue);
+
+        // test add column
         AlterTableAddColumnEvent addColumnEvent =
                 AlterTableAddColumnEvent.add(
                         catalogTable.getTableId(),
                         PhysicalColumn.of("f4", BasicType.LONG_TYPE, null, null, true, null, null));
+        transform.mapSchemaChangeEvent(addColumnEvent);
+
+        result =
+                transform.transformRow(
+                        new SeaTunnelRow(
+                                new Object[] {
+                                    Integer.valueOf(1),
+                                    "Cosmos",
+                                    Integer.valueOf(30),
+                                    Integer.valueOf(14)
+                                }));
+        columnNames =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map(Column::getName)
+                        .collect(Collectors.toList());
+        columnType =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map((e) -> e.getDataType().getSqlType().name())
+                        .collect(Collectors.toList());
+        assertNames = Lists.newArrayList("id", "name", "age", "f4");
+        assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "BIGINT");
+
+        columnValues = result.get(0).getFields();
+        assertValue =
+                new Object[] {
+                    Integer.valueOf(1), "Cosmos", Integer.valueOf(30), Integer.valueOf(14)
+                };
+        Assertions.assertIterableEquals(columnNames, assertNames);
+        Assertions.assertIterableEquals(columnType, assertTypes);
+        Assertions.assertArrayEquals(columnValues, assertValue);
+
+        // test modify column
         AlterTableModifyColumnEvent modifyColumnEvent =
                 AlterTableModifyColumnEvent.modify(
                         catalogTable.getTableId(),
-                        PhysicalColumn.of("f4", BasicType.INT_TYPE, null, null, true, null, null));
+                        PhysicalColumn.of(
+                                "f4", BasicType.STRING_TYPE, null, null, true, null, null));
+        transform.mapSchemaChangeEvent(modifyColumnEvent);
+        result =
+                transform.transformRow(
+                        new SeaTunnelRow(
+                                new Object[] {
+                                    Integer.valueOf(1), "Cosmos", Integer.valueOf(30), "Cosmos"
+                                }));
+        columnNames =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map(Column::getName)
+                        .collect(Collectors.toList());
+        columnType =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map((e) -> e.getDataType().getSqlType().name())
+                        .collect(Collectors.toList());
+        assertNames = Lists.newArrayList("id", "name", "age", "f4");
+        assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "STRING");
+
+        columnValues = result.get(0).getFields();
+        assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30), "Cosmos"};
+        Assertions.assertIterableEquals(columnNames, assertNames);
+        Assertions.assertIterableEquals(columnType, assertTypes);
+        Assertions.assertArrayEquals(columnValues, assertValue);
+
+        // test change column
         AlterTableChangeColumnEvent changeColumnEvent =
                 AlterTableChangeColumnEvent.change(
                         catalogTable.getTableId(),
                         "f4",
                         PhysicalColumn.of("f5", BasicType.INT_TYPE, null, null, true, null, null));
+        transform.mapSchemaChangeEvent(changeColumnEvent);
+        result =
+                transform.transformRow(
+                        new SeaTunnelRow(
+                                new Object[] {
+                                    Integer.valueOf(1),
+                                    "Cosmos",
+                                    Integer.valueOf(30),
+                                    Integer.valueOf(14)
+                                }));
+        columnNames =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map(Column::getName)
+                        .collect(Collectors.toList());
+        columnType =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map((e) -> e.getDataType().getSqlType().name())
+                        .collect(Collectors.toList());
+        assertNames = Lists.newArrayList("id", "name", "age", "f5");
+        assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "INT");
+
+        columnValues = result.get(0).getFields();
+        assertValue =
+                new Object[] {
+                    Integer.valueOf(1), "Cosmos", Integer.valueOf(30), Integer.valueOf(14)
+                };
+        Assertions.assertIterableEquals(columnNames, assertNames);
+        Assertions.assertIterableEquals(columnType, assertTypes);
+        Assertions.assertArrayEquals(columnValues, assertValue);
+
+        // test drop column
         AlterTableDropColumnEvent dropColumnEvent =
                 new AlterTableDropColumnEvent(catalogTable.getTableId(), "f5");
-        ReadonlyConfig config =
-                ReadonlyConfig.fromMap(
-                        new HashMap<String, Object>() {
-                            {
-                                put("query", "select * from dual");
-                            }
-                        });
+        transform.mapSchemaChangeEvent(dropColumnEvent);
+        result =
+                transform.transformRow(
+                        new SeaTunnelRow(
+                                new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)}));
+        columnNames =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map(Column::getName)
+                        .collect(Collectors.toList());
+        columnType =
+                transform.getProducedCatalogTable().getTableSchema().getColumns().stream()
+                        .map((e) -> e.getDataType().getSqlType().name())
+                        .collect(Collectors.toList());
+        assertNames = Lists.newArrayList("id", "name", "age");
+        assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT");
 
-        SQLTransform transform = new SQLTransform(config, catalogTable);
-        AlterTableAddColumnEvent outputAddEvent =
-                (AlterTableAddColumnEvent) transform.mapSchemaChangeEvent(addColumnEvent);
-        AlterTableModifyColumnEvent outputModifyEvent =
-                (AlterTableModifyColumnEvent) transform.mapSchemaChangeEvent(modifyColumnEvent);
-        AlterTableChangeColumnEvent outputChangeEvent =
-                (AlterTableChangeColumnEvent) transform.mapSchemaChangeEvent(changeColumnEvent);
-        AlterTableDropColumnEvent outputDropEvent =
-                (AlterTableDropColumnEvent) transform.mapSchemaChangeEvent(dropColumnEvent);
-        CatalogTable outputCatalogTable = transform.getProducedCatalogTable();
-        Assertions.assertIterableEquals(
-                Arrays.asList("f1", "f2", "f3"),
-                Arrays.asList(outputCatalogTable.getTableSchema().getFieldNames()));
-        Assertions.assertIterableEquals(
-                Arrays.asList("f1"),
-                outputCatalogTable.getTableSchema().getPrimaryKey().getColumnNames());
-        outputCatalogTable.getTableSchema().getConstraintKeys().stream()
-                .forEach(
-                        key ->
-                                Assertions.assertIterableEquals(
-                                        Arrays.asList("f2", "f3"),
-                                        key.getColumnNames().stream()
-                                                .map(
-                                                        ConstraintKey.ConstraintKeyColumn
-                                                                ::getColumnName)
-                                                .collect(Collectors.toList())));
-        Assertions.assertEquals("f4", outputAddEvent.getColumn().getName());
-        Assertions.assertEquals("f4", outputModifyEvent.getColumn().getName());
-        Assertions.assertEquals("f4", outputChangeEvent.getOldColumn());
-        Assertions.assertEquals("f5", outputChangeEvent.getColumn().getName());
-        Assertions.assertEquals("f5", outputDropEvent.getColumn());
+        columnValues = result.get(0).getFields();
+        assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)};
+        Assertions.assertIterableEquals(columnNames, assertNames);
+        Assertions.assertIterableEquals(columnType, assertTypes);
+        Assertions.assertArrayEquals(columnValues, assertValue);
     }
 }