class PKLookupPhase extends AbstractProjectionsPhase implements CollectPhase {

    private final List<ColumnIdent> partitionedByColumns;
    private final List<Symbol> toCollect;
    private final Map<String, Map<ShardId, List<PKAndVersion>>> idsByShardByNode;
    private DistributionInfo distInfo = DistributionInfo.DEFAULT_BROADCAST;

    public PKLookupPhase(UUID jobId,
                         int phaseId,
                         List<ColumnIdent> partitionedByColumns,
                         List<Symbol> toCollect,
                         Map<String, Map<ShardId, List<PKAndVersion>>> idsByShardByNode) {
        super(jobId, phaseId, "pkLookup", Collections.emptyList());
        assert toCollect.stream().noneMatch(
            st -> SymbolVisitors.any(s -> s instanceof ScopedSymbol || s instanceof SelectSymbol, st))
            : "toCollect must not contain any fields or selectSymbols: " + toCollect;
        this.partitionedByColumns = partitionedByColumns;
        this.toCollect = toCollect;
        this.idsByShardByNode = idsByShardByNode;
    }

    public PKLookupPhase(StreamInput in) throws IOException {
        super(in);
        distInfo = new DistributionInfo(in);
        toCollect = Symbols.listFromStream(in);

        int numNodes = in.readVInt();
        idsByShardByNode = new HashMap<>(numNodes);
        for (int nodeIdx = 0; nodeIdx < numNodes; nodeIdx++) {
            String nodeId = in.readString();
            int numShards = in.readVInt();
            HashMap<ShardId, List<PKAndVersion>> idsByShard = new HashMap<>(numShards);
            idsByShardByNode.put(nodeId, idsByShard);
            for (int shardIdx = 0; shardIdx < numShards; shardIdx++) {
                ShardId shardId = new ShardId(in);
                int numPks = in.readVInt();
                ArrayList<PKAndVersion> pks = new ArrayList<>(numPks);
                for (int pkIdx = 0; pkIdx < numPks; pkIdx++) {
                    pks.add(new PKAndVersion(in));
                }
                idsByShard.put(shardId, pks);
            }
        }

        int numPartitionedByCols = in.readVInt();
        if (numPartitionedByCols == 0) {
            partitionedByColumns = Collections.emptyList();
        } else {
            partitionedByColumns = new ArrayList<>(numPartitionedByCols);
            for (int i = 0; i < numPartitionedByCols; i++) {
                partitionedByColumns.add(new ColumnIdent(in));
            }
        }
    }

    @Override
    public void writeTo(StreamOutput out) throws IOException {<FILL_FUNCTION_BODY>}

    @Override
    public List<DataType<?>> outputTypes() {
        if (projections.isEmpty()) {
            return Symbols.typeView(toCollect);
        }
        return super.outputTypes();
    }

    @Override
    public List<Symbol> toCollect() {
        return toCollect;
    }

    @Override
    public DistributionInfo distributionInfo() {
        return distInfo;
    }

    @Override
    public void distributionInfo(DistributionInfo distributionInfo) {
        this.distInfo = distributionInfo;
    }

    @Override
    public Type type() {
        return Type.PKLookup;
    }

    @Override
    public Collection<String> nodeIds() {
        return idsByShardByNode.keySet();
    }

    @Override
    public <C, R> R accept(ExecutionPhaseVisitor<C, R> visitor, C context) {
        return visitor.visitPKLookup(this, context);
    }

    public Map<ShardId,List<PKAndVersion>> getIdsByShardId(String nodeId) {
        return idsByShardByNode.getOrDefault(nodeId, Collections.emptyMap());
    }

    public List<ColumnIdent> partitionedByColumns() {
        return partitionedByColumns;
    }
}

        super.writeTo(out);
        distInfo.writeTo(out);
        Symbols.toStream(toCollect, out);

        out.writeVInt(idsByShardByNode.size());
        for (Map.Entry<String, Map<ShardId, List<PKAndVersion>>> byNodeEntry : idsByShardByNode.entrySet()) {
            Map<ShardId, List<PKAndVersion>> idsByShard = byNodeEntry.getValue();
            out.writeString(byNodeEntry.getKey());
            out.writeVInt(idsByShard.size());
            for (Map.Entry<ShardId, List<PKAndVersion>> shardEntry : idsByShard.entrySet()) {
                List<PKAndVersion> ids = shardEntry.getValue();

                shardEntry.getKey().writeTo(out);
                out.writeVInt(ids.size());
                for (PKAndVersion id : ids) {
                    id.writeTo(out);
                }
            }
        }

        out.writeVInt(partitionedByColumns.size());
        for (ColumnIdent partitionedByColumn : partitionedByColumns) {
            partitionedByColumn.writeTo(out);
        }
    
<methods>public void addProjection(io.crate.execution.dsl.projection.Projection) ,public boolean equals(java.lang.Object) ,public Optional<io.crate.execution.dsl.projection.Projection> finalProjection() ,public boolean hasProjections() ,public int hashCode() ,public java.util.UUID jobId() ,public java.lang.String name() ,public List<DataType<?>> outputTypes() ,public int phaseId() ,public List<io.crate.execution.dsl.projection.Projection> projections() ,public java.lang.String toString() ,public void writeTo(org.elasticsearch.common.io.stream.StreamOutput) throws java.io.IOException<variables>private final non-sealed int executionPhaseId,private final non-sealed java.util.UUID jobId,private final non-sealed java.lang.String name,protected List<DataType<?>> outputTypes,protected final non-sealed List<io.crate.execution.dsl.projection.Projection> projections