[Improve][Transform-V2] Migrate transforms validation to declarative OptionRule

docs/en/introduction/concepts/incompatible-changes.md

@@ -102,6 +102,17 @@ You need to check this document before you upgrade to related version.
 
   **Migration Guide**: If you are using custom datetime format patterns in `PARSEDATETIME`, `TO_DATE`, or `IS_DATE` functions, you must update your queries to use one of the supported patterns above. If your data uses a different format, you may need to preprocess the input data to match a supported format, or use string manipulation functions to transform the format before parsing.
 - DataValidator transform: In `row_error_handle_way = ROUTE_TO_TABLE` mode, the routed error row `table_id` now includes the upstream database/schema prefix (for example, `db1.ffp` / `db1.schema1.ffp` instead of `ffp`).
+- **[BREAKING]** Several transform plugins now perform stricter submission-time config validation via declarative `OptionRule`. Configs that previously passed submission but failed at runtime will now be rejected at submission time with a descriptive `OptionValidationException`:
+
+  | Transform | Newly Rejected Config | Previous Behavior | Migration |
+  |-----------|----------------------|-------------------|-----------|
+  | `DefineSinkType` | `columns` entries with null/empty `column` or `type` | Runtime NPE or undefined behavior | Ensure every entry has non-empty `column` and `type` fields |
+  | `DefineSinkType` | `columns` with duplicate column names | Silent override or runtime conflict | Remove duplicate column entries |
+  | `FieldEncrypt` | `max_field_length` set to ≤ 0 | Ignored or unexpected truncation | Set `max_field_length` to a positive integer, or remove the option to use the default |
+  | `DynamicCompile` | `compile_pattern = SOURCE_CODE` without a non-blank `source_code` | Runtime compilation failure | Provide `source_code` when using `SOURCE_CODE` pattern |
+  | `DynamicCompile` | `compile_pattern = ABSOLUTE_PATH` without a non-blank `absolute_path` | Runtime file-read failure | Provide `absolute_path` when using `ABSOLUTE_PATH` pattern |
+
+  **Migration Guide**: Review your transform configs against the table above. If any of your existing configs match a "Newly Rejected" pattern, update them before upgrading. The error messages at submission time now clearly identify which option is invalid and why.
 - Adjusted SQL Transform date & time functions:
   - `DATEDIFF(<start>, <end>, 'MONTH')` now returns the total number of months between the two dates across years (for example, from `2023-01-01` to `2024-03-01` returns `14` instead of `15`).
   - `WEEK(<datetime>)` now returns the ISO week number directly (previous behavior added an extra `+1` to the ISO week value).

docs/zh/introduction/concepts/incompatible-changes.md

@@ -98,6 +98,18 @@
   **迁移指南**: 如果您在 `PARSEDATETIME`、`TO_DATE` 或 `IS_DATE` 函数中使用自定义日期时间格式模式，您必须更新查询以使用上述支持的模式之一。如果您的数据使用不同的格式，您可能需要预处理输入数据以匹配支持的格式，或使用字符串操作函数在解析之前转换格式。
 
 - DataValidator 转换：当 `row_error_handle_way = ROUTE_TO_TABLE` 时，路由到错误表的行 `table_id` 现在会携带上游的 database/schema 前缀（例如从 `ffp` 变为 `db1.ffp` / `db1.schema1.ffp`）。
+- **[BREAKING]** 多个转换插件现在通过声明式 `OptionRule` 在提交时执行更严格的配置校验。以前在提交时能通过但运行时失败的配置，现在会在提交时被拒绝，并抛出描述清晰的 `OptionValidationException`：
+
+  | 转换插件 | 新增拒绝的配置 | 以前的行为 | 迁移方式 |
+  |---------|--------------|-----------|---------|
+  | `DefineSinkType` | `columns` 条目中 `column` 或 `type` 为空 | 运行时 NPE 或未定义行为 | 确保每个条目都有非空的 `column` 和 `type` 字段 |
+  | `DefineSinkType` | `columns` 中存在重复列名 | 静默覆盖或运行时冲突 | 移除重复的列条目 |
+  | `FieldEncrypt` | `max_field_length` 设置为 ≤ 0 | 被忽略或产生意外截断 | 设置为正整数，或移除该选项以使用默认值 |
+  | `DynamicCompile` | `compile_pattern = SOURCE_CODE` 但 `source_code` 为空 | 运行时编译失败 | 使用 `SOURCE_CODE` 模式时提供 `source_code` |
+  | `DynamicCompile` | `compile_pattern = ABSOLUTE_PATH` 但 `absolute_path` 为空 | 运行时文件读取失败 | 使用 `ABSOLUTE_PATH` 模式时提供 `absolute_path` |
+
+  **迁移指南**：升级前请对照上表检查您的转换配置。如果现有配置匹配了"新增拒绝的配置"中的情况，请在升级前修改。提交时的错误消息会清楚标明哪个选项无效及原因。
+
 ### 引擎行为变更
 
 ### 依赖升级

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformConfig.java

@@ -33,8 +33,6 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkArgument;
-
 @Data
 @AllArgsConstructor
 public class DefineSinkTypeTransformConfig implements Serializable {
@@ -82,16 +80,6 @@ public static class TableTransforms implements Serializable {
 
     public static DefineSinkTypeTransformConfig of(ReadonlyConfig config) {
         List<DefineColumnType> columns = config.get(COLUMNS);
-
-        checkArgument(columns != null && !columns.isEmpty(), "The columns must be set");
-        columns.forEach(
-                defineColumnType -> {
-                    checkArgument(
-                            defineColumnType.getColumn() != null, "The column name must be set");
-                    checkArgument(
-                            defineColumnType.getType() != null, "The column type must be set");
-                });
-
         return new DefineSinkTypeTransformConfig(columns);
     }
 }

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/adaptsink/DefineSinkTypeTransformFactory.java

@@ -17,15 +17,24 @@
 
 package org.apache.seatunnel.transform.adaptsink;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableTransformFactory;
 import org.apache.seatunnel.api.table.factory.TableTransformFactoryContext;
+import org.apache.seatunnel.transform.adaptsink.DefineSinkTypeTransformConfig.DefineColumnType;
 import org.apache.seatunnel.transform.common.TransformCommonOptions;
 
 import com.google.auto.service.AutoService;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
 @AutoService(Factory.class)
 public class DefineSinkTypeTransformFactory implements TableTransformFactory {
     @Override
@@ -36,7 +45,13 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(DefineSinkTypeTransformConfig.COLUMNS)
+                .required(
+                        DefineSinkTypeTransformConfig.COLUMNS,
+                        Conditions.notEmpty(DefineSinkTypeTransformConfig.COLUMNS)
+                                .and(
+                                        Conditions.extension(
+                                                DefineSinkTypeTransformConfig.COLUMNS,
+                                                new ColumnsStructureValidator())))
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();
@@ -48,4 +63,39 @@ public TableTransform createTransform(TableTransformFactoryContext context) {
                 new DefineSinkTypeMultiCatalogTransform(
                         context.getCatalogTables(), context.getOptions());
     }
+
+    static class ColumnsStructureValidator implements ConditionExtension<List<DefineColumnType>> {
+        @Override
+        public String description() {
+            return "each column entry must contain non-null 'column' and 'type'";
+        }
+
+        @Override
+        public boolean evaluate(ReadonlyConfig config, List<DefineColumnType> value)
+                throws OptionValidationException {
+            if (value == null) {
+                return false;
+            }
+            Set<String> seen = new HashSet<>();
+            for (int i = 0; i < value.size(); i++) {
+                DefineColumnType entry = value.get(i);
+                if (entry.getColumn() == null || entry.getColumn().trim().isEmpty()) {
+                    throw new OptionValidationException(
+                            String.format(
+                                    "columns[%d]: 'column' name must not be null or empty", i));
+                }
+                if (entry.getType() == null || entry.getType().trim().isEmpty()) {
+                    throw new OptionValidationException(
+                            String.format("columns[%d]: 'type' must not be null or empty", i));
+                }
+                if (!seen.add(entry.getColumn())) {
+                    throw new OptionValidationException(
+                            String.format(
+                                    "columns[%d]: duplicate column name '%s'",
+                                    i, entry.getColumn()));
+                }
+            }
+            return true;
+        }
+    }
 }

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/copy/CopyFieldTransformFactory.java

@@ -17,6 +17,9 @@
 
 package org.apache.seatunnel.transform.copy;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -36,8 +39,15 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .bundled(CopyTransformConfig.SRC_FIELD, CopyTransformConfig.DEST_FIELD)
-                .bundled(CopyTransformConfig.FIELDS)
+                .exclusive(CopyTransformConfig.FIELDS, CopyTransformConfig.SRC_FIELD)
+                .optional(
+                        CopyTransformConfig.FIELDS,
+                        Conditions.mapNotEmpty(CopyTransformConfig.FIELDS))
+                .optional(
+                        CopyTransformConfig.SRC_FIELD,
+                        Conditions.extension(
+                                CopyTransformConfig.SRC_FIELD, new RequireDestFieldValidator()))
+                .optional(CopyTransformConfig.DEST_FIELD)
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();
@@ -49,4 +59,17 @@ public TableTransform createTransform(TableTransformFactoryContext context) {
                 new CopyFieldMultiCatalogTransform(
                         context.getCatalogTables(), context.getOptions());
     }
+
+    static class RequireDestFieldValidator implements ConditionExtension<String> {
+        @Override
+        public String description() {
+            return "'dest_field' is required when 'src_field' is provided";
+        }
+
+        @Override
+        public boolean evaluate(ReadonlyConfig config, String value) {
+            String destField = config.get(CopyTransformConfig.DEST_FIELD);
+            return destField != null && !destField.trim().isEmpty();
+        }
+    }
 }

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/dynamiccompile/DynamicCompileTransformFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.transform.dynamiccompile;
 
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -36,17 +37,24 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .optional(
-                        DynamicCompileTransformConfig.COMPILE_LANGUAGE,
-                        DynamicCompileTransformConfig.COMPILE_PATTERN)
+                .required(DynamicCompileTransformConfig.COMPILE_LANGUAGE)
+                .optional(DynamicCompileTransformConfig.COMPILE_PATTERN)
                 .conditional(
                         DynamicCompileTransformConfig.COMPILE_PATTERN,
                         CompilePattern.SOURCE_CODE,
                         DynamicCompileTransformConfig.SOURCE_CODE)
+                .conditional(
+                        DynamicCompileTransformConfig.COMPILE_PATTERN,
+                        CompilePattern.SOURCE_CODE,
+                        Conditions.notBlank(DynamicCompileTransformConfig.SOURCE_CODE))
                 .conditional(
                         DynamicCompileTransformConfig.COMPILE_PATTERN,
                         CompilePattern.ABSOLUTE_PATH,
                         DynamicCompileTransformConfig.ABSOLUTE_PATH)
+                .conditional(
+                        DynamicCompileTransformConfig.COMPILE_PATTERN,
+                        CompilePattern.ABSOLUTE_PATH,
+                        Conditions.notBlank(DynamicCompileTransformConfig.ABSOLUTE_PATH))
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/encrypt/FieldEncryptTransformFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.transform.encrypt;
 
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -38,10 +39,17 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(FieldEncryptTransformConfig.FIELDS)
-                .required(FieldEncryptTransformConfig.KEY)
+                .required(
+                        FieldEncryptTransformConfig.FIELDS,
+                        Conditions.notEmpty(FieldEncryptTransformConfig.FIELDS))
+                .required(
+                        FieldEncryptTransformConfig.KEY,
+                        Conditions.notBlank(FieldEncryptTransformConfig.KEY))
                 .optional(FieldEncryptTransformConfig.ALGORITHM)
                 .optional(FieldEncryptTransformConfig.MODE)
+                .optional(
+                        FieldEncryptTransformConfig.MAX_FIELD_LENGTH,
+                        Conditions.greaterThan(FieldEncryptTransformConfig.MAX_FIELD_LENGTH, 0))
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/fieldmapper/FieldMapperTransformFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.transform.fieldmapper;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -37,7 +38,9 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .optional(FieldMapperTransformConfig.FIELD_MAPPER)
+                .required(
+                        FieldMapperTransformConfig.FIELD_MAPPER,
+                        Conditions.mapNotEmpty(FieldMapperTransformConfig.FIELD_MAPPER))
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransform.java

@@ -18,8 +18,6 @@
 package org.apache.seatunnel.transform.filter;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
-import org.apache.seatunnel.api.configuration.util.ConfigValidator;
-import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
@@ -58,14 +56,6 @@ public FilterFieldTransform(
         SeaTunnelRowType seaTunnelRowType = catalogTable.getTableSchema().toPhysicalRowDataType();
         includeFields = config.get(FilterFieldTransformConfig.INCLUDE_FIELDS);
         excludeFields = config.get(FilterFieldTransformConfig.EXCLUDE_FIELDS);
-        // exactly only one should be set
-        ConfigValidator.of(config)
-                .validate(
-                        OptionRule.builder()
-                                .exclusive(
-                                        FilterFieldTransformConfig.INCLUDE_FIELDS,
-                                        FilterFieldTransformConfig.EXCLUDE_FIELDS)
-                                .build());
         List<String> canNotFoundFields =
                 Stream.concat(
                                 Optional.ofNullable(includeFields).orElse(new ArrayList<>())

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/filter/FilterFieldTransformFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.transform.filter;
 
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableTransform;
 import org.apache.seatunnel.api.table.factory.Factory;
@@ -38,9 +39,15 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .optional(
+                .exclusive(
                         FilterFieldTransformConfig.INCLUDE_FIELDS,
                         FilterFieldTransformConfig.EXCLUDE_FIELDS)
+                .optional(
+                        FilterFieldTransformConfig.INCLUDE_FIELDS,
+                        Conditions.notEmpty(FilterFieldTransformConfig.INCLUDE_FIELDS))
+                .optional(
+                        FilterFieldTransformConfig.EXCLUDE_FIELDS,
+                        Conditions.notEmpty(FilterFieldTransformConfig.EXCLUDE_FIELDS))
                 .optional(TransformCommonOptions.MULTI_TABLES)
                 .optional(TransformCommonOptions.TABLE_MATCH_REGEX)
                 .build();