presto-main/src/main/java/io/prestosql/operator/aggregation/GenericAccumulatorFactory.java/DistinctingAccumulator/addInput
class DistinctingAccumulator:        @Override
        public void addInput(Page page)
        {
            // 1. filter out positions based on mask, if present
            Page filtered = maskChannel
                    .map(channel -> filter(page, page.getBlock(channel)))
                    .orElse(page);

            if (filtered.getPositionCount() == 0) {
                return;
            }

            // 2. compute a distinct mask
            Work<Block> work = hash.markDistinctRows(filtered);
            checkState(work.process());
            Block distinctMask = work.getResult();

            // 3. feed a Page with a new mask to the underlying aggregation
            accumulator.addInput(filtered.prependColumn(distinctMask));
        }