presto-main/src/main/java/io/prestosql/operator/MarkDistinctOperator.java/MarkDistinctOperator/addInput
class MarkDistinctOperator:    @Override
    public void addInput(Page page)
    {
        requireNonNull(page, "page is null");
        checkState(needsInput());

        inputPage = page;

        unfinishedWork = markDistinctHash.markDistinctRows(page);
        updateMemoryReservation();
    }