@Override
    public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
        IcebergSource icebergSource = (IcebergSource) streamNodeInfo;
        FieldTypeMappingStrategy fieldTypeMappingStrategy =
                fieldTypeStrategyFactory.getInstance(icebergSource.getSourceType());
        List<FieldInfo> fieldInfos = parseStreamFieldInfos(icebergSource.getFieldList(), icebergSource.getSourceName(),
                fieldTypeMappingStrategy);
        Map<String, String> properties = parseProperties(icebergSource.getProperties());

        return new IcebergExtractNode(icebergSource.getSourceName(),
                icebergSource.getSourceName(),
                fieldInfos,
                null,
                icebergSource.getUri(),
                icebergSource.getWarehouse(),
                icebergSource.getDatabase(),
                icebergSource.getTableName(),
                CatalogType.HIVE,
                "HIVE",
                icebergSource.getPrimaryKey(),
                null,
                properties);
    }
The Provider for creating Iceberg load nodes.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/IcebergProvider.java