@Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        Preconditions.checkState(!overwrite || context.isBounded(),
                "Unbounded data stream doesn't support overwrite operation.");

        List<String> equalityColumns = tableSchema.getPrimaryKey()
                .map(UniqueConstraint::getColumns)
                .orElseGet(ImmutableList::of);
        LOG.info("iceberg sink running with equality columns {}", equalityColumns);
        final ReadableConfig tableOptions = Configuration.fromMap(catalogTable.getOptions());
        boolean multipleSink = tableOptions.get(SINK_MULTIPLE_ENABLE);
        boolean schemaChange = tableOptions.get(SINK_SCHEMA_CHANGE_ENABLE);
        boolean autoCreateTableWhenSnapshot = tableOptions.get(SINK_AUTO_CREATE_TABLE_WHEN_SNAPSHOT);
        String schemaChangePolicies = tableOptions.getOptional(SINK_SCHEMA_CHANGE_POLICIES).orElse(null);
        LOG.info("iceberg sink running with policy {}", tableOptions.get(SINK_MULTIPLE_SCHEMA_UPDATE_POLICY));
        if (multipleSink) {
            return (DataStreamSinkProvider) dataStream -> FlinkSink.forRowData(dataStream)
                    .overwrite(overwrite)
                    .appendMode(tableOptions.get(IGNORE_ALL_CHANGELOG))
                    .metric(tableOptions.get(INLONG_METRIC), tableOptions.get(INLONG_AUDIT),
                            tableOptions.get(AUDIT_KEYS))
                    .catalogLoader(catalogLoader)
                    .tableSchema(tableSchema)
                    .multipleSink(tableOptions.get(SINK_MULTIPLE_ENABLE))
                    .switchAppendUpsertEnable(tableOptions.get(SWITCH_APPEND_UPSERT_ENABLE))
                    .multipleSinkOption(MultipleSinkOption.builder()
                            .withFormat(tableOptions.get(SINK_MULTIPLE_FORMAT))
                            .withSparkEngineEnable(tableOptions.get(SINK_MULTIPLE_TYPE_MAP_COMPATIBLE_WITH_SPARK))
                            .withDatabasePattern(tableOptions.get(SINK_MULTIPLE_DATABASE_PATTERN))
                            .withTablePattern(tableOptions.get(SINK_MULTIPLE_TABLE_PATTERN))
                            .withSchemaUpdatePolicy(tableOptions.get(SINK_MULTIPLE_SCHEMA_UPDATE_POLICY))
                            .withPkAutoGenerated(tableOptions.get(SINK_MULTIPLE_PK_AUTO_GENERATED))
                            .build())
                    .writeParallelism(tableOptions.getOptional(WRITE_PARALLELISM).orElse(dataStream.getParallelism()))
                    .dirtyOptions(dirtyOptions)
                    .dirtySink(dirtySink)
                    .action(actionsProvider)
                    .tableOptions(tableOptions)
                    .distributionMode(DistributionMode.fromName(tableOptions.get(WRITE_DISTRIBUTION_MODE)))
                    .enableSchemaChange(schemaChange)
                    .schemaChangePolicies(schemaChangePolicies)
                    .autoCreateTableWhenSnapshot(autoCreateTableWhenSnapshot)
                    .append();
        } else {
            return (DataStreamSinkProvider) dataStream -> FlinkSink.forRowData(dataStream)
                    .tableLoader(tableLoader)
                    .tableSchema(tableSchema)
                    .writeParallelism(tableOptions.getOptional(WRITE_PARALLELISM).orElse(dataStream.getParallelism()))
                    .equalityFieldColumns(equalityColumns)
                    .overwrite(overwrite)
                    .appendMode(tableOptions.get(IGNORE_ALL_CHANGELOG))
                    .metric(tableOptions.get(INLONG_METRIC), tableOptions.get(INLONG_AUDIT),
                            tableOptions.get(AUDIT_KEYS))
                    .dirtyOptions(dirtyOptions)
                    .dirtySink(dirtySink)
                    .action(actionsProvider)
                    .tableOptions(tableOptions)
                    .distributionMode(DistributionMode.fromName(tableOptions.get(WRITE_DISTRIBUTION_MODE)))
                    .switchAppendUpsertEnable(tableOptions.get(SWITCH_APPEND_UPSERT_ENABLE))
                    .append();
        }
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
Add an option `sink.ignore.changelog` to support insert-only mode without primaryKey.
Add a table property `write.compact.enable` to support small file compact.
Add option `inlong.metric` and `metrics.audit.proxy.hosts` to support collect inlong metrics and audit
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java