class TableFetchInfo {

        private final Streamer<?>[] streamers;
        private final Collection<Reference> refs;
        private final FetchTask fetchTask;

        TableFetchInfo(Collection<Reference> refs, FetchTask fetchTask) {
            this.refs = refs;
            this.fetchTask = fetchTask;
            this.streamers = Symbols.streamerArray(refs);
        }

        FetchCollector createCollector(int readerId, RamAccounting ramAccounting) {
            IndexService indexService = fetchTask.indexService(readerId);
            var mapperService = indexService.mapperService();
            LuceneReferenceResolver resolver = new LuceneReferenceResolver(
                indexService.index().getName(),
                mapperService::fieldType,
                fetchTask.table(readerId).partitionedByColumns()
            );
            ArrayList<LuceneCollectorExpression<?>> exprs = new ArrayList<>(refs.size());
            for (Reference reference : refs) {
                exprs.add(resolver.getImplementation(reference));
            }
            return new FetchCollector(
                exprs,
                streamers,
                fetchTask,
                ramAccounting,
                readerId
            );
        }
    }

    public NodeFetchOperation(ThreadPoolExecutor executor,
                              int numProcessors,
                              JobsLogs jobsLogs,
                              TasksService tasksService,
                              CircuitBreaker circuitBreaker) {
        this.executor = executor;
        this.numProcessors = numProcessors;
        this.jobsLogs = jobsLogs;
        this.tasksService = tasksService;
        this.circuitBreaker = circuitBreaker;
    }

    public CompletableFuture<? extends IntObjectMap<StreamBucket>> fetch(UUID jobId,
                                                                         int phaseId,
                                                                         @Nullable IntObjectMap<IntArrayList> docIdsToFetch,
                                                                         boolean closeTaskOnFinish) {<FILL_FUNCTION_BODY>

        if (docIdsToFetch == null) {
            if (closeTaskOnFinish) {
                tryCloseTask(jobId, phaseId);
            }
            jobsLogs.operationStarted(phaseId, jobId, "fetch", () -> -1);
            jobsLogs.operationFinished(phaseId, jobId, null);
            return CompletableFuture.completedFuture(new IntObjectHashMap<>(0));
        }

        RootTask context = tasksService.getTask(jobId);
        FetchTask fetchTask = context.getTask(phaseId);
        jobsLogs.operationStarted(phaseId, jobId, "fetch", () -> -1);
        BiConsumer<? super IntObjectMap<StreamBucket>, ? super Throwable> whenComplete = (res, err) -> {
            if (closeTaskOnFinish) {
                if (err == null) {
                    fetchTask.close();
                } else {
                    fetchTask.kill(err);
                }
            }
            if (err == null) {
                jobsLogs.operationFinished(phaseId, jobId, null);
            } else {
                jobsLogs.operationFinished(phaseId, jobId, SQLExceptions.messageOf(err));
            }
        };
        try {
            return doFetch(fetchTask, docIdsToFetch).whenComplete(whenComplete);
        } catch (Throwable t) {
            whenComplete.accept(null, t);
            return CompletableFuture.failedFuture(t);
        }
    
<no_super_class>