[Feature][Connector] add elasticsearch save_mode (#6046)

Author: chl-wxp

docs/en/connector-v2/sink/Elasticsearch.md

@@ -19,24 +19,26 @@ Engine Supported
 
 ## Options
 
-|          name           |  type   | required | default value |
-|-------------------------|---------|----------|---------------|
-| hosts                   | array   | yes      | -             |
-| index                   | string  | yes      | -             |
-| index_type              | string  | no       |               |
-| primary_keys            | list    | no       |               |
-| key_delimiter           | string  | no       | `_`           |
-| username                | string  | no       |               |
-| password                | string  | no       |               |
-| max_retry_count         | int     | no       | 3             |
-| max_batch_size          | int     | no       | 10            |
-| tls_verify_certificate  | boolean | no       | true          |
-| tls_verify_hostnames    | boolean | no       | true          |
-| tls_keystore_path       | string  | no       | -             |
-| tls_keystore_password   | string  | no       | -             |
-| tls_truststore_path     | string  | no       | -             |
-| tls_truststore_password | string  | no       | -             |
-| common-options          |         | no       | -             |
+|          name           |  type   | required |        default value         |
+|-------------------------|---------|----------|------------------------------|
+| hosts                   | array   | yes      | -                            |
+| index                   | string  | yes      | -                            |
+| schema_save_mode        | string  | yes      | CREATE_SCHEMA_WHEN_NOT_EXIST |
+| data_save_mode          | string  | yes      | APPEND_DATA                  |
+| index_type              | string  | no       |                              |
+| primary_keys            | list    | no       |                              |
+| key_delimiter           | string  | no       | `_`                          |
+| username                | string  | no       |                              |
+| password                | string  | no       |                              |
+| max_retry_count         | int     | no       | 3                            |
+| max_batch_size          | int     | no       | 10                           |
+| tls_verify_certificate  | boolean | no       | true                         |
+| tls_verify_hostnames    | boolean | no       | true                         |
+| tls_keystore_path       | string  | no       | -                            |
+| tls_keystore_password   | string  | no       | -                            |
+| tls_truststore_path     | string  | no       | -                            |
+| tls_truststore_password | string  | no       | -                            |
+| common-options          |         | no       | -                            |
 
 ### hosts [array]
 
@@ -103,6 +105,22 @@ The key password for the trust store specified
 
 Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details
 
+### schema_save_mode
+
+Before the synchronous task is turned on, different treatment schemes are selected for the existing surface structure of the target side.
+Option introduction：  
+RECREATE_SCHEMA ：Will create when the table does not exist, delete and rebuild when the table is saved  
+CREATE_SCHEMA_WHEN_NOT_EXIST ：Will Created when the table does not exist, skipped when the table is saved  
+ERROR_WHEN_SCHEMA_NOT_EXIST ：Error will be reported when the table does not exist
+
+### data_save_mode
+
+Before the synchronous task is turned on, different processing schemes are selected for data existing data on the target side.
+Option introduction：  
+DROP_DATA： Preserve database structure and delete data  
+APPEND_DATA：Preserve database structure, preserve data  
+ERROR_WHEN_DATA_EXISTS：When there is data, an error is reported
+
 ## Examples
 
 Simple
@@ -173,6 +191,21 @@ sink {
 }
 ```
 
+SAVE_MODE (Add saveMode function)
+
+```hocon
+sink {
+    Elasticsearch {
+        hosts = ["https://localhost:9200"]
+        username = "elastic"
+        password = "elasticsearch"
+        
+        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+        data_save_mode = "APPEND_DATA"
+    }
+}
+```
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalog.java

@@ -39,6 +39,7 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Lists;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -52,6 +53,7 @@
  *
  * <p>In ElasticSearch, we use the index as the database and table.
  */
+@Slf4j
 public class ElasticSearchCatalog implements Catalog {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchCatalog.class);
@@ -108,11 +110,12 @@ public boolean databaseExists(String databaseName) throws CatalogException {
             List<IndexDocsCount> indexDocsCount = esRestClient.getIndexDocsCount(databaseName);
             return true;
         } catch (Exception e) {
-            throw new CatalogException(
+            log.error(
                     String.format(
                             "Failed to check if catalog %s database %s exists",
                             catalogName, databaseName),
                     e);
+            return false;
         }
     }
 
@@ -177,13 +180,6 @@ public void createTable(TablePath tablePath, CatalogTable table, boolean ignoreI
             throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
         // Create the index
         checkNotNull(tablePath, "tablePath cannot be null");
-        if (tableExists(tablePath)) {
-            if (ignoreIfExists) {
-                return;
-            } else {
-                throw new TableAlreadyExistException(catalogName, tablePath, null);
-            }
-        }
         esRestClient.createIndex(tablePath.getTableName());
     }
 
@@ -217,6 +213,19 @@ public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
         dropTable(tablePath, ignoreIfNotExists);
     }
 
+    @Override
+    public void truncateTable(TablePath tablePath, boolean ignoreIfNotExists) {
+        dropTable(tablePath, ignoreIfNotExists);
+        createTable(tablePath, null, ignoreIfNotExists);
+    }
+
+    @Override
+    public boolean isExistsData(TablePath tablePath) {
+        final List<IndexDocsCount> indexDocsCount =
+                esRestClient.getIndexDocsCount(tablePath.getTableName());
+        return indexDocsCount.get(0).getDocsCount() > 0;
+    }
+
     private Map<String, String> buildTableOptions(TablePath tablePath) {
         Map<String, String> options = new HashMap<>();
         options.put("connector", "elasticsearch");

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalogFactory.java

@@ -17,28 +17,32 @@
 
 package org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog;
 
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.catalog.Catalog;
 import org.apache.seatunnel.api.table.factory.CatalogFactory;
+import org.apache.seatunnel.api.table.factory.Factory;
+
+import com.google.auto.service.AutoService;
 
+@AutoService(Factory.class)
 public class ElasticSearchCatalogFactory implements CatalogFactory {
 
     @Override
     public Catalog createCatalog(String catalogName, ReadonlyConfig options) {
-        // todo:
-        return null;
+        Config config = options.toConfig();
+        return new ElasticSearchCatalog(catalogName, "", config);
     }
 
     @Override
     public String factoryIdentifier() {
-        // todo:
         return "Elasticsearch";
     }
 
     @Override
     public OptionRule optionRule() {
-        // todo:
-        return null;
+        return OptionRule.builder().build();
     }
 }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/SinkConfig.java

@@ -19,9 +19,16 @@
 
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
 
+import java.util.Arrays;
 import java.util.List;
 
+import static org.apache.seatunnel.api.sink.DataSaveMode.APPEND_DATA;
+import static org.apache.seatunnel.api.sink.DataSaveMode.DROP_DATA;
+import static org.apache.seatunnel.api.sink.DataSaveMode.ERROR_WHEN_DATA_EXISTS;
+
 public class SinkConfig {
 
     public static final Option<String> INDEX =
@@ -62,4 +69,18 @@ public class SinkConfig {
                     .intType()
                     .defaultValue(3)
                     .withDescription("one bulk request max try count");
+
+    public static final Option<SchemaSaveMode> SCHEMA_SAVE_MODE =
+            Options.key("schema_save_mode")
+                    .enumType(SchemaSaveMode.class)
+                    .defaultValue(SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST)
+                    .withDescription("schema_save_mode");
+
+    public static final Option<DataSaveMode> DATA_SAVE_MODE =
+            Options.key("data_save_mode")
+                    .singleChoice(
+                            DataSaveMode.class,
+                            Arrays.asList(DROP_DATA, APPEND_DATA, ERROR_WHEN_DATA_EXISTS))
+                    .defaultValue(APPEND_DATA)
+                    .withDescription("data_save_mode");
 }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSink.java

@@ -20,26 +20,40 @@
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.DefaultSaveModeHandler;
+import org.apache.seatunnel.api.sink.SaveModeHandler;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportSaveMode;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.factory.CatalogFactory;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.state.ElasticsearchAggregatedCommitInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.state.ElasticsearchCommitInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.state.ElasticsearchSinkState;
 
 import com.google.auto.service.AutoService;
 
+import java.util.Optional;
+
+import static org.apache.seatunnel.api.table.factory.FactoryUtil.discoverFactory;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SinkConfig.MAX_BATCH_SIZE;
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SinkConfig.MAX_RETRY_COUNT;
 
 @AutoService(SeaTunnelSink.class)
 public class ElasticsearchSink
         implements SeaTunnelSink<
-                SeaTunnelRow,
-                ElasticsearchSinkState,
-                ElasticsearchCommitInfo,
-                ElasticsearchAggregatedCommitInfo> {
+                        SeaTunnelRow,
+                        ElasticsearchSinkState,
+                        ElasticsearchCommitInfo,
+                        ElasticsearchAggregatedCommitInfo>,
+                SupportSaveMode {
 
     private Config pluginConfig;
     private SeaTunnelRowType seaTunnelRowType;
@@ -75,4 +89,27 @@ public SinkWriter<SeaTunnelRow, ElasticsearchCommitInfo, ElasticsearchSinkState>
         return new ElasticsearchSinkWriter(
                 context, seaTunnelRowType, pluginConfig, maxBatchSize, maxRetryCount);
     }
+
+    @Override
+    public Optional<SaveModeHandler> getSaveModeHandler() {
+        CatalogFactory catalogFactory =
+                discoverFactory(
+                        Thread.currentThread().getContextClassLoader(),
+                        CatalogFactory.class,
+                        getPluginName());
+        if (catalogFactory == null) {
+            return Optional.empty();
+        }
+        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(pluginConfig);
+        Catalog catalog =
+                catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), readonlyConfig);
+        SchemaSaveMode schemaSaveMode = readonlyConfig.get(SinkConfig.SCHEMA_SAVE_MODE);
+        DataSaveMode dataSaveMode = readonlyConfig.get(SinkConfig.DATA_SAVE_MODE);
+
+        TablePath tablePath = TablePath.of("", readonlyConfig.get(SinkConfig.INDEX));
+        catalog.open();
+        return Optional.of(
+                new DefaultSaveModeHandler(
+                        schemaSaveMode, dataSaveMode, catalog, tablePath, null, null));
+    }
 }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkFactory.java

@@ -20,6 +20,7 @@
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SinkConfig;
 
 import com.google.auto.service.AutoService;
 
@@ -49,7 +50,7 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(HOSTS, INDEX)
+                .required(HOSTS, INDEX, SinkConfig.SCHEMA_SAVE_MODE, SinkConfig.DATA_SAVE_MODE)
                 .optional(
                         INDEX_TYPE,
                         PRIMARY_KEYS,

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java

@@ -20,8 +20,11 @@
 import org.apache.seatunnel.shade.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
 
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog.ElasticSearchCatalog;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.ScrollResult;
 import org.apache.seatunnel.e2e.common.TestResource;
@@ -49,6 +52,7 @@
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -255,4 +259,37 @@ public void tearDown() {
         }
         container.close();
     }
+
+    @TestTemplate
+    public void testCatalog(TestContainer container2) throws IOException, InterruptedException {
+        Map<String, Object> configMap = new HashMap<>();
+        configMap.put("username", "elastic");
+        configMap.put("password", "elasticsearch");
+        configMap.put("hosts", Arrays.asList("https://" + container.getHttpHostAddress()));
+        configMap.put("index", "st_index3");
+        configMap.put("tls_verify_certificate", false);
+        configMap.put("tls_verify_hostname", false);
+        configMap.put("index_type", "st");
+        final ElasticSearchCatalog elasticSearchCatalog =
+                new ElasticSearchCatalog("Elasticsearch", "", ConfigFactory.parseMap(configMap));
+        elasticSearchCatalog.open();
+        TablePath tablePath = TablePath.of("", "st_index3");
+        // index exists
+        final boolean existsBefore = elasticSearchCatalog.tableExists(tablePath);
+        Assertions.assertFalse(existsBefore);
+        // create index
+        elasticSearchCatalog.createTable(tablePath, null, false);
+        final boolean existsAfter = elasticSearchCatalog.tableExists(tablePath);
+        Assertions.assertTrue(existsAfter);
+        // data exists?
+        final boolean existsData = elasticSearchCatalog.isExistsData(tablePath);
+        Assertions.assertFalse(existsData);
+        // truncate
+        elasticSearchCatalog.truncateTable(tablePath, false);
+        Assertions.assertTrue(elasticSearchCatalog.tableExists(tablePath));
+        // drop
+        elasticSearchCatalog.dropTable(tablePath, false);
+        Assertions.assertFalse(elasticSearchCatalog.tableExists(tablePath));
+        elasticSearchCatalog.close();
+    }
 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/elasticsearch_source_and_sink.conf

@@ -69,5 +69,7 @@ sink {
 
     index = "st_index2"
     index_type = "st"
+    "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
+    "data_save_mode"="APPEND_DATA"
   }
 }