class FileCollectSource implements CollectSource {

    private final ClusterService clusterService;
    private final Map<String, FileInputFactory> fileInputFactoryMap;
    private final InputFactory inputFactory;
    private final NodeContext nodeCtx;
    private final ThreadPool threadPool;
    private final Roles roles;

    @Inject
    public FileCollectSource(NodeContext nodeCtx,
                             ClusterService clusterService,
                             Map<String, FileInputFactory> fileInputFactoryMap,
                             ThreadPool threadPool,
                             Roles roles) {
        this.fileInputFactoryMap = fileInputFactoryMap;
        this.nodeCtx = nodeCtx;
        this.inputFactory = new InputFactory(nodeCtx);
        this.clusterService = clusterService;
        this.threadPool = threadPool;
        this.roles = roles;
    }

    @Override
    public CompletableFuture<BatchIterator<Row>> getIterator(TransactionContext txnCtx,
                                                             CollectPhase collectPhase,
                                                             CollectTask collectTask,
                                                             boolean supportMoveToStart) {
        FileUriCollectPhase fileUriCollectPhase = (FileUriCollectPhase) collectPhase;
        InputFactory.Context<LineCollectorExpression<?>> ctx =
            inputFactory.ctxForRefs(txnCtx, FileLineReferenceResolver::getImplementation);
        ctx.add(collectPhase.toCollect());

        Role user = requireNonNull(roles.findUser(txnCtx.sessionSettings().userName()), "User who invoked a statement must exist");
        List<URI> fileUris = targetUriToStringList(txnCtx, nodeCtx, fileUriCollectPhase.targetUri()).stream()
            .map(s -> {
                var uri = FileReadingIterator.toURI(s);
                if (uri.getScheme().equals("file") && user.isSuperUser() == false) {
                    throw new UnauthorizedException("Only a superuser can read from the local file system");
                }
                return uri;
            })
            .toList();
        FileReadingIterator fileReadingIterator = new FileReadingIterator(
            fileUris,
            fileUriCollectPhase.compression(),
            fileInputFactoryMap,
            fileUriCollectPhase.sharedStorage(),
            fileUriCollectPhase.nodeIds().size(),
            getReaderNumber(fileUriCollectPhase.nodeIds(), clusterService.state().nodes().getLocalNodeId()),
            fileUriCollectPhase.withClauseOptions(),
            threadPool.scheduler()
        );
        CopyFromParserProperties parserProperties = fileUriCollectPhase.parserProperties();
        LineProcessor lineProcessor = new LineProcessor(
            parserProperties.skipNumLines() > 0
                ? new SkippingBatchIterator<>(fileReadingIterator, (int) parserProperties.skipNumLines())
                : fileReadingIterator,
            ctx.topLevelInputs(),
            ctx.expressions(),
            fileUriCollectPhase.inputFormat(),
            parserProperties,
            fileUriCollectPhase.targetColumns()
        );
        return CompletableFuture.completedFuture(lineProcessor);
    }

    @VisibleForTesting
    public static int getReaderNumber(Collection<String> nodeIds, String localNodeId) {
        String[] readers = nodeIds.toArray(new String[0]);
        Arrays.sort(readers);
        return Arrays.binarySearch(readers, localNodeId);
    }

    private static List<String> targetUriToStringList(TransactionContext txnCtx,
                                                      NodeContext nodeCtx,
                                                      Symbol targetUri) {<FILL_FUNCTION_BODY>}
}

        Object value = SymbolEvaluator.evaluate(txnCtx, nodeCtx, targetUri, Row.EMPTY, SubQueryResults.EMPTY);
        if (targetUri.valueType().id() == DataTypes.STRING.id()) {
            String uri = (String) value;
            return Collections.singletonList(uri);
        } else if (DataTypes.STRING_ARRAY.equals(targetUri.valueType())) {
            return DataTypes.STRING_ARRAY.implicitCast(value);
        }

        // this case actually never happens because the check is already done in the analyzer
        throw AnalyzedCopyFrom.raiseInvalidType(targetUri.valueType());
    
<no_super_class>