@Override
  public Result apply(AggregationNode parent, Captures captures, Context context) {
    if (!shouldAddMarkDistinct(parent, context)) {
      return Result.empty();
    }

    // the distinct marker for the given set of input columns
    Map<Set<Symbol>, Symbol> markers = new HashMap<>();

    Map<Symbol, AggregationNode.Aggregation> newAggregations = new HashMap<>();
    PlanNode subPlan = parent.getChild();

    for (Map.Entry<Symbol, AggregationNode.Aggregation> entry :
        parent.getAggregations().entrySet()) {
      AggregationNode.Aggregation aggregation = entry.getValue();

      if (aggregation.isDistinct()
          && !aggregation.getFilter().isPresent()
          && !aggregation.getMask().isPresent()) {
        Set<Symbol> inputs = aggregation.getArguments().stream().map(Symbol::from).collect(toSet());

        Symbol marker = markers.get(inputs);
        if (marker == null) {
          marker =
              context
                  .getSymbolAllocator()
                  .newSymbol(Iterables.getLast(inputs).getName(), BOOLEAN, "distinct");
          markers.put(inputs, marker);

          ImmutableSet.Builder<Symbol> distinctSymbols =
              ImmutableSet.<Symbol>builder().addAll(parent.getGroupingKeys()).addAll(inputs);
          parent.getGroupIdSymbol().ifPresent(distinctSymbols::add);

          subPlan =
              new MarkDistinctNode(
                  context.getIdAllocator().genPlanNodeId(),
                  subPlan,
                  marker,
                  ImmutableList.copyOf(distinctSymbols.build()),
                  Optional.empty());
        }

        // remove the distinct flag and set the distinct marker
        newAggregations.put(
            entry.getKey(),
            new AggregationNode.Aggregation(
                aggregation.getResolvedFunction(),
                aggregation.getArguments(),
                false,
                aggregation.getFilter(),
                aggregation.getOrderingScheme(),
                Optional.of(marker)));
      } else {
        newAggregations.put(entry.getKey(), aggregation);
      }
    }

    return Result.ofPlanNode(
        AggregationNode.builderFrom(parent)
            .setSource(subPlan)
            .setAggregations(newAggregations)
            .setPreGroupedSymbols(ImmutableList.of())
            .build());
  }
Implements distinct aggregations with different inputs by transforming plans of the following
shape:

<pre>
- Aggregation
       GROUP BY (k)
       F1(DISTINCT a0, a1, ...)
       F2(DISTINCT b0, b1, ...)
       F3(c0, c1, ...)
    - X
</pre>

into

<pre>
- Aggregation
       GROUP BY (k)
       F1(a0, a1, ...) mask ($0)
       F2(b0, b1, ...) mask ($1)
       F3(c0, c1, ...)
    - MarkDistinct (k, a0, a1, ...) -> $0
         - MarkDistinct (k, b0, b1, ...) -> $1
             - X
</pre>
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/MultipleDistinctAggregationToMarkDistinct.java