@Override
    public RexNode visitTableInputRef(RexTableInputRef ref) {
      int source = tableInputRefMapping.get(ref);
      return rexBuilder.makeInputRef(newInput.getRowType().getFieldList().get(source).getType(), source);
    }
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