@Override
  public RelNode trim(RelBuilder relBuilder, RelNode root) {
    try {
      if (root.getInputs().size() != 1) {
        LOG.debug("Only plans where root has one input are supported. Root: {}", root);
        return root;
      }

      REL_BUILDER.set(relBuilder);

      RexBuilder rexBuilder = relBuilder.getRexBuilder();
      RelNode rootInput = root.getInput(0);

      // Build the list of RexInputRef from root input RowType
      List<RexInputRef> rootFieldList = new ArrayList<>(rootInput.getRowType().getFieldCount());
      List<String> newColumnNames = new ArrayList<>();
      for (int i = 0; i < rootInput.getRowType().getFieldList().size(); ++i) {
        RelDataTypeField relDataTypeField = rootInput.getRowType().getFieldList().get(i);
        rootFieldList.add(rexBuilder.makeInputRef(relDataTypeField.getType(), i));
        newColumnNames.add(relDataTypeField.getName());
      }

      // Bit set to gather the refs that backtrack to constant values
      BitSet constants = new BitSet();
      List<JoinedBackFields> lineages = getExpressionLineageOf(rootFieldList, rootInput, constants);
      if (lineages == null) {
        LOG.debug("Some projected field lineage can not be determined");
        return root;
      }

      // 1. Collect candidate tables for join back and map RexNodes coming from those tables to their index in the
      // rootInput row type
      // Collect all used fields from original plan
      ImmutableBitSet fieldsUsed = ImmutableBitSet.of(constants.stream().toArray());
      List<TableToJoinBack> tableToJoinBackList = new ArrayList<>(lineages.size());
      Map<Integer, RexNode> rexNodesToShuttle = new HashMap<>(rootInput.getRowType().getFieldCount());

      for (JoinedBackFields joinedBackFields : lineages) {
        Optional<ImmutableBitSet> projectedKeys = joinedBackFields.relOptHiveTable.getNonNullableKeys().stream()
            .filter(joinedBackFields.fieldsInSourceTable::contains)
            .findFirst();

        if (projectedKeys.isPresent() && !projectedKeys.get().equals(joinedBackFields.fieldsInSourceTable)) {
          TableToJoinBack tableToJoinBack = new TableToJoinBack(projectedKeys.get(), joinedBackFields);
          tableToJoinBackList.add(tableToJoinBack);
          fieldsUsed = fieldsUsed.union(joinedBackFields.getSource(projectedKeys.get()));

          for (TableInputRefHolder mapping : joinedBackFields.mapping) {
            if (!fieldsUsed.get(mapping.indexInOriginalRowType)) {
              rexNodesToShuttle.put(mapping.indexInOriginalRowType, mapping.rexNode);
            }
          }
        } else {
          fieldsUsed = fieldsUsed.union(joinedBackFields.fieldsInOriginalRowType);
        }
      }

      if (tableToJoinBackList.isEmpty()) {
        LOG.debug("None of the tables has keys projected, unable to join back");
        return root;
      }

      // 2. Trim out non-key fields of joined back tables
      Set<RelDataTypeField> extraFields = Collections.emptySet();
      TrimResult trimResult = dispatchTrimFields(rootInput, fieldsUsed, extraFields);
      RelNode newInput = trimResult.left;
      if (newInput.getRowType().equals(rootInput.getRowType())) {
        LOG.debug("Nothing was trimmed out.");
        return root;
      }

      // 3. Join back tables to the top of original plan
      Mapping newInputMapping = trimResult.right;
      Map<RexTableInputRef, Integer> tableInputRefMapping = new HashMap<>();

      for (TableToJoinBack tableToJoinBack : tableToJoinBackList) {
        LOG.debug("Joining back table {}", tableToJoinBack.joinedBackFields.relOptHiveTable.getName());

        // 3.1. Create new TableScan of tables to join back
        RelOptHiveTable relOptTable = tableToJoinBack.joinedBackFields.relOptHiveTable;
        RelOptCluster cluster = relBuilder.getCluster();
        HiveTableScan tableScan = new HiveTableScan(cluster, cluster.traitSetOf(HiveRelNode.CONVENTION),
            relOptTable, relOptTable.getHiveTableMD().getTableName(), null, false, false);
        // 3.2. Create Project with the required fields from this table
        RelNode projectTableAccessRel = tableScan.project(
            tableToJoinBack.joinedBackFields.fieldsInSourceTable, new HashSet<>(0), REL_BUILDER.get());

        // 3.3. Create mapping between the Project and TableScan
        Mapping projectMapping = Mappings.create(MappingType.INVERSE_SURJECTION,
            tableScan.getRowType().getFieldCount(),
            tableToJoinBack.joinedBackFields.fieldsInSourceTable.cardinality());
        int projectIndex = 0;
        for (int i : tableToJoinBack.joinedBackFields.fieldsInSourceTable) {
          projectMapping.set(i, projectIndex);
          ++projectIndex;
        }

        int offset = newInput.getRowType().getFieldCount();

        // 3.4. Map rexTableInputRef to the index where it can be found in the new Input row type
        for (TableInputRefHolder mapping : tableToJoinBack.joinedBackFields.mapping) {
          int indexInSourceTable = mapping.tableInputRef.getIndex();
          if (!tableToJoinBack.keys.get(indexInSourceTable)) {
            // 3.5. if this is not a key field it is shifted by the left input field count
            tableInputRefMapping.put(mapping.tableInputRef, offset + projectMapping.getTarget(indexInSourceTable));
          }
        }

        // 3.7. Create Join
        relBuilder.push(newInput);
        relBuilder.push(projectTableAccessRel);

        RexNode joinCondition = joinCondition(
            newInput, newInputMapping, tableToJoinBack, projectTableAccessRel, projectMapping, rexBuilder);

        newInput = relBuilder.join(JoinRelType.INNER, joinCondition).build();
      }

      // 4. Collect rexNodes for Project
      TableInputRefMapper mapper = new TableInputRefMapper(tableInputRefMapping, rexBuilder, newInput);
      List<RexNode> rexNodeList = new ArrayList<>(rootInput.getRowType().getFieldCount());
      for (int i = 0; i < rootInput.getRowType().getFieldCount(); i++) {
        RexNode rexNode = rexNodesToShuttle.get(i);
        if (rexNode != null) {
          rexNodeList.add(mapper.apply(rexNode));
        } else {
          int target = newInputMapping.getTarget(i);
          rexNodeList.add(
              rexBuilder.makeInputRef(newInput.getRowType().getFieldList().get(target).getType(), target));
        }
      }

      // 5. Create Project on top of all Join backs
      relBuilder.push(newInput);
      relBuilder.project(rexNodeList, newColumnNames);

      return root.copy(root.getTraitSet(), singletonList(relBuilder.build()));
    } finally {
      REL_BUILDER.remove();
    }
  }
Optimization to reduce the amount of broadcasted/shuffled data throughout the DAG processing.
This optimization targets queries with one or more tables joined multiple times on their keys
and several columns are projected from those tables.

Example:
with sq as (
select c_customer_id customer_id
      ,c_first_name customer_first_name
      ,c_last_name customer_last_name
  from customer
)
select c1.customer_id
      ,c1.customer_first_name
      ,c1.customer_last_name
 from sq c1
     ,sq c2
      ...
     ,sq cn
where c1.customer_id = c2.customer_id
  and ...

In this case all column data in the cte will be shuffled.

Goal of this optimization: rewrite the plan to include only primary key or non null unique key columns of
affected tables and join the them back to the result set of the main query to fetch the rest of the wide columns.
This reduces the data size of the affected tables that is broadcast/shuffled throughout the DAG processing.

  HiveProject(customer_id=[$0], c_first_name=[$2], c_last_name=[$3])
    HiveJoin(condition=[=($0, $1)], joinType=[inner], algorithm=[none], cost=[not available])
      (original plan)
      HiveProject(customer_id=[$2])
        HiveJoin(...)
          ...
      (joined back customer table)
      HiveProject(c_customer_id=[$1], c_first_name=[$8], c_last_name=[$9])
        HiveTableScan(table=[[default, customer]], table:alias=[customer])
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveCardinalityPreservingJoinOptimization.java