[feature](iceberg)Support write to iceberg for 2.1

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java

@@ -22,6 +22,7 @@
 import org.apache.doris.datasource.SessionContext;
 import org.apache.doris.datasource.operations.ExternalMetadataOperations;
 import org.apache.doris.datasource.property.PropertyConverter;
+import org.apache.doris.transaction.TransactionManagerFactory;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.Constants;
@@ -51,7 +52,9 @@ public IcebergExternalCatalog(long catalogId, String name, String comment) {
     @Override
     protected void initLocalObjectsImpl() {
         initCatalog();
-        metadataOps = ExternalMetadataOperations.newIcebergMetadataOps(this, catalog);
+        IcebergMetadataOps ops = ExternalMetadataOperations.newIcebergMetadataOps(this, catalog);
+        transactionManager = TransactionManagerFactory.createIcebergTransactionManager(ops);
+        metadataOps = ops;
     }
 
     public Catalog getCatalog() {

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java

@@ -28,9 +28,14 @@
 import org.apache.doris.thrift.TTableDescriptor;
 import org.apache.doris.thrift.TTableType;
 
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.Table;
+
 import java.util.HashMap;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 public class IcebergExternalTable extends ExternalTable {
 
@@ -83,4 +88,15 @@ public long fetchRowCount() {
         makeSureInitialized();
         return IcebergUtils.getIcebergRowCount(getCatalog(), getDbName(), getName());
     }
+
+    public Table getIcebergTable() {
+        return IcebergUtils.getIcebergTable(getCatalog(), getDbName(), getName());
+    }
+
+    @Override
+    public Set<String> getPartitionNames() {
+        getIcebergTable();
+        return IcebergUtils.getIcebergTable(getCatalog(), getDbName(), getName())
+                .spec().fields().stream().map(PartitionField::name).collect(Collectors.toSet());
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java

@@ -0,0 +1,186 @@
+// 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.
+// This file is copied from
+// https://github.com/trinodb/trino/blob/438/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
+// and modified by Doris
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.thrift.TFileContent;
+import org.apache.doris.thrift.TIcebergCommitData;
+import org.apache.doris.transaction.Transaction;
+
+import com.google.common.base.VerifyException;
+import com.google.common.collect.Lists;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+public class IcebergTransaction implements Transaction {
+
+    private static final Logger LOG = LogManager.getLogger(IcebergTransaction.class);
+    private final IcebergMetadataOps ops;
+    private org.apache.iceberg.Transaction transaction;
+    private final List<TIcebergCommitData> commitDataList = Lists.newArrayList();
+
+    public IcebergTransaction(IcebergMetadataOps ops) {
+        this.ops = ops;
+    }
+
+    public void updateIcebergCommitData(List<TIcebergCommitData> commitDataList) {
+        synchronized (this) {
+            this.commitDataList.addAll(commitDataList);
+        }
+    }
+
+    public void beginInsert(String dbName, String tbName) {
+        Table icebergTable = ops.getCatalog().loadTable(TableIdentifier.of(dbName, tbName));
+        transaction = icebergTable.newTransaction();
+    }
+
+    public void finishInsert() {
+        Table icebergTable = transaction.table();
+        AppendFiles appendFiles = transaction.newAppend();
+
+        for (CommitTaskData task : convertToCommitTaskData()) {
+            DataFiles.Builder builder = DataFiles.builder(icebergTable.spec())
+                    .withPath(task.getPath())
+                    .withFileSizeInBytes(task.getFileSizeInBytes())
+                    .withFormat(IcebergUtils.getFileFormat(icebergTable))
+                    .withMetrics(task.getMetrics());
+
+            if (icebergTable.spec().isPartitioned()) {
+                List<String> partitionValues = task.getPartitionValues()
+                        .orElseThrow(() -> new VerifyException("No partition data for partitioned table"));
+                builder.withPartitionValues(partitionValues);
+            }
+            appendFiles.appendFile(builder.build());
+        }
+
+        // in appendFiles.commit, it will generate metadata(manifest and snapshot)
+        // after appendFiles.commit, in current transaction, you can already see the new snapshot
+        appendFiles.commit();
+    }
+
+    public List<CommitTaskData> convertToCommitTaskData() {
+        List<CommitTaskData> commitTaskData = new ArrayList<>();
+        for (TIcebergCommitData data : this.commitDataList) {
+            commitTaskData.add(new CommitTaskData(
+                    data.getFilePath(),
+                    data.getFileSize(),
+                    new Metrics(
+                            data.getRowCount(),
+                            Collections.EMPTY_MAP,
+                            Collections.EMPTY_MAP,
+                            Collections.EMPTY_MAP,
+                            Collections.EMPTY_MAP
+                    ),
+                    data.isSetPartitionValues() ? Optional.of(data.getPartitionValues()) : Optional.empty(),
+                    convertToFileContent(data.getFileContent()),
+                    data.isSetReferencedDataFiles() ? Optional.of(data.getReferencedDataFiles()) : Optional.empty()
+            ));
+        }
+        return commitTaskData;
+    }
+
+    private FileContent convertToFileContent(TFileContent content) {
+        if (content.equals(TFileContent.DATA)) {
+            return FileContent.DATA;
+        } else if (content.equals(TFileContent.POSITION_DELETES)) {
+            return FileContent.POSITION_DELETES;
+        } else {
+            return FileContent.EQUALITY_DELETES;
+        }
+    }
+
+    @Override
+    public void commit() throws UserException {
+        // Externally readable
+        // Manipulate the relevant data so that others can also see the latest table, such as:
+        //   1. hadoop: it will change the version number information in 'version-hint.text'
+        //   2. hive: it will change the table properties, the most important thing is to revise 'metadata_location'
+        //   3. and so on ...
+        transaction.commitTransaction();
+    }
+
+    @Override
+    public void rollback() {
+
+    }
+
+    public long getUpdateCnt() {
+        return commitDataList.stream().mapToLong(TIcebergCommitData::getRowCount).sum();
+    }
+
+    public static class CommitTaskData {
+        private final String path;
+        private final long fileSizeInBytes;
+        private final Metrics metrics;
+        private final Optional<List<String>> partitionValues;
+        private final FileContent content;
+        private final Optional<List<String>> referencedDataFiles;
+
+        public CommitTaskData(String path,
+                              long fileSizeInBytes,
+                              Metrics metrics,
+                              Optional<List<String>> partitionValues,
+                              FileContent content,
+                              Optional<List<String>> referencedDataFiles) {
+            this.path = path;
+            this.fileSizeInBytes = fileSizeInBytes;
+            this.metrics = metrics;
+            this.partitionValues = partitionValues;
+            this.content = content;
+            this.referencedDataFiles = referencedDataFiles;
+        }
+
+        public String getPath() {
+            return path;
+        }
+
+        public long getFileSizeInBytes() {
+            return fileSizeInBytes;
+        }
+
+        public Metrics getMetrics() {
+            return metrics;
+        }
+
+        public Optional<List<String>> getPartitionValues() {
+            return partitionValues;
+        }
+
+        public FileContent getContent() {
+            return content;
+        }
+
+        public Optional<List<String>> getReferencedDataFiles() {
+            return referencedDataFiles;
+        }
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java

@@ -46,13 +46,15 @@
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.hive.HiveMetaStoreClientHelper;
+import org.apache.doris.nereids.exceptions.NotSupportedException;
 import org.apache.doris.thrift.TExprOpcode;
 
 import com.google.common.collect.Lists;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.expressions.And;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
@@ -61,6 +63,7 @@
 import org.apache.iceberg.expressions.Unbound;
 import org.apache.iceberg.types.Type.TypeID;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.LocationUtil;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -90,6 +93,13 @@ public Integer initialValue() {
     public static final String TOTAL_POSITION_DELETES = "total-position-deletes";
     public static final String TOTAL_EQUALITY_DELETES = "total-equality-deletes";
 
+    // nickname in flink and spark
+    public static final String WRITE_FORMAT = "write-format";
+    public static final String COMPRESSION_CODEC = "compression-codec";
+
+    // nickname in spark
+    public static final String SPARK_SQL_COMPRESSION_CODEC = "spark.sql.iceberg.compression-codec";
+
     public static Expression convertToIcebergExpr(Expr expr, Schema schema) {
         if (expr == null) {
             return null;
@@ -573,4 +583,51 @@ public static long getIcebergRowCount(ExternalCatalog catalog, String dbName, St
         }
         return -1;
     }
+
+    public static String getFileFormat(Table table) {
+        Map<String, String> properties = table.properties();
+        if (properties.containsKey(WRITE_FORMAT)) {
+            return properties.get(WRITE_FORMAT);
+        }
+        if (properties.containsKey(TableProperties.DEFAULT_FILE_FORMAT)) {
+            return properties.get(TableProperties.DEFAULT_FILE_FORMAT);
+        }
+        return TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+    }
+
+    public static String getFileCompress(Table table) {
+        Map<String, String> properties = table.properties();
+        if (properties.containsKey(COMPRESSION_CODEC)) {
+            return properties.get(COMPRESSION_CODEC);
+        } else if (properties.containsKey(SPARK_SQL_COMPRESSION_CODEC)) {
+            return properties.get(SPARK_SQL_COMPRESSION_CODEC);
+        }
+        String fileFormat = getFileFormat(table);
+        if (fileFormat.equalsIgnoreCase("parquet")) {
+            return properties.getOrDefault(
+                    TableProperties.PARQUET_COMPRESSION, TableProperties.PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0);
+        } else if (fileFormat.equalsIgnoreCase("orc")) {
+            return properties.getOrDefault(
+                    TableProperties.ORC_COMPRESSION, TableProperties.ORC_COMPRESSION_DEFAULT);
+        }
+        throw new NotSupportedException("Unsupported file format: " + fileFormat);
+    }
+
+    public static String dataLocation(Table table) {
+        Map<String, String> properties = table.properties();
+        if (properties.containsKey(TableProperties.WRITE_LOCATION_PROVIDER_IMPL)) {
+            throw new NotSupportedException(
+                "Table " + table.name() + " specifies " + properties.get(TableProperties.WRITE_LOCATION_PROVIDER_IMPL)
+                    + " as a location provider. "
+                    + "Writing to Iceberg tables with custom location provider is not supported.");
+        }
+        String dataLocation = properties.get(TableProperties.WRITE_DATA_LOCATION);
+        if (dataLocation == null) {
+            dataLocation = properties.get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION);
+            if (dataLocation == null) {
+                dataLocation = String.format("%s/data", LocationUtil.stripTrailingSlash(table.location()));
+            }
+        }
+        return dataLocation;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergApiSource.java

@@ -24,11 +24,11 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.datasource.iceberg.IcebergUtils;
 import org.apache.doris.planner.ColumnRange;
 import org.apache.doris.thrift.TFileAttributes;
 
 import org.apache.iceberg.Table;
-import org.apache.iceberg.TableProperties;
 
 import java.util.Map;
 
@@ -61,14 +61,7 @@ public TupleDescriptor getDesc() {
 
     @Override
     public String getFileFormat() {
-        Map<String, String> properties = originTable.properties();
-        if (properties.containsKey(TableProperties.DEFAULT_FILE_FORMAT)) {
-            return properties.get(TableProperties.DEFAULT_FILE_FORMAT);
-        }
-        if (properties.containsKey(FLINK_WRITE_FORMAT)) {
-            return properties.get(FLINK_WRITE_FORMAT);
-        }
-        return TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+        return IcebergUtils.getFileFormat(originTable);
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergHMSSource.java

@@ -26,12 +26,11 @@
 import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.hive.HMSExternalTable;
 import org.apache.doris.datasource.hive.source.HiveScanNode;
+import org.apache.doris.datasource.iceberg.IcebergUtils;
 import org.apache.doris.planner.ColumnRange;
 import org.apache.doris.thrift.TFileAttributes;
 import org.apache.doris.thrift.TFileTextScanRangeParams;
 
-import org.apache.iceberg.TableProperties;
-
 import java.util.Map;
 
 public class IcebergHMSSource implements IcebergSource {
@@ -59,14 +58,7 @@ public TupleDescriptor getDesc() {
 
     @Override
     public String getFileFormat() throws DdlException, MetaNotFoundException {
-        Map<String, String> properties = hmsTable.getRemoteTable().getParameters();
-        if (properties.containsKey(TableProperties.DEFAULT_FILE_FORMAT)) {
-            return properties.get(TableProperties.DEFAULT_FILE_FORMAT);
-        }
-        if (properties.containsKey(FLINK_WRITE_FORMAT)) {
-            return properties.get(FLINK_WRITE_FORMAT);
-        }
-        return TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+        return IcebergUtils.getFileFormat(icebergTable);
     }
 
     public org.apache.iceberg.Table getIcebergTable() throws MetaNotFoundException {

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSource.java

@@ -27,9 +27,6 @@
 
 public interface IcebergSource {
 
-    // compatible with flink, which is "write.format.default" in spark
-    String FLINK_WRITE_FORMAT = "write-format";
-
     TupleDescriptor getDesc();
 
     org.apache.iceberg.Table getIcebergTable() throws MetaNotFoundException;