[Fix][Connector-V2] Support schema evolution in Redis sink

docs/en/connectors/sink/Redis.md

@@ -96,6 +96,25 @@ Replica count for multi-table sink writers. It applies when upstream rows carry
 For multi-table jobs, `key` may include `${table_name}` so rows from different upstream tables are written to separate
 Redis keys, for example `key = "redis-result-${table_name}"`.
 
+## Schema Evolution
+
+Redis Sink supports schema evolution with SeaTunnel Zeta. When the upstream is a CDC source, enable
+`schema-changes.enabled = true` in the source configuration so schema change events are sent to the sink.
+
+Redis is schema-less, so schema evolution does not execute DDL in Redis. Instead, when Redis Sink serializes the whole
+upstream row as JSON or TEXT, it refreshes the serializer after a supported schema change event. Newly added fields are
+included, and dropped fields are no longer written. See
+[Schema Evolution](../../introduction/configuration/schema-evolution.md) for the supported event types.
+
+Schema evolution does not rewrite field names configured in `key`, custom key placeholders, `value_field`,
+`hash_key_field`, or `hash_value_field`. Do not rename or drop a field referenced by these options while the job is
+running. If a configured field no longer exists, Redis Sink applies the missing-field behavior described in
+[Write Rules](#write-rules), which can turn the configured field name into a literal key or value.
+
+Before applying a schema change, Redis Sink flushes rows buffered with the previous schema. It also stores the latest
+schema in checkpoint state and restores that schema after recovery. Restoring a job from a checkpoint taken after a DDL
+while increasing the Redis sink parallelism is not currently supported.
+
 ## Examples
 
 ### Write Rows To A Redis List

docs/en/introduction/configuration/schema-evolution.md

@@ -28,6 +28,7 @@ Schema Evolution means that the schema of a data table can be changed and the da
 [Doris](../../connectors/sink/Doris.md)
 [Paimon](../../connectors/sink/Paimon.md#schema-evolution)
 [Elasticsearch](../../connectors/sink/Elasticsearch.md#schema-evolution)
+[Redis](../../connectors/sink/Redis.md#schema-evolution)
 
 Note:  
 * The schema evolution is not support the transform at now. The schema evolution of different types of databases（Oracle-CDC -> Jdbc-Mysql）is currently not supported the default value of the column in ddl.

docs/zh/connectors/sink/Redis.md

@@ -93,6 +93,22 @@ Redis 接收器连接器可以在批处理或流处理作业中把上游数据
 多表作业中，`key` 可以包含 `${table_name}`，这样不同上游表的数据会写入不同 Redis key，例如
 `key = "redis-result-${table_name}"`。
 
+## 模式演变
+
+Redis Sink 在 SeaTunnel Zeta 引擎中支持模式演变。上游使用 CDC Source 时，需要在 Source 配置中设置
+`schema-changes.enabled = true`，使模式变更事件能够发送到 Sink。
+
+Redis 本身没有表结构，因此模式演变不会在 Redis 中执行 DDL。当 Redis Sink 使用 JSON 或 TEXT
+序列化完整上游行时，它会在收到支持的模式变更事件后更新序列化器：新增字段会写入 Redis，已删除字段不再写入。
+支持的事件类型请参见[模式演变](../../introduction/configuration/schema-evolution.md)。
+
+模式演变不会自动改写 `key`、自定义 Key 占位符、`value_field`、`hash_key_field` 或 `hash_value_field` 中配置的
+字段名。作业运行期间，请勿重命名或删除这些选项引用的字段。如果配置的字段已不存在，Redis Sink 会按
+[写入规则](#写入规则)中的缺失字段行为处理，配置的字段名可能会被当作字面量 Key 或 Value 写入。
+
+应用模式变更前，Redis Sink 会先刷新按旧模式缓存的数据。它还会把最新模式保存到 checkpoint 状态，并在恢复时使用
+该模式。当前不支持从包含 DDL 后状态的 checkpoint 恢复作业时增加 Redis Sink 的并行度。
+
 ## 示例
 
 ### 写入 Redis List

docs/zh/introduction/configuration/schema-evolution.md

@@ -28,6 +28,7 @@
 [Doris](../../connectors/sink/Doris.md)
 [Paimon](../../connectors/sink/Paimon.md#模式演变)
 [Elasticsearch](../../connectors/sink/Elasticsearch.md#模式演变)
+[Redis](../../connectors/sink/Redis.md#模式演变)
 
 注意: 
 * 目前模式演进不支持transform。不同类型数据库(Oracle-CDC -> Jdbc-Mysql)的模式演进目前不支持ddl中列的默认值。

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java

@@ -18,30 +18,36 @@
 package org.apache.seatunnel.connectors.seatunnel.redis.sink;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.serialization.DefaultSerializer;
+import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
 import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions;
 import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 
-public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, Void>
-        implements SupportMultiTableSink {
+public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, TableSchema>
+        implements SupportMultiTableSink, SupportSchemaEvolutionSink {
     private final RedisParameters redisParameters = new RedisParameters();
-    private final SeaTunnelRowType seaTunnelRowType;
+    private final TableSchema tableSchema;
     private final ReadonlyConfig readonlyConfig;
     private final CatalogTable catalogTable;
 
     public RedisSink(ReadonlyConfig config, CatalogTable table) {
         this.readonlyConfig = config;
         this.catalogTable = table;
         this.redisParameters.buildWithConfig(config);
-        this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
+        this.tableSchema = catalogTable.getTableSchema();
     }
 
     @Override
@@ -51,11 +57,56 @@ public String getPluginName() {
 
     @Override
     public RedisSinkWriter createWriter(SinkWriter.Context context) throws IOException {
-        return new RedisSinkWriter(seaTunnelRowType, redisParameters);
+        return new RedisSinkWriter(tableSchema, redisParameters);
+    }
+
+    /**
+     * Restores the latest writer schema. Rescaling can provide state from multiple writers, and
+     * every state must contain the same schema because selecting the widest schema would restore a
+     * stale definition after a drop-column event.
+     */
+    @Override
+    public RedisSinkWriter restoreWriter(SinkWriter.Context context, List<TableSchema> states)
+            throws IOException {
+        if (states == null || states.isEmpty()) {
+            return createWriter(context);
+        }
+        TableSchema restoredSchema = states.get(0);
+        for (int stateIndex = 1; stateIndex < states.size(); stateIndex++) {
+            TableSchema state = states.get(stateIndex);
+            if (!restoredSchema.equals(state)) {
+                throw new IOException(
+                        String.format(
+                                "Redis sink cannot restore writer for table %s because state 0 fields %s differ from state %d fields %s",
+                                catalogTable.getTablePath().getFullName(),
+                                schemaFields(restoredSchema),
+                                stateIndex,
+                                schemaFields(state)));
+            }
+        }
+        return new RedisSinkWriter(restoredSchema, redisParameters);
+    }
+
+    @Override
+    public Optional<Serializer<TableSchema>> getWriterStateSerializer() {
+        return Optional.of(new DefaultSerializer<>());
     }
 
     @Override
     public Optional<CatalogTable> getWriteCatalogTable() {
         return Optional.ofNullable(catalogTable);
     }
+
+    @Override
+    public List<SchemaChangeType> supports() {
+        return Arrays.asList(
+                SchemaChangeType.ADD_COLUMN,
+                SchemaChangeType.DROP_COLUMN,
+                SchemaChangeType.RENAME_COLUMN,
+                SchemaChangeType.UPDATE_COLUMN);
+    }
+
+    private static String schemaFields(TableSchema schema) {
+        return Arrays.toString(schema.toPhysicalRowDataType().getFieldNames());
+    }
 }

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java

@@ -22,7 +22,13 @@
 import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 import org.apache.seatunnel.api.serialization.SerializationSchema;
 import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
+import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
 import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.constants.PluginType;
@@ -43,6 +49,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -51,27 +58,51 @@
 import java.util.regex.Pattern;
 
 @Slf4j
-public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
-        implements SupportMultiTableSinkWriter<Void> {
+public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, TableSchema>
+        implements SupportMultiTableSinkWriter<Void>, SupportSchemaEvolutionSinkWriter {
     private static final Pattern LEGACY_PLACEHOLDER_PATTERN =
             Pattern.compile("(?<!\\$)\\{([^{}]+)\\}");
     private static final Pattern PLACEHOLDER_PATTERN = Pattern.compile("\\$\\{([^}]+)\\}");
-    private final SeaTunnelRowType seaTunnelRowType;
+
+    /** Authoritative schema saved in checkpoints and updated by ordered schema-change events. */
+    private TableSchema tableSchema;
+
+    /** Physical row type derived from {@link #tableSchema} for key and value field lookup. */
+    private SeaTunnelRowType seaTunnelRowType;
+
     private final RedisParameters redisParameters;
-    private final SerializationSchema serializationSchema;
+
+    /** Serializer derived from {@link #tableSchema} and refreshed together with the row type. */
+    private SerializationSchema serializationSchema;
+
     private final RedisClient redisClient;
+    private final TableSchemaChangeEventDispatcher schemaChangeEventDispatcher;
 
     private final int batchSize;
 
     private final List<RowKind> rowKinds;
     private final List<String> keyBuffer;
     private final List<String> valueBuffer;
 
+    /**
+     * Creates a writer from a physical row type.
+     *
+     * @deprecated Use {@link #RedisSinkWriter(TableSchema, RedisParameters)} so schema evolution
+     *     and checkpoint recovery retain the complete table schema.
+     */
+    @Deprecated
     public RedisSinkWriter(SeaTunnelRowType seaTunnelRowType, RedisParameters redisParameters) {
-        this.seaTunnelRowType = seaTunnelRowType;
+        this(toTableSchema(seaTunnelRowType), redisParameters);
+    }
+
+    /** Creates a writer whose schema can be updated and restored from checkpoint state. */
+    public RedisSinkWriter(TableSchema tableSchema, RedisParameters redisParameters) {
+        this.tableSchema = normalizeTableSchema(tableSchema);
+        this.seaTunnelRowType = this.tableSchema.toPhysicalRowDataType();
         this.redisParameters = redisParameters;
         this.serializationSchema = createSerializationSchema(redisParameters, seaTunnelRowType);
         this.redisClient = redisParameters.buildRedisClient();
+        this.schemaChangeEventDispatcher = new TableSchemaChangeEventDispatcher();
         this.batchSize = redisParameters.getBatchSize();
         this.rowKinds = new ArrayList<>(batchSize);
         this.keyBuffer = new ArrayList<>(batchSize);
@@ -93,6 +124,40 @@ public void write(SeaTunnelRow element) throws IOException {
         log.debug("write redis key: {}, value: {}， rowKind: {}", key, value, element.getRowKind());
     }
 
+    /**
+     * Flushes all rows serialized with the previous schema before refreshing the writer's
+     * schema-derived views. The engine orders this callback with row processing, so all three views
+     * are updated before any subsequent row is processed.
+     */
+    @Override
+    public void applySchemaChange(SchemaChangeEvent event) {
+        flush();
+        tableSchema = schemaChangeEventDispatcher.reset(tableSchema).apply(event);
+        seaTunnelRowType = tableSchema.toPhysicalRowDataType();
+        serializationSchema = createSerializationSchema(redisParameters, seaTunnelRowType);
+    }
+
+    private static TableSchema normalizeTableSchema(TableSchema tableSchema) {
+        if (tableSchema.getConstraintKeys() != null) {
+            return tableSchema;
+        }
+        return TableSchema.builder()
+                .columns(tableSchema.getColumns())
+                .primaryKey(tableSchema.getPrimaryKey())
+                .build();
+    }
+
+    private static TableSchema toTableSchema(SeaTunnelRowType seaTunnelRowType) {
+        TableSchema.Builder schemaBuilder = TableSchema.builder();
+        String[] fieldNames = seaTunnelRowType.getFieldNames();
+        SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes();
+        for (int i = 0; i < fieldNames.length; i++) {
+            schemaBuilder.column(
+                    PhysicalColumn.of(fieldNames[i], fieldTypes[i], 0L, true, null, null));
+        }
+        return schemaBuilder.build();
+    }
+
     private String getKey(SeaTunnelRow element, List<String> fields) {
         String key = redisParameters.getKeyField();
         Boolean supportCustomKey = redisParameters.getSupportCustomKey();
@@ -291,6 +356,12 @@ public Optional<Void> prepareCommit() {
         return Optional.empty();
     }
 
+    /** Returns an isolated copy of the latest schema for checkpoint and rescale recovery. */
+    @Override
+    public List<TableSchema> snapshotState(long checkpointId) {
+        return Collections.singletonList(tableSchema.copy());
+    }
+
     private synchronized void flush() {
         if (!keyBuffer.isEmpty()) {
             doBatchWrite();

seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java

@@ -21,13 +21,15 @@
 import org.apache.seatunnel.api.configuration.util.ConfigValidator;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
 import org.apache.seatunnel.api.table.catalog.TableIdentifier;
 import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
 import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
 import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSink;
 import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory;
 import org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory;
 
@@ -37,6 +39,7 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -144,6 +147,20 @@ void sinkCreatedViaFactoryContext() {
                 () -> new RedisSinkFactory().createSink(context).createSink());
     }
 
+    @Test
+    void sinkExposesSchemaWriterStateSerializer() throws IOException {
+        RedisSink sink = new RedisSink(ReadonlyConfig.fromMap(singleSinkConfig()), catalogTable());
+
+        Assertions.assertTrue(sink.getWriterStateSerializer().isPresent());
+        @SuppressWarnings("unchecked")
+        Serializer<TableSchema> serializer =
+                (Serializer<TableSchema>) (Serializer<?>) sink.getWriterStateSerializer().get();
+        TableSchema tableSchema = catalogTable().getTableSchema();
+
+        Assertions.assertEquals(
+                tableSchema, serializer.deserialize(serializer.serialize(tableSchema)));
+    }
+
     // parameterized-case providers
 
     static Stream<Arguments> invalidSingleConnections() {