[Feature][API] add savemode feature API

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/Options.java

@@ -20,6 +20,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.NonNull;
 import org.apache.commons.lang3.StringUtils;
 
 import java.lang.reflect.Type;
@@ -162,6 +163,18 @@ public Type getType() {
             });
         }
 
+        /**
+         * Construct an option with multiple options and only one of them can be selected
+         */
+        public <T> SingleChoiceOptionBuilder<T> singleChoice(@NonNull Class<T> optionType, @NonNull List<T> optionValues) {
+            return new SingleChoiceOptionBuilder<T>(key, new TypeReference<T>() {
+                @Override
+                public Type getType() {
+                    return optionType;
+                }
+            }, optionValues);
+        }
+
         /**
          * The value of the definition option should be represented as T.
          *
@@ -205,4 +218,35 @@ public Option<T> noDefaultValue() {
             return new Option<>(key, typeReference, null);
         }
     }
+
+    public static class SingleChoiceOptionBuilder<T> {
+        private final List<T> optionValues;
+        private final String key;
+        private final TypeReference<T> typeReference;
+
+        SingleChoiceOptionBuilder(String key, TypeReference typeReference, List<T> optionValues) {
+            this.optionValues = optionValues;
+            this.key = key;
+            this.typeReference = typeReference;
+        }
+
+        /**
+         * Creates a Option with the given default value.
+         *
+         * @param value The default value for the config option
+         * @return The config option with the default value.
+         */
+        public Option<T> defaultValue(T value) {
+            return new SingleChoiceOption<T>(key, typeReference, optionValues, value);
+        }
+
+        /**
+         * Creates a Option without a default value.
+         *
+         * @return The config option without a default value.
+         */
+        public Option<T> noDefaultValue() {
+            return new SingleChoiceOption<T>(key, typeReference, optionValues, null);
+        }
+    }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/SingleChoiceOption.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.configuration;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.Getter;
+
+import java.util.List;
+
+public class SingleChoiceOption<T> extends Option{
+
+    @Getter
+    private final List<T> optionValues;
+
+    public SingleChoiceOption(String key,
+                              TypeReference<T> typeReference,
+                              List<T> optionValues,
+                              T defaultValue) {
+        super(key, typeReference, defaultValue);
+        this.optionValues = optionValues;
+    }
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConfigValidator.java

@@ -21,8 +21,12 @@
 
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+
+import org.apache.commons.collections4.CollectionUtils;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 
@@ -41,6 +45,38 @@ public void validate(OptionRule rule) {
         List<RequiredOption> requiredOptions = rule.getRequiredOptions();
         for (RequiredOption requiredOption : requiredOptions) {
             validate(requiredOption);
+            requiredOption.getOptions().forEach(option -> {
+                if (SingleChoiceOption.class.isAssignableFrom(option.getClass())) {
+                    validateSingleChoice(option);
+                }
+            });
+        }
+
+        for (Option option : rule.getOptionalOptions()) {
+            if (SingleChoiceOption.class.isAssignableFrom(option.getClass())) {
+                validateSingleChoice(option);
+            }
+        }
+    }
+
+    void validateSingleChoice(Option option) {
+        SingleChoiceOption singleChoiceOption = (SingleChoiceOption) option;
+        List optionValues = singleChoiceOption.getOptionValues();
+        if (CollectionUtils.isEmpty(optionValues)) {
+            throw new OptionValidationException("These options(%s) are SingleChoiceOption, the optionValues must not be null.", getOptionKeys(
+                Arrays.asList(singleChoiceOption)));
+        }
+
+        Object o = singleChoiceOption.defaultValue();
+        if (o != null && !optionValues.contains(o)) {
+            throw new OptionValidationException("These options(%s) are SingleChoiceOption, the defaultValue(%s) must be one of the optionValues.", getOptionKeys(
+                Arrays.asList(singleChoiceOption)), o);
+        }
+
+        Object value = config.get(option);
+        if (value != null && !optionValues.contains(value)) {
+            throw new OptionValidationException("These options(%s) are SingleChoiceOption, the value(%s) must be one of the optionValues.", getOptionKeys(
+                Arrays.asList(singleChoiceOption)), value);
         }
     }
 

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/DataSaveMode.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.sink;
+
+/**
+ * The SaveMode for the Sink connectors that use table or other table structures to organize data
+ */
+public enum DataSaveMode {
+    // Will drop table in MySQL, Will drop path for File Connector.
+    DROP_SCHEMA,
+
+    // Only drop the data in MySQL, Only drop the files in the path for File Connector.
+    KEEP_SCHEMA_DROP_DATA,
+
+    // Keep the table and data and continue to write data to the existing table for MySQL. Keep the path and files in the path, create new files in the path.
+    KEEP_SCHEMA_AND_DATA,
+
+    // Throw error when table is exists for MySQL. Throw error when path is exists.
+    ERROR_WHEN_EXISTS
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SinkCommonOptions.java

@@ -22,6 +22,8 @@
 
 public class SinkCommonOptions {
 
+    public static final String DATA_SAVE_MODE = "save_mode";
+
     public static final Option<String> SOURCE_TABLE_NAME =
         Options.key("source_table_name")
             .stringType()

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SupportDataSaveMode.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.sink;
+
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * The Sink Connectors which support data SaveMode should implement this interface
+ */
+public interface SupportDataSaveMode {
+
+    /**
+     * We hope every sink connector use the same option name to config SaveMode, So I add checkOptions method to this interface.
+     * checkOptions method have a default implement to check whether `save_mode` parameter is in config.
+     *
+     * @param config config of Sink Connector
+     * @return TableSaveMode TableSaveMode
+     */
+    default void checkOptions(Config config) {
+        if (config.hasPath(SinkCommonOptions.DATA_SAVE_MODE)) {
+            String tableSaveMode = config.getString(SinkCommonOptions.DATA_SAVE_MODE);
+            DataSaveMode dataSaveMode = DataSaveMode.valueOf(tableSaveMode.toUpperCase(Locale.ROOT));
+            if (!supportedDataSaveModeValues().contains(dataSaveMode)) {
+                throw new SeaTunnelRuntimeException(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                    "This connector don't support save mode: " + dataSaveMode);
+            }
+        } else {
+            throw new SeaTunnelRuntimeException(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                SinkCommonOptions.DATA_SAVE_MODE + " must in config");
+        }
+    }
+
+    DataSaveMode getDataSaveModeUsed();
+
+    /**
+     * Return the DataSaveMode list supported by this connector
+     * @return
+     */
+    List<DataSaveMode> supportedDataSaveModeValues();
+
+    void handleSaveMode(DataSaveMode tableSaveMode);
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java

@@ -17,13 +17,19 @@
 
 package org.apache.seatunnel.api.table.factory;
 
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.sink.DataSaveMode;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkCommonOptions;
+import org.apache.seatunnel.api.sink.SupportDataSaveMode;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceCommonOptions;
 import org.apache.seatunnel.api.source.SourceSplit;
 import org.apache.seatunnel.api.table.catalog.Catalog;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.connector.TableSink;
 import org.apache.seatunnel.api.table.connector.TableSource;
 
 import lombok.NonNull;
@@ -174,4 +180,34 @@ public static OptionRule sourceFullOptionRule(@NonNull Factory factory) {
         sourceOptionRule.getOptionalOptions().addAll(sourceCommonOptionRule.getOptionalOptions());
         return sourceOptionRule;
     }
+
+    /**
+     * This method is called by SeaTunnel Web to get the full option rule of a sink.
+     * @return
+     */
+    public static OptionRule sinkFullOptionRule(@NonNull TableSinkFactory factory) {
+        OptionRule sinkOptionRule = factory.optionRule();
+        if (sinkOptionRule == null) {
+            throw new FactoryException("sinkOptionRule can not be null");
+        }
+
+        try {
+            TableSink sink = factory.createSink(null);
+            if (SupportDataSaveMode.class.isAssignableFrom(sink.getClass())) {
+                SupportDataSaveMode supportDataSaveModeSink = (SupportDataSaveMode) sink;
+                Option<DataSaveMode> saveMode =
+                    Options.key(SinkCommonOptions.DATA_SAVE_MODE)
+                        .singleChoice(DataSaveMode.class, supportDataSaveModeSink.supportedDataSaveModeValues())
+                        .noDefaultValue()
+                        .withDescription("data save mode");
+                OptionRule sinkCommonOptionRule =
+                    OptionRule.builder().required(saveMode).build();
+                sinkOptionRule.getOptionalOptions().addAll(sinkCommonOptionRule.getOptionalOptions());
+            }
+        } catch (UnsupportedOperationException e) {
+            LOG.warn("Add save mode option need sink connector support create sink by TableSinkFactory");
+        }
+
+        return sinkOptionRule;
+    }
 }

seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java

@@ -63,6 +63,18 @@ public class ConfigValidatorTest {
             .noDefaultValue()
             .withDescription("base64 encoded kerberos ticket of the Neo4j. for Auth.");
 
+    public static final Option<String> SINGLE_CHOICE_TEST =
+        Options.key("single_choice_test")
+            .singleChoice(String.class, Arrays.asList("A", "B", "C"))
+            .defaultValue("M")
+            .withDescription("test single choice error");
+
+    public static final Option<String> SINGLE_CHOICE_VALUE_TEST =
+        Options.key("single_choice_test")
+            .singleChoice(String.class, Arrays.asList("A", "B", "C"))
+            .defaultValue("A")
+            .withDescription("test single choice value");
+
     void validate(Map<String, Object> config, OptionRule rule) {
         ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule);
     }
@@ -245,4 +257,28 @@ public void testComplexConditionalRequiredOptions() {
         config.put(KEY_USERNAME.key(), "asuka111");
         Assertions.assertDoesNotThrow(executable);
     }
+
+    @Test
+    public void testSingleChoiceOptionDefaultValueValidator() {
+        OptionRule optionRule = OptionRule.builder().required(SINGLE_CHOICE_TEST).build();
+        Map<String, Object> config = new HashMap<>();
+        config.put(SINGLE_CHOICE_TEST.key(), "A");
+        Executable executable = () -> validate(config, optionRule);
+        assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options('single_choice_test') are SingleChoiceOption, the defaultValue(M) must be one of the optionValues.",
+            assertThrows(OptionValidationException.class, executable).getMessage());
+    }
+
+    @Test
+    public void testSingleChoiceOptionValueValidator() {
+        OptionRule optionRule = OptionRule.builder().required(SINGLE_CHOICE_VALUE_TEST).build();
+        Map<String, Object> config = new HashMap<>();
+        config.put(SINGLE_CHOICE_VALUE_TEST.key(), "A");
+        Executable executable = () -> validate(config, optionRule);
+        Assertions.assertDoesNotThrow(executable);
+
+        config.put(SINGLE_CHOICE_VALUE_TEST.key(), "N");
+        executable = () -> validate(config, optionRule);
+        assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options('single_choice_test') are SingleChoiceOption, the value(N) must be one of the optionValues.",
+            assertThrows(OptionValidationException.class, executable).getMessage());
+    }
 }