class FetchRows {

    public static FetchRows create(TransactionContext txnCtx,
                                   NodeContext nodeCtx,
                                   Map<RelationName, FetchSource> fetchSourceByTable,
                                   List<Symbol> outputSymbols) {
        IntArrayList fetchIdPositions = new IntArrayList();
        ArrayList<Object[]> nullRows = new ArrayList<>();
        IntObjectHashMap<UnsafeArrayRow> fetchedRows = new IntObjectHashMap<>();
        for (var fetchSource : fetchSourceByTable.values()) {
            Object[] nullRow = new Object[fetchSource.references().size()];
            for (InputColumn ic : fetchSource.fetchIdCols()) {
                fetchIdPositions.add(ic.index());
                nullRows.add(nullRow);
                fetchedRows.put(ic.index(), new UnsafeArrayRow());
            }
        }
        final UnsafeArrayRow inputRow = new UnsafeArrayRow();
        var visitor = new BaseImplementationSymbolVisitor<Void>(txnCtx, nodeCtx) {

            @Override
            public Input<?> visitInputColumn(final InputColumn inputColumn, final Void context) {
                final int idx = inputColumn.index();
                return () -> inputRow.get(idx);
            }

            @Override
            public Input<?> visitFetchReference(final FetchReference fetchReference, final Void context) {
                var ref = fetchReference.ref();
                UnsafeArrayRow row = fetchedRows.get(fetchReference.fetchId().index());
                int posInFetchedRow = fetchSourceByTable.get(ref.ident().tableIdent()).references().indexOf(ref);
                return () -> row.get(posInFetchedRow);
            }
        };
        List<Input<?>> outputExpressions = Lists.map(outputSymbols, x -> x.accept(visitor, null));
        return new FetchRows(fetchIdPositions, outputExpressions, inputRow, fetchedRows, nullRows);
    }

    /**
     * The indices in the incoming cells where there is a _fetchId value
     */
    private final int[] fetchIdPositions;

    /**
     * Row that is always mutated to contain the values of the current incoming row
     **/
    private final UnsafeArrayRow inputRow;

    /**
     * Outgoing row. Encapsulates the value retrieval from incoming row and readerBuckets.
     **/
    private final Row output;

    private final IntObjectHashMap<UnsafeArrayRow> fetchedRows;


    /**
     * Contains an entry per fetchIdPosition.
     * Each entry contains only null values; the length matches `fetchSource.references().size`.
     *
     * This is used as substitute fetchedRow if the _fetchId in the incoming row is null.
     */
    private final ArrayList<Object[]> nullRows;


    public FetchRows(IntArrayList fetchIdPositions,
                     List<Input<?>> outputExpressions,
                     UnsafeArrayRow inputRow,
                     IntObjectHashMap<UnsafeArrayRow> fetchedRows,
                     ArrayList<Object[]> nullRows) {
        this.fetchedRows = fetchedRows;
        this.nullRows = nullRows;
        this.fetchIdPositions = fetchIdPositions.toArray();
        this.output = new InputRow(outputExpressions);
        this.inputRow = inputRow;
    }

    public Row updatedOutputRow(Object[] incomingCells, IntFunction<ReaderBucket> getReaderBucket) {<FILL_FUNCTION_BODY>}

    public int[] fetchIdPositions() {
        return fetchIdPositions;
    }
}

        for (int i = 0; i < fetchIdPositions.length; i++) {
            int fetchIdPos = fetchIdPositions[i];
            Long fetchId = (Long) incomingCells[fetchIdPos];
            UnsafeArrayRow fetchedRow = fetchedRows.get(fetchIdPos);
            if (fetchId == null) {
                fetchedRow.cells(nullRows.get(i));
            } else {
                int readerId = FetchId.decodeReaderId(fetchId);
                int docId = FetchId.decodeDocId(fetchId);
                ReaderBucket readerBucket = getReaderBucket.apply(readerId);
                Object[] cells = readerBucket.get(docId);
                assert cells != null : "Must have cells in readerBucket docId=" + docId + ", readerId=" + readerId;
                fetchedRow.cells(cells);
            }
        }
        inputRow.cells(incomingCells);
        return output;
    
<no_super_class>