presto-main/src/main/java/io/prestosql/sql/planner/optimizations/AddLocalExchanges.java/AddLocalExchanges/visitMarkDistinct
class AddLocalExchanges:        @Override
        public PlanWithProperties visitMarkDistinct(MarkDistinctNode node, StreamPreferredProperties parentPreferences)
        {
            // mark distinct requires that all data partitioned
            StreamPreferredProperties childRequirements = parentPreferences
                    .constrainTo(node.getSource().getOutputSymbols())
                    .withDefaultParallelism(session)
                    .withPartitioning(node.getDistinctSymbols());

            PlanWithProperties child = planAndEnforce(node.getSource(), childRequirements, childRequirements);

            MarkDistinctNode result = new MarkDistinctNode(
                    node.getId(),
                    child.getNode(),
                    node.getMarkerSymbol(),
                    pruneMarkDistinctSymbols(node, child.getProperties().getLocalProperties()),
                    node.getHashSymbol());

            return deriveProperties(result, child.getProperties());
        }