@Override
    public LoadNode createLoadNode(StreamNode nodeInfo, Map<String, StreamField> constantFieldMap) {
        IcebergSink icebergSink = (IcebergSink) nodeInfo;
        Map<String, String> properties = parseProperties(icebergSink.getProperties());
        FieldTypeMappingStrategy fieldTypeMappingStrategy =
                fieldTypeStrategyFactory.getInstance(icebergSink.getSinkType());
        List<FieldInfo> fieldInfos = parseSinkFieldInfos(icebergSink.getSinkFieldList(), icebergSink.getSinkName(),
                fieldTypeMappingStrategy);
        List<FieldRelation> fieldRelations = parseSinkFields(icebergSink.getSinkFieldList(), constantFieldMap);
        IcebergConstant.CatalogType catalogType = CatalogType.forName(icebergSink.getCatalogType());
        Format format = parsingSinkMultipleFormat(icebergSink.getSinkMultipleEnable(),
                icebergSink.getSinkMultipleFormat());
        return new IcebergLoadNode(
                icebergSink.getSinkName(),
                icebergSink.getSinkName(),
                fieldInfos,
                fieldRelations,
                null,
                null,
                null,
                properties,
                icebergSink.getDbName(),
                icebergSink.getTableName(),
                icebergSink.getPrimaryKey(),
                catalogType,
                icebergSink.getCatalogUri(),
                icebergSink.getWarehouse(),
                icebergSink.getAppendMode(),
                icebergSink.getSinkMultipleEnable(),
                format,
                icebergSink.getDatabasePattern(),
                icebergSink.getTablePattern(),
                icebergSink.getEnableSchemaChange(),
                null);
    }
The Provider for creating Iceberg load nodes.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/IcebergProvider.java