class PKLookupOperation {

    private final IndicesService indicesService;
    private final ShardCollectSource shardCollectSource;

    public PKLookupOperation(IndicesService indicesService, ShardCollectSource shardCollectSource) {
        this.indicesService = indicesService;
        this.shardCollectSource = shardCollectSource;
    }

    @Nullable
    public static Doc lookupDoc(IndexShard shard,
                                String id,
                                long version,
                                VersionType versionType,
                                long seqNo,
                                long primaryTerm,
                                @Nullable SourceParser sourceParser) {
        Term uidTerm = new Term(IdFieldMapper.NAME, Uid.encodeId(id));
        Engine.Get get = new Engine.Get(id, uidTerm)
            .version(version)
            .versionType(versionType)
            .setIfSeqNo(seqNo)
            .setIfPrimaryTerm(primaryTerm);

        try (Engine.GetResult getResult = shard.get(get)) {
            var docIdAndVersion = getResult.docIdAndVersion();
            if (docIdAndVersion == null) {
                return null;
            }
            SourceFieldVisitor visitor = new SourceFieldVisitor();
            try {
                StoredFields storedFields = docIdAndVersion.reader.storedFields();
                storedFields.document(docIdAndVersion.docId, visitor);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
            Map<String, Object> sourceMap;
            if (sourceParser == null) {
                sourceMap = Map.of();
            } else {
                sourceMap = sourceParser.parse(visitor.source());
            }
            return new Doc(
                docIdAndVersion.docId,
                shard.shardId().getIndexName(),
                id,
                docIdAndVersion.version,
                docIdAndVersion.seqNo,
                docIdAndVersion.primaryTerm,
                sourceMap,
                () -> visitor.source().utf8ToString()
            );
        }
    }

    public BatchIterator<Row> lookup(UUID jobId,
                                     TransactionContext txnCtx,
                                     Supplier<RamAccounting> ramAccountingSupplier,
                                     Supplier<MemoryManager> memoryManagerSupplier,
                                     boolean ignoreMissing,
                                     Map<ShardId, List<PKAndVersion>> idsByShard,
                                     Collection<? extends Projection> projections,
                                     boolean requiresScroll,
                                     Function<Doc, Row> resultToRow,
                                     SourceParser sourceParser) {<FILL_FUNCTION_BODY>}
}

        ArrayList<BatchIterator<Row>> iterators = new ArrayList<>(idsByShard.size());
        for (Map.Entry<ShardId, List<PKAndVersion>> idsByShardEntry : idsByShard.entrySet()) {
            ShardId shardId = idsByShardEntry.getKey();
            IndexService indexService = indicesService.indexService(shardId.getIndex());
            if (indexService == null) {
                if (ignoreMissing) {
                    continue;
                }
                throw new IndexNotFoundException(shardId.getIndex());
            }
            IndexShard shard = indexService.getShardOrNull(shardId.id());
            if (shard == null) {
                if (ignoreMissing) {
                    continue;
                }
                throw new ShardNotFoundException(shardId);
            }
            Stream<Row> rowStream = idsByShardEntry.getValue().stream()
                .map(pkAndVersion -> lookupDoc(
                    shard,
                    pkAndVersion.id(),
                    pkAndVersion.version(),
                    VersionType.EXTERNAL,
                    pkAndVersion.seqNo(),
                    pkAndVersion.primaryTerm(),
                    sourceParser
                ))
                .filter(Objects::nonNull)
                .map(resultToRow);

            if (projections.isEmpty()) {
                final Iterable<Row> rowIterable = requiresScroll
                    ? rowStream.map(row -> new RowN(row.materialize())).collect(Collectors.toList())
                    : rowStream::iterator;
                iterators.add(InMemoryBatchIterator.of(rowIterable, SentinelRow.SENTINEL, true));
            } else {
                ProjectorFactory projectorFactory;
                try {
                    projectorFactory = shardCollectSource.getProjectorFactory(shardId);
                } catch (ShardNotFoundException e) {
                    if (ignoreMissing) {
                        continue;
                    }
                    throw e;
                }
                Projectors projectors = new Projectors(
                    projections,
                    jobId,
                    txnCtx,
                    ramAccountingSupplier.get(),
                    memoryManagerSupplier.get(),
                    projectorFactory);
                final Iterable<Row> rowIterable = requiresScroll && !projectors.providesIndependentScroll()
                    ? rowStream.map(row -> new RowN(row.materialize())).collect(Collectors.toList())
                    : rowStream::iterator;
                iterators.add(projectors.wrap(InMemoryBatchIterator.of(rowIterable, SentinelRow.SENTINEL, true)));
            }
        }
        return CompositeBatchIterator.seqComposite(iterators);
    
<no_super_class>