import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final LogicalAggregate aggregate = (LogicalAggregate) call.rel(0);
  final RelNode input = call.rel(1);

  if (aggregate.containsDistinctCall()) {
    // currently, don't use this rule if any of the aggregates contains DISTINCT
    return;
  }

  final RelTraitSet traits = aggregate.getTraitSet().plus(DrillRel.DRILL_LOGICAL);
  final RelNode convertedInput = convert(input, input.getTraitSet().plus(DrillRel.DRILL_LOGICAL));
  try {
    call.transformTo(new DrillAggregateRel(aggregate.getCluster(), traits, convertedInput, aggregate.indicator,
        aggregate.getGroupSet(), aggregate.getGroupSets(), aggregate.getAggCallList()));
  } catch (InvalidRelException e) {
    tracer.warning(e.toString());
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final UnionRel union = (UnionRel) call.rel(0);
  final List<RelNode> inputs = union.getInputs();
  List<RelNode> convertedInputList = Lists.newArrayList();
  RelTraitSet traits = call.getPlanner().emptyTraitSet().plus(Prel.PHYSICAL);

  try {
    for (int i = 0; i < inputs.size(); i++) {
      RelNode convertedInput = convert(inputs.get(i), PrelUtil.fixTraits(call, traits));
      convertedInputList.add(convertedInput);
    }

    traits = call.getPlanner().emptyTraitSet().plus(Prel.PHYSICAL).plus(DistributionTrait.SINGLETON);
    UnionDistinctPrel unionDistinct =
        new UnionDistinctPrel(union.getCluster(), traits, convertedInputList,
            false /* compatibility already checked during logical phase */);

    call.transformTo(unionDistinct);

  } catch (InvalidRelException e) {
    tracer.warn(e.toString());
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final LogicalUnion union = (LogicalUnion) call.rel(0);

  // This rule applies to Union-All only
  if(!union.all) {
    return;
  }

  final RelTraitSet traits = union.getTraitSet().plus(DrillRel.DRILL_LOGICAL);
  final List<RelNode> convertedInputs = new ArrayList<>();
  for (RelNode input : union.getInputs()) {
    final RelNode convertedInput = convert(input, input.getTraitSet().plus(DrillRel.DRILL_LOGICAL));
    convertedInputs.add(convertedInput);
  }
  try {
    call.transformTo(new DrillUnionRel(union.getCluster(), traits, convertedInputs, union.all,
        true /* check compatibility */));
  } catch (InvalidRelException e) {
    tracer.warning(e.toString()) ;
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final DrillUnionRel union = (DrillUnionRel) call.rel(0);
  final List<RelNode> inputs = union.getInputs();
  List<RelNode> convertedInputList = Lists.newArrayList();
  RelTraitSet traits = call.getPlanner().emptyTraitSet().plus(Prel.DRILL_PHYSICAL);

  try {
    for (int i = 0; i < inputs.size(); i++) {
      RelNode convertedInput = convert(inputs.get(i), PrelUtil.fixTraits(call, traits));
      convertedInputList.add(convertedInput);
    }

    traits = call.getPlanner().emptyTraitSet().plus(Prel.DRILL_PHYSICAL).plus(DrillDistributionTrait.SINGLETON);
    UnionDistinctPrel unionDistinct =
        new UnionDistinctPrel(union.getCluster(), traits, convertedInputList,
            false /* compatibility already checked during logical phase */);

    call.transformTo(unionDistinct);

  } catch (InvalidRelException e) {
    tracer.warning(e.toString());
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final DrillSortRel sort = (DrillSortRel) call.rel(0);
  final RelNode input = sort.getInput();

  // Keep the collation in logical sort. Convert input into a RelNode with 1) this collation, 2) Physical, 3) hash distributed on

  DrillDistributionTrait hashDistribution =
      new DrillDistributionTrait(DrillDistributionTrait.DistributionType.HASH_DISTRIBUTED, ImmutableList.copyOf(getDistributionField(sort)));

  final RelTraitSet traits = sort.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(hashDistribution);

  final RelNode convertedInput = convert(input, traits);

  if(isSingleMode(call)){
    call.transformTo(convertedInput);
  }else{
    RelNode exch = new SingleMergeExchangePrel(sort.getCluster(), sort.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(DrillDistributionTrait.SINGLETON), convertedInput, sort.getCollation());
    call.transformTo(exch);  // transform logical "sort" into "SingleMergeExchange".

  }

}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final LogicalUnion union = (LogicalUnion) call.rel(0);

  // This rule applies to Union-All only
  if(!union.all) {
    return;
  }

  final RelTraitSet traits = union.getTraitSet().plus(Rel.LOGICAL);
  final List<RelNode> convertedInputs = new ArrayList<>();
  for (RelNode input : union.getInputs()) {
    final RelNode convertedInput = convert(input, input.getTraitSet().plus(Rel.LOGICAL).simplify());
    convertedInputs.add(convertedInput);
  }
  try {
    call.transformTo(new UnionRel(union.getCluster(), traits, convertedInputs, union.all,
        true /* check compatibility */));
  } catch (InvalidRelException e) {
    tracer.warn(e.toString()) ;
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
private void createTransformRequest(RelOptRuleCall call, AggregateRel aggregate,
                                    RelNode input, RelTraitSet traits) throws InvalidRelException {

  final RelNode convertedInput = convert(input, traits);

  StreamAggPrel newAgg = new StreamAggPrel(
      aggregate.getCluster(),
      traits,
      convertedInput,
      aggregate.indicator,
      aggregate.getGroupSet(),
      aggregate.getGroupSets(),
      aggregate.getAggCallList(),
      OperatorPhase.PHASE_1of1);

  call.transformTo(newAgg);
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final DrillProjectRel project = (DrillProjectRel) call.rel(0);
  final RelNode input = project.getInput();

  RelTraitSet traits = input.getTraitSet().plus(Prel.DRILL_PHYSICAL);
  RelNode convertedInput = convert(input, traits);

  // Maintain two different map for distribution trait and collation trait.
  // For now, the only difference comes from the way how cast function impacts propagating trait.
  final Map<Integer, Integer> distributionMap = getDistributionMap(project);
  final Map<Integer, Integer> collationMap = getCollationMap(project);

  boolean traitPull = new ProjectTraitPull(call, distributionMap, collationMap).go(project, convertedInput);

  if(!traitPull){
    call.transformTo(new ProjectPrel(project.getCluster(), convertedInput.getTraitSet(), convertedInput, project.getProjects(), project.getRowType()));
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
private static RelTraitSet traits(
    RelOptCluster cluster,
    double rowCount,
    int splitCount,
    RelTraitSet traitSet) {
  PlannerSettings settings = PrelUtil.getPlannerSettings(cluster.getPlanner());
  boolean smallInput = rowCount < (double)settings.getSliceTarget();

  DistributionTrait distribution;
  if(settings.isMultiPhaseAggEnabled() && !settings.isSingleMode() && !smallInput && splitCount > 1) {
    distribution = DistributionTrait.ANY;
  } else {
    distribution = DistributionTrait.SINGLETON;
  }

  return traitSet.plus(distribution);
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
/**
 * Creates a window relational expression.
 *
 * @param cluster Cluster
 * @param traits
 * @param child   Input relational expression
 * @param rowType Output row type
 * @param groups Windows
 */
public DrillWindowRel(
    RelOptCluster cluster,
    RelTraitSet traits,
    RelNode child,
    List<RexLiteral> constants,
    RelDataType rowType,
    List<Group> groups) {
  super(cluster, traits, child, constants, rowType, groups);
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final Sort incomingSort = call.rel(0);
  final RelTraitSet incomingTraits = incomingSort.getTraitSet();
  RelNode input = incomingSort.getInput();

  // if the Optiq sort rel includes a collation and a limit, we need to create a copy the sort rel that excludes the
  // limit information.
  if (!incomingSort.getCollation().getFieldCollations().isEmpty()) {
    input = incomingSort.copy(incomingTraits, input, incomingSort.getCollation(), null, null);
  }

  RelNode convertedInput = convert(input, input.getTraitSet().plus(DrillRel.DRILL_LOGICAL));
  call.transformTo(new DrillLimitRel(incomingSort.getCluster(), convertedInput.getTraitSet().plus(DrillRel.DRILL_LOGICAL), convertedInput, incomingSort.offset, incomingSort.fetch));
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
public ElasticIntermediateScanPrel(
    RelOptCluster cluster,
    RelTraitSet traitSet,
    RelOptTable table,
    TableMetadata dataset,
    List<SchemaPath> projectedColumns,
    double observedRowcountAdjustment) {
  super(cluster, traits(cluster, table.getRowCount(), dataset.getSplitCount(), traitSet), table, dataset.getStoragePluginId(), dataset, projectedColumns, observedRowcountAdjustment);
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
/** Creates a DrillJoinRel.
 * We do not throw InvalidRelException in Logical planning phase. It's up to the post-logical planning check or physical planning
 * to detect the unsupported join type, and throw exception.
 * */
public DrillJoinRel(RelOptCluster cluster, RelTraitSet traits, RelNode left, RelNode right, RexNode condition,
    JoinRelType joinType)  {
  super(cluster, traits, left, right, condition, joinType);
  assert traits.contains(DrillRel.DRILL_LOGICAL);
  RelOptUtil.splitJoinCondition(left, right, condition, leftKeys, rightKeys);
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
/**
 * Creates an <code>AbstractRelNode</code>.
 *
 * @param cluster
 * @param traitSet
 */
public OldScanRelBase(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable relOptTable, RelDataType rowType, GroupScan groupScan, double discountRowCount) {
  super(cluster, traitSet, relOptTable);
  this.rowType = Preconditions.checkNotNull(rowType);
  this.groupScan = Preconditions.checkNotNull(groupScan);
  Preconditions.checkArgument(discountRowCount >= 0 && discountRowCount <= 1, "rowCountDiscount cannot be set to " + discountRowCount);
  this.rowCountDiscount = discountRowCount;
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
protected DrillFilterRelBase(Convention convention, RelOptCluster cluster, RelTraitSet traits, RelNode child, RexNode condition) {
  super(cluster, traits, child, condition);
  assert getConvention() == convention;

  // save the number of conjuncts that make up the filter condition such
  // that repeated calls to the costing function can use the saved copy
  conjunctions = RelOptUtil.conjunctions(condition);
  numConjuncts = conjunctions.size();
  // assert numConjuncts >= 1;

}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
  try {
    return new DrillAggregateRel(getCluster(), traitSet, input, indicator, groupSet, groupSets, aggCalls);
  } catch (InvalidRelException e) {
    throw new AssertionError(e);
  }
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public RelNode convertChild(ProjectRel project, RelNode rel) throws RuntimeException {
  DistributionTrait childDist = rel.getTraitSet().getTrait(DistributionTraitDef.INSTANCE);
  RelCollation childCollation = rel.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE);


  DistributionTrait newDist = convertDist(childDist, distributionMap);
  RelCollation newCollation = convertRelCollation(childCollation, collationMap);
  RelTraitSet newProjectTraits = newTraitSet(Prel.PHYSICAL, newDist, newCollation);
  return new ProjectPrel(project.getCluster(), newProjectTraits, rel, project.getProjects(), project.getRowType());
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
@Override
public void onMatch(RelOptRuleCall call) {
  final Project toTransform = call.rel(0);
  final RelNode input = toTransform.getInput();
  final RelTraitSet traits = toTransform.getTraitSet().plus(Rel.LOGICAL);
  final RelNode convertedInput = convert(input, input.getTraitSet().plus(Rel.LOGICAL).simplify());
  call.transformTo(new ProjectRel(toTransform.getCluster(), traits, convertedInput, toTransform.getProjects(), toTransform.getRowType()));
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
/** Creates a DrillScan. */
public DrillScanRel(final RelOptCluster cluster, final RelTraitSet traits,
    final RelOptTable table) {
  // By default, scan does not support project pushdown.
  // Decision whether push projects into scan will be made solely in DrillPushProjIntoScanRule.
  this(cluster, traits, table, table.getRowType(), GroupScan.ALL_COLUMNS);
  this.settings = PrelUtil.getPlannerSettings(cluster.getPlanner());
}
 

import org.apache.calcite.plan.RelTraitSet; //导入依赖的package包/类
public ElasticScanPrel(
    RelOptCluster cluster,
    RelTraitSet traitSet,
    ElasticsearchPrel input,
    ScanBuilder scanBuilder,
    FunctionLookupContext functionLookupContext) {
  super(cluster, traitSet);
  this.input = input;
  this.rowType = input.getRowType();
  this.scanBuilder = scanBuilder;
}