class ForeignDataWrappers implements CollectSource {

    public static Setting<Boolean> ALLOW_LOCAL = Setting.boolSetting(
        "fdw.allow_local",
        false,
        Property.NodeScope,
        Property.Exposed
    );

    private final ClusterService clusterService;
    private final InputFactory inputFactory;
    private final Map<String, ForeignDataWrapper> wrappers;
    private final Roles roles;

    @Inject
    public ForeignDataWrappers(Settings settings,
                               ClusterService clusterService,
                               NodeContext nodeContext) {
        this.clusterService = clusterService;
        this.inputFactory = new InputFactory(nodeContext);
        this.wrappers = Map.of(
            "jdbc", new JdbcForeignDataWrapper(settings, inputFactory)
        );
        this.roles = nodeContext.roles();
    }

    public boolean contains(String fdw) {
        return wrappers.containsKey(fdw);
    }

    public ForeignDataWrapper get(String fdw) {
        var foreignDataWrapper = wrappers.get(fdw);
        if (foreignDataWrapper == null) {
            throw new IllegalArgumentException(
                "foreign-data wrapper " + fdw + " does not exist");
        }
        return foreignDataWrapper;
    }

    @Override
    public CompletableFuture<BatchIterator<Row>> getIterator(TransactionContext txnCtx,
                                                             CollectPhase collectPhase,
                                                             CollectTask collectTask,
                                                             boolean supportMoveToStart) {<FILL_FUNCTION_BODY>}
}

        if (!(collectPhase instanceof ForeignCollectPhase phase)) {
            throw new IllegalArgumentException(
                "ForeignDataWrappers requires ForeignCollectPhase, not: " + collectPhase);
        }
        Metadata metadata = clusterService.state().metadata();
        ForeignTablesMetadata foreignTables = metadata.custom(ForeignTablesMetadata.TYPE);
        if (foreignTables == null) {
            throw new RelationUnknown(phase.relationName());
        }
        ForeignTable foreignTable = foreignTables.get(phase.relationName());
        if (foreignTable == null) {
            throw new RelationUnknown(phase.relationName());
        }
        ServersMetadata servers = metadata.custom(ServersMetadata.TYPE);
        if (servers == null) {
            throw new ResourceNotFoundException(
                String.format(Locale.ENGLISH, "Server `%s` not found", foreignTable.server()));
        }
        Server server = servers.get(foreignTable.server());
        ForeignDataWrapper fdw = wrappers.get(server.fdw());
        if (fdw == null) {
            throw new ResourceNotFoundException(String.format(
                Locale.ENGLISH,
                "Foreign data wrapper '%s' used by server '%s' no longer exists",
                server.fdw(),
                foreignTable.server()
            ));
        }
        return fdw.getIterator(
            requireNonNull(roles.findUser(txnCtx.sessionSettings().userName()), "current user must exit"),
            server,
            foreignTable,
            txnCtx,
            collectPhase.toCollect(),
            phase.query()
        );
    
<no_super_class>