[Feature][Paimon] paimon sink support savemode

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalog.java

@@ -0,0 +1,215 @@
+/*
+ * 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.connectors.seatunnel.paimon.catalog;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
+import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonSinkConfig;
+import org.apache.seatunnel.connectors.seatunnel.paimon.utils.SchemaUtil;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataField;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+@Slf4j
+public class PaimonCatalog implements Catalog, PaimonTable {
+    private static final String DEFAULT_DATABASE = "default";
+
+    private String catalogName;
+    private ReadonlyConfig readonlyConfig;
+    private PaimonCatalogLoader paimonCatalogLoader;
+    private org.apache.paimon.catalog.Catalog catalog;
+
+    public PaimonCatalog(String catalogName, ReadonlyConfig readonlyConfig) {
+        this.readonlyConfig = readonlyConfig;
+        this.catalogName = catalogName;
+        this.paimonCatalogLoader = new PaimonCatalogLoader(new PaimonSinkConfig(readonlyConfig));
+    }
+
+    @Override
+    public void open() throws CatalogException {
+        this.catalog = paimonCatalogLoader.loadCatalog();
+    }
+
+    @Override
+    public void close() throws CatalogException {
+        if (catalog != null && catalog instanceof Closeable) {
+            try {
+                ((Closeable) catalog).close();
+            } catch (IOException e) {
+                log.error("Error while closing IcebergCatalog.", e);
+                throw new CatalogException(e);
+            }
+        }
+    }
+
+    @Override
+    public String name() {
+        return this.catalogName;
+    }
+
+    @Override
+    public String getDefaultDatabase() throws CatalogException {
+        return DEFAULT_DATABASE;
+    }
+
+    @Override
+    public boolean databaseExists(String databaseName) throws CatalogException {
+        return catalog.databaseExists(databaseName);
+    }
+
+    @Override
+    public List<String> listDatabases() throws CatalogException {
+        return catalog.listDatabases();
+    }
+
+    @Override
+    public List<String> listTables(String databaseName)
+            throws CatalogException, DatabaseNotExistException {
+        try {
+            return catalog.listTables(databaseName);
+        } catch (org.apache.paimon.catalog.Catalog.DatabaseNotExistException e) {
+            throw new DatabaseNotExistException(this.catalogName, databaseName);
+        }
+    }
+
+    @Override
+    public boolean tableExists(TablePath tablePath) throws CatalogException {
+        return catalog.tableExists(toIdentifier(tablePath));
+    }
+
+    @Override
+    public CatalogTable getTable(TablePath tablePath)
+            throws CatalogException, TableNotExistException {
+        try {
+            FileStoreTable paimonFileStoreTableTable = (FileStoreTable) getPaimonTable(tablePath);
+            return toCatalogTable(paimonFileStoreTableTable, tablePath);
+        } catch (Exception e) {
+            throw new TableNotExistException(this.catalogName, tablePath);
+        }
+    }
+
+    @Override
+    public Table getPaimonTable(TablePath tablePath)
+            throws CatalogException, TableNotExistException {
+        try {
+            return catalog.getTable(toIdentifier(tablePath));
+        } catch (org.apache.paimon.catalog.Catalog.TableNotExistException e) {
+            throw new TableNotExistException(this.catalogName, tablePath);
+        }
+    }
+
+    @Override
+    public void createTable(TablePath tablePath, CatalogTable table, boolean ignoreIfExists)
+            throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
+        try {
+            catalog.createTable(
+                    toIdentifier(tablePath),
+                    SchemaUtil.toPaimonSchema(table.getTableSchema()),
+                    ignoreIfExists);
+        } catch (org.apache.paimon.catalog.Catalog.TableAlreadyExistException e) {
+            throw new TableAlreadyExistException(this.catalogName, tablePath);
+        } catch (org.apache.paimon.catalog.Catalog.DatabaseNotExistException e) {
+            throw new DatabaseNotExistException(this.catalogName, tablePath.getDatabaseName());
+        }
+    }
+
+    @Override
+    public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
+            throws TableNotExistException, CatalogException {
+        try {
+            catalog.dropTable(toIdentifier(tablePath), ignoreIfNotExists);
+        } catch (org.apache.paimon.catalog.Catalog.TableNotExistException e) {
+            throw new TableNotExistException(this.catalogName, tablePath);
+        }
+    }
+
+    @Override
+    public void createDatabase(TablePath tablePath, boolean ignoreIfExists)
+            throws DatabaseAlreadyExistException, CatalogException {
+        try {
+            catalog.createDatabase(tablePath.getDatabaseName(), ignoreIfExists);
+        } catch (org.apache.paimon.catalog.Catalog.DatabaseAlreadyExistException e) {
+            throw new DatabaseAlreadyExistException(this.catalogName, tablePath.getDatabaseName());
+        }
+    }
+
+    @Override
+    public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
+            throws DatabaseNotExistException, CatalogException {
+        try {
+            catalog.dropDatabase(tablePath.getDatabaseName(), ignoreIfNotExists, true);
+        } catch (Exception e) {
+            throw new DatabaseNotExistException(this.catalogName, tablePath.getDatabaseName());
+        }
+    }
+
+    private CatalogTable toCatalogTable(
+            FileStoreTable paimonFileStoreTableTable, TablePath tablePath) {
+        org.apache.paimon.schema.TableSchema schema = paimonFileStoreTableTable.schema();
+        List<DataField> dataFields = schema.fields();
+        TableSchema.Builder builder = TableSchema.builder();
+        dataFields.forEach(
+                dataField -> {
+                    String name = dataField.name();
+                    SeaTunnelDataType<?> seaTunnelType =
+                            SchemaUtil.toSeaTunnelType(dataField.type());
+                    PhysicalColumn physicalColumn =
+                            PhysicalColumn.of(
+                                    name,
+                                    seaTunnelType,
+                                    (Long) null,
+                                    true,
+                                    null,
+                                    dataField.description());
+                    builder.column(physicalColumn);
+                });
+
+        List<String> partitionKeys = schema.partitionKeys();
+
+        return CatalogTable.of(
+                org.apache.seatunnel.api.table.catalog.TableIdentifier.of(
+                        catalogName, tablePath.getDatabaseName(), tablePath.getTableName()),
+                builder.build(),
+                paimonFileStoreTableTable.options(),
+                partitionKeys,
+                null,
+                catalogName);
+    }
+
+    private Identifier toIdentifier(TablePath tablePath) {
+        return Identifier.create(tablePath.getDatabaseName(), tablePath.getTableName());
+    }
+}

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogFactory.java

@@ -0,0 +1,54 @@
+/*
+ * 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.connectors.seatunnel.paimon.catalog;
+
+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 org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonSinkConfig;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(Factory.class)
+public class PaimonCatalogFactory implements CatalogFactory {
+    @Override
+    public Catalog createCatalog(String catalogName, ReadonlyConfig readonlyConfig) {
+        return new PaimonCatalog(catalogName, readonlyConfig);
+    }
+
+    @Override
+    public String factoryIdentifier() {
+        return "Paimon";
+    }
+
+    @Override
+    public OptionRule optionRule() {
+        return OptionRule.builder()
+                .required(
+                        PaimonSinkConfig.WAREHOUSE,
+                        PaimonSinkConfig.DATABASE,
+                        PaimonSinkConfig.TABLE)
+                .optional(
+                        PaimonSinkConfig.HDFS_SITE_PATH,
+                        PaimonSinkConfig.SCHEMA_SAVE_MODE,
+                        PaimonSinkConfig.DATA_SAVE_MODE)
+                .build();
+    }
+}

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogLoader.java

@@ -0,0 +1,61 @@
+/*
+ * 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.connectors.seatunnel.paimon.catalog;
+
+import org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonSinkConfig;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.options.Options;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonConfig.WAREHOUSE;
+
+@Slf4j
+public class PaimonCatalogLoader implements Serializable {
+    private PaimonSinkConfig config;
+
+    public PaimonCatalogLoader(PaimonSinkConfig config) {
+        this.config = config;
+    }
+
+    public Catalog loadCatalog() {
+        // When using the seatunel engine, set the current class loader to prevent loading failures
+        Thread.currentThread().setContextClassLoader(PaimonCatalogLoader.class.getClassLoader());
+        final String warehouse = config.getWarehouse();
+        final Map<String, String> optionsMap = new HashMap<>();
+        optionsMap.put(WAREHOUSE.key(), warehouse);
+        final Options options = Options.fromMap(optionsMap);
+        final Configuration hadoopConf = new Configuration();
+        String hdfsSitePathOptional = config.getHdfsSitePath();
+        if (StringUtils.isNotBlank(hdfsSitePathOptional)) {
+            hadoopConf.addResource(new Path(hdfsSitePathOptional));
+        }
+        final CatalogContext catalogContext = CatalogContext.create(options, hadoopConf);
+        return CatalogFactory.createCatalog(catalogContext);
+    }
+}

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonTable.java

@@ -0,0 +1,28 @@
+/*
+ * 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.connectors.seatunnel.paimon.catalog;
+
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
+import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
+
+import org.apache.paimon.table.Table;
+
+public interface PaimonTable {
+    Table getPaimonTable(TablePath tablePath) throws CatalogException, TableNotExistException;
+}

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonConfig.java

@@ -22,20 +22,27 @@
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 
+import java.io.Serializable;
 import java.util.List;
 
 /**
  * Utility class to store configuration options, used by {@link SeaTunnelSource} and {@link
  * SeaTunnelSink}.
  */
-public class PaimonConfig {
+public class PaimonConfig implements Serializable {
 
     public static final Option<String> WAREHOUSE =
             Options.key("warehouse")
                     .stringType()
                     .noDefaultValue()
                     .withDescription("The warehouse path of paimon");
 
+    public static final Option<String> CATALOG_NAME =
+            Options.key("catalog_name")
+                    .stringType()
+                    .defaultValue("paimon")
+                    .withDescription(" the iceberg catalog name");
+
     public static final Option<String> DATABASE =
             Options.key("database")
                     .stringType()