@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put(IcebergConstant.CONNECTOR_KEY, IcebergConstant.CONNECTOR);
        options.put(IcebergConstant.DATABASE_KEY, dbName);
        options.put(IcebergConstant.TABLE_KEY, tableName);
        options.put(IcebergConstant.CATALOG_TYPE_KEY, catalogType.name());
        options.put(IcebergConstant.CATALOG_NAME_KEY, catalogName);
        // support streaming only
        options.put(IcebergConstant.STREAMING, "true");
        options.put(IcebergConstant.STARTING_STRATEGY_KEY,
                IcebergConstant.StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL.name());
        if (null != uri) {
            options.put(IcebergConstant.URI_KEY, uri);
        }
        if (null != warehouse) {
            options.put(IcebergConstant.WAREHOUSE_KEY, warehouse);
        }
        if (null != startSnapShotId) {
            options.put(IcebergConstant.START_SNAPSHOT_ID, startSnapShotId.toString());
            options.put(IcebergConstant.STARTING_STRATEGY_KEY,
                    IcebergConstant.StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_ID.name());
        }
        return options;
    }
Iceberg extract node for extract data from iceberg
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/IcebergExtractNode.java