[INLONG-8982][Sort] Add Iceberg sink on flink 1.15 (#9035)

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/IcebergLoadNode.java

@@ -17,8 +17,10 @@
 
 package org.apache.inlong.sort.protocol.node.load;
 
+import org.apache.inlong.common.enums.MetaField;
 import org.apache.inlong.sort.protocol.FieldInfo;
 import org.apache.inlong.sort.protocol.InlongMetric;
+import org.apache.inlong.sort.protocol.Metadata;
 import org.apache.inlong.sort.protocol.constant.IcebergConstant;
 import org.apache.inlong.sort.protocol.constant.IcebergConstant.CatalogType;
 import org.apache.inlong.sort.protocol.enums.FilterStrategy;
@@ -38,14 +40,16 @@
 import javax.annotation.Nullable;
 
 import java.io.Serializable;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 @JsonTypeName("icebergLoad")
 @Data
 @NoArgsConstructor
 @EqualsAndHashCode(callSuper = true)
-public class IcebergLoadNode extends LoadNode implements InlongMetric, Serializable {
+public class IcebergLoadNode extends LoadNode implements InlongMetric, Metadata, Serializable {
 
     private static final long serialVersionUID = -1L;
 
@@ -128,4 +132,13 @@ public List<FieldInfo> getPartitionFields() {
         return super.getPartitionFields();
     }
 
+    @Override
+    public boolean isVirtual(MetaField metaField) {
+        return true;
+    }
+
+    @Override
+    public Set<MetaField> supportedMetaFields() {
+        return EnumSet.of(MetaField.AUDIT_DATA_TIME);
+    }
 }

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkDynamicTableFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.inlong.sort.iceberg;
 
+import org.apache.inlong.sort.iceberg.sink.IcebergTableSink;
 import org.apache.inlong.sort.iceberg.source.IcebergTableSource;
 
 import org.apache.flink.configuration.ConfigOption;
@@ -38,7 +39,6 @@
 import org.apache.flink.util.Preconditions;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
-import org.apache.iceberg.flink.IcebergTableSink;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergWritableMetadata.java

@@ -0,0 +1,66 @@
+/*
+ * 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.inlong.sort.iceberg;
+
+import org.apache.inlong.sort.base.Constants;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import java.io.Serializable;
+
+public enum IcebergWritableMetadata {
+
+    AUDIT_DATA_TIME(
+            Constants.META_AUDIT_DATA_TIME,
+            DataTypes.BIGINT().notNull(),
+            (r, p) -> r.getLong(p)),
+
+    NULL(
+            "",
+            DataTypes.NULL(),
+            (r, p) -> null);
+
+    private final String key;
+    private final DataType dataType;
+    private final MetadataConverter converter;
+
+    public String getKey() {
+        return key;
+    }
+
+    public DataType getDataType() {
+        return dataType;
+    }
+
+    public MetadataConverter getConverter() {
+        return converter;
+    }
+
+    IcebergWritableMetadata(String key, DataType dataType, MetadataConverter converter) {
+        this.key = key;
+        this.dataType = dataType;
+        this.converter = converter;
+    }
+
+    public interface MetadataConverter extends Serializable {
+
+        Object read(RowData rowData, int pos);
+    }
+}

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/CommitSummary.java

@@ -0,0 +1,96 @@
+/*
+ * 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.inlong.sort.iceberg.sink;
+
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+
+import java.util.Arrays;
+import java.util.NavigableMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
+ */
+class CommitSummary {
+
+    private final AtomicLong dataFilesCount = new AtomicLong();
+    private final AtomicLong dataFilesRecordCount = new AtomicLong();
+    private final AtomicLong dataFilesByteCount = new AtomicLong();
+    private final AtomicLong deleteFilesCount = new AtomicLong();
+    private final AtomicLong deleteFilesRecordCount = new AtomicLong();
+    private final AtomicLong deleteFilesByteCount = new AtomicLong();
+
+    CommitSummary(NavigableMap<Long, WriteResult> pendingResults) {
+        pendingResults
+                .values()
+                .forEach(
+                        writeResult -> {
+                            dataFilesCount.addAndGet(writeResult.dataFiles().length);
+                            Arrays.stream(writeResult.dataFiles())
+                                    .forEach(
+                                            dataFile -> {
+                                                dataFilesRecordCount.addAndGet(dataFile.recordCount());
+                                                dataFilesByteCount.addAndGet(dataFile.fileSizeInBytes());
+                                            });
+                            deleteFilesCount.addAndGet(writeResult.deleteFiles().length);
+                            Arrays.stream(writeResult.deleteFiles())
+                                    .forEach(
+                                            deleteFile -> {
+                                                deleteFilesRecordCount.addAndGet(deleteFile.recordCount());
+                                                deleteFilesByteCount.addAndGet(deleteFile.fileSizeInBytes());
+                                            });
+                        });
+    }
+
+    long dataFilesCount() {
+        return dataFilesCount.get();
+    }
+
+    long dataFilesRecordCount() {
+        return dataFilesRecordCount.get();
+    }
+
+    long dataFilesByteCount() {
+        return dataFilesByteCount.get();
+    }
+
+    long deleteFilesCount() {
+        return deleteFilesCount.get();
+    }
+
+    long deleteFilesRecordCount() {
+        return deleteFilesRecordCount.get();
+    }
+
+    long deleteFilesByteCount() {
+        return deleteFilesByteCount.get();
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("dataFilesCount", dataFilesCount)
+                .add("dataFilesRecordCount", dataFilesRecordCount)
+                .add("dataFilesByteCount", dataFilesByteCount)
+                .add("deleteFilesCount", deleteFilesCount)
+                .add("deleteFilesRecordCount", deleteFilesRecordCount)
+                .add("deleteFilesByteCount", deleteFilesByteCount)
+                .toString();
+    }
+}

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/DeltaManifests.java

@@ -0,0 +1,74 @@
+/*
+ * 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.inlong.sort.iceberg.sink;
+
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+import java.util.List;
+
+/**
+ * Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
+ */
+class DeltaManifests {
+
+    private static final CharSequence[] EMPTY_REF_DATA_FILES = new CharSequence[0];
+
+    private final ManifestFile dataManifest;
+    private final ManifestFile deleteManifest;
+    private final CharSequence[] referencedDataFiles;
+
+    DeltaManifests(ManifestFile dataManifest, ManifestFile deleteManifest) {
+        this(dataManifest, deleteManifest, EMPTY_REF_DATA_FILES);
+    }
+
+    DeltaManifests(
+            ManifestFile dataManifest, ManifestFile deleteManifest, CharSequence[] referencedDataFiles) {
+        Preconditions.checkNotNull(referencedDataFiles, "Referenced data files shouldn't be null.");
+
+        this.dataManifest = dataManifest;
+        this.deleteManifest = deleteManifest;
+        this.referencedDataFiles = referencedDataFiles;
+    }
+
+    ManifestFile dataManifest() {
+        return dataManifest;
+    }
+
+    ManifestFile deleteManifest() {
+        return deleteManifest;
+    }
+
+    CharSequence[] referencedDataFiles() {
+        return referencedDataFiles;
+    }
+
+    List<ManifestFile> manifests() {
+        List<ManifestFile> manifests = Lists.newArrayListWithCapacity(2);
+        if (dataManifest != null) {
+            manifests.add(dataManifest);
+        }
+
+        if (deleteManifest != null) {
+            manifests.add(deleteManifest);
+        }
+
+        return manifests;
+    }
+}