@Override
  public void onMatch(RelOptRuleCall call) {
    final Project project = call.rel(0);
    final RexBuilder rexBuilder = project.getCluster().getRexBuilder();
    List<RexNode> topProjExprs = project.getProjects();
    Join join = call.rel(1);
    final JoinRelType joinType = join.getJoinType();
    final RelNode leftInput = join.getLeft();
    final RelNode rightInput = join.getRight();
    final RexNode cond = join.getCondition();

    //TODO:https://issues.apache.org/jira/browse/HIVE-23920
    if (joinType == JoinRelType.ANTI) {
      return;
    }

    // 1) If it is an inner, check whether project only uses columns from one side.
    // That side will need to be the FK side.
    // If it is a left outer, left will be the FK side.
    // If it is a right outer, right will be the FK side.
    final RelNode fkInput;
    final RelNode nonFkInput;
    final ImmutableBitSet topRefs =
        RelOptUtil.InputFinder.bits(topProjExprs, null);
    final ImmutableBitSet leftBits =
        ImmutableBitSet.range(leftInput.getRowType().getFieldCount());
    final ImmutableBitSet rightBits =
        ImmutableBitSet.range(leftInput.getRowType().getFieldCount(),
            join.getRowType().getFieldCount());
    // These boolean values represent corresponding left, right input which is potential FK
    boolean leftInputPotentialFK = topRefs.intersects(leftBits);
    boolean rightInputPotentialFK = topRefs.intersects(rightBits);
    if (leftInputPotentialFK && rightInputPotentialFK &&
        (joinType == JoinRelType.INNER || joinType == JoinRelType.SEMI)) {
      // Both inputs are referenced. Before making a decision, try to swap
      // references in join condition if it is an inner join, i.e. if a join
      // condition column is referenced above the join, then we can just
      // reference the column from the other side.
      // For example, given two relations R(a1,a2), S(b1) :
      // SELECT a2, b1 FROM R, S ON R.a1=R.b1 =>
      // SELECT a2, a1 FROM R, S ON R.a1=R.b1
      int joinFieldCount = join.getRowType().getFieldCount();
      Mapping mappingLR = Mappings.create(MappingType.PARTIAL_FUNCTION, joinFieldCount, joinFieldCount);
      Mapping mappingRL = Mappings.create(MappingType.PARTIAL_FUNCTION, joinFieldCount, joinFieldCount);
      for (RexNode conj : RelOptUtil.conjunctions(cond)) {
        if (!conj.isA(SqlKind.EQUALS)) {
          continue;
        }
        RexCall eq = (RexCall) conj;
        RexNode op1 = eq.getOperands().get(0);
        RexNode op2 = eq.getOperands().get(1);
        if (op1 instanceof RexInputRef && op2 instanceof RexInputRef) {
          // Check references
          int ref1 = ((RexInputRef) op1).getIndex();
          int ref2 = ((RexInputRef) op2).getIndex();
          int leftRef = -1;
          int rightRef = -1;
          if (leftBits.get(ref1) && rightBits.get(ref2)) {
            leftRef = ref1;
            rightRef = ref2;
          } else if (rightBits.get(ref1) && leftBits.get(ref2)) {
            leftRef = ref2;
            rightRef = ref1;
          }
          if (leftRef != -1 && rightRef != -1) {
            // We do not add more than one mapping per source
            // as it is useless
            if (mappingLR.getTargetOpt(leftRef) == -1) {
              mappingLR.set(leftRef, rightRef);
            }
            if (mappingRL.getTargetOpt(rightRef) == -1) {
              mappingRL.set(rightRef, leftRef);
            }
          }
        }
      }
      if (mappingLR.size() != 0) {
        // First insert missing elements into the mapping as identity mappings
        for (int i = 0; i < joinFieldCount; i++) {
          if (mappingLR.getTargetOpt(i) == -1) {
            mappingLR.set(i, i);
          }
          if (mappingRL.getTargetOpt(i) == -1) {
            mappingRL.set(i, i);
          }
        }
        // Then, we start by trying to reference only left side in top projections
        List<RexNode> swappedTopProjExprs = topProjExprs.stream()
            .map(projExpr -> projExpr.accept(new RexPermuteInputsShuttle(mappingRL, call.rel(1))))
            .collect(Collectors.toList());
        rightInputPotentialFK = RelOptUtil.InputFinder.bits(swappedTopProjExprs, null).intersects(rightBits);
        if (!rightInputPotentialFK) {
          topProjExprs = swappedTopProjExprs;
        } else {
          // If it did not work, we try to reference only right side in top projections
          swappedTopProjExprs = topProjExprs.stream()
              .map(projExpr -> projExpr.accept(new RexPermuteInputsShuttle(mappingLR, call.rel(1))))
              .collect(Collectors.toList());
          leftInputPotentialFK = RelOptUtil.InputFinder.bits(swappedTopProjExprs, null).intersects(leftBits);
          if (!leftInputPotentialFK) {
            topProjExprs = swappedTopProjExprs;
          }
        }
      }
    } else if (!leftInputPotentialFK && !rightInputPotentialFK) {
      // TODO: There are no references in the project operator above.
      // In this case, we should probably do two passes, one for
      // left as FK and one for right as FK, although it may be expensive.
      // Currently we only assume left as FK
      leftInputPotentialFK = true;
    }

    final Mode mode;
    switch (joinType) {
    case SEMI:
    case INNER:
      //case ANTI: //TODO:https://issues.apache.org/jira/browse/HIVE-23920
      if (leftInputPotentialFK && rightInputPotentialFK) {
        // Bails out as it references columns from both sides (or no columns)
        // and there is nothing to transform
        return;
      }
      fkInput = leftInputPotentialFK ? leftInput : rightInput;
      nonFkInput = leftInputPotentialFK ? rightInput: leftInput;
      mode = Mode.REMOVE;
      break;
    case LEFT:
      fkInput = leftInput;
      nonFkInput = rightInput;
      mode = leftInputPotentialFK && !rightInputPotentialFK ? Mode.REMOVE : Mode.TRANSFORM;
      break;
    case RIGHT:
      fkInput = rightInput;
      nonFkInput = leftInput;
      mode = !leftInputPotentialFK && rightInputPotentialFK ? Mode.REMOVE : Mode.TRANSFORM;
      break;
    default:
      // Other type, bail out
      return;
    }

    // 2) Check whether this join can be rewritten or removed
    RewritablePKFKJoinInfo r = HiveRelOptUtil.isRewritablePKFKJoin(
        join, fkInput,  nonFkInput, call.getMetadataQuery());

    // 3) If it is the only condition, we can trigger the rewriting
    if (r.rewritable) {
      rewrite(mode, fkInput, nonFkInput, join, topProjExprs, call, project, r.nullableNodes);
    } else {
      // check if FK side could be removed instead
      // Possibly this could be enhanced to take other join type into consideration.
      if (joinType != JoinRelType.INNER) {
        return;
      }

      //first swap fk and non-fk input and see if we can rewrite them
      RewritablePKFKJoinInfo fkRemoval = HiveRelOptUtil.isRewritablePKFKJoin(
          join, nonFkInput, fkInput, call.getMetadataQuery());

      if (fkRemoval.rewritable) {
        // we have established that nonFkInput is FK, and fkInput is PK
        // and there is no row filtering on FK side
        // check that FK side join column is distinct (i.e. have a group by)
        ImmutableBitSet fkSideBitSet;
        if (nonFkInput == leftInput) {
          fkSideBitSet = leftBits;
        } else {
          fkSideBitSet = rightBits;
        }

        ImmutableBitSet.Builder fkJoinColBuilder = ImmutableBitSet.builder();
        for (RexNode conj : RelOptUtil.conjunctions(cond)) {
          if (!conj.isA(SqlKind.EQUALS)) {
            return;
          }
          RexCall eq = (RexCall) conj;
          RexNode op1 = eq.getOperands().get(0);
          RexNode op2 = eq.getOperands().get(1);
          if (op1 instanceof RexInputRef && op2 instanceof RexInputRef) {
            // Check references
            int ref1 = ((RexInputRef) op1).getIndex();
            int ref2 = ((RexInputRef) op2).getIndex();
            int leftRef = -1;
            int rightRef = -1;
            if (fkSideBitSet.get(ref1)) {
              // check that join columns are not nullable
              if (op1.getType().isNullable()) {
                return;
              }
              fkJoinColBuilder.set(fkSideBitSet.indexOf(ref1));
            } else {
              if (op2.getType().isNullable()) {
                return;
              }
              fkJoinColBuilder.set(fkSideBitSet.indexOf(ref2));
            }
          }
        }

        final Boolean isUnique = call.getMetadataQuery().areColumnsUnique(nonFkInput, fkJoinColBuilder.build());
        if (isUnique == null || !isUnique) {
          return;
        }

        // all conditions are met, therefore we can perform rewrite to remove fk side
        rewrite(mode, fkInput, nonFkInput, join, topProjExprs, call, project, fkRemoval.nullableNodes);

      }

    }
  }
The rule can perform two different optimizations.
1) Removes a join if is does not alter the cardinality of the one of its inputs.
In particular, this rule is triggered if:
- it is a join on PK-FK/UK-FK,
- project on top only references columns from the FK side, and
- PK/UK side is not filtered
It optionally adds an IS NOT NULL filter if any FK column can be nullable
2) Transforms a left/right outer join into an inner join if:
- it is a join on PK-FK/UK-FK,
- FK is not nullable
- PK/UK side is not filtered
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinConstraintsRule.java