public static RewritablePKFKJoinInfo isRewritablePKFKJoin(Join join,
        final RelNode fkInput, final RelNode nonFkInput,
        RelMetadataQuery mq) {
    final JoinRelType joinType = join.getJoinType();
    final RexNode cond = join.getCondition();

    Preconditions.checkArgument(fkInput == join.getLeft()
        || fkInput == join.getRight(), "Invalid input: " + fkInput);
    Preconditions.checkArgument(nonFkInput == join.getLeft()
        || nonFkInput == join.getRight(), "Invalid input: " + nonFkInput);

    final RewritablePKFKJoinInfo nonRewritable = RewritablePKFKJoinInfo.of(false, null);

    // TODO : Need to handle Anti join.
    // https://issues.apache.org/jira/browse/HIVE-23906
    if (joinType == JoinRelType.ANTI) {
      return nonRewritable;
    }

    if (joinType != JoinRelType.INNER && !join.isSemiJoin()) {
      // If it is not an inner, we transform it as the metadata
      // providers for expressions do not pull information through
      // outer join (as it would not be correct)
      join = join.copy(join.getTraitSet(), cond,
          join.getLeft(), join.getRight(), JoinRelType.INNER,
          false);
    }

    // 1) Check whether there is any filtering condition on the
    // non-FK side. Basically we check whether the operators
    // below altered the PK cardinality in any way
    if (HiveRelOptUtil.isRowFilteringPlan(mq, nonFkInput)) {
      return nonRewritable;
    }

    // 2) Check whether there is an FK relationship
    final Map<RexTableInputRef, RexNode> refToRex = new HashMap<>();
    final EquivalenceClasses ec = new EquivalenceClasses();
    for (RexNode conj : RelOptUtil.conjunctions(cond)) {
      if (!conj.isA(SqlKind.EQUALS)) {
        // Not an equality, we bail out
        return nonRewritable;
      }
      RexCall equiCond = (RexCall) conj;
      RexNode eqOp1 = equiCond.getOperands().get(0);
      Set<RexNode> eqOp1ExprsLineage = mq.getExpressionLineage(join, eqOp1);
      if (eqOp1ExprsLineage == null) {
        // Cannot be mapped, bail out
        return nonRewritable;
      }
      RexNode eqOp2 = equiCond.getOperands().get(1);
      Set<RexNode> eqOp2ExprsLineage = mq.getExpressionLineage(join, eqOp2);
      if (eqOp2ExprsLineage == null) {
        // Cannot be mapped, bail out
        return nonRewritable;
      }
      List<RexTableInputRef> eqOp2ExprsFiltered = null;
      for (RexNode eqOpExprLineage1 : eqOp1ExprsLineage) {
        RexTableInputRef inputRef1 = extractTableInputRef(eqOpExprLineage1);
        if (inputRef1 == null) {
          // Bail out as this condition could not be map into an input reference
          return nonRewritable;
        }
        refToRex.put(inputRef1, eqOp1);
        if (eqOp2ExprsFiltered == null) {
          // First iteration
          eqOp2ExprsFiltered = new ArrayList<>();
          for (RexNode eqOpExprLineage2 : eqOp2ExprsLineage) {
            RexTableInputRef inputRef2 = extractTableInputRef(eqOpExprLineage2);
            if (inputRef2 == null) {
              // Bail out as this condition could not be map into an input reference
              return nonRewritable;
            }
            // Add to list of expressions for follow-up iterations
            eqOp2ExprsFiltered.add(inputRef2);
            // Add to equivalence classes and backwards mapping
            ec.addEquivalence(inputRef1, inputRef2);
            refToRex.put(inputRef2, eqOp2);
          }
        } else {
          // Rest of iterations, only adding, no checking
          for (RexTableInputRef inputRef2 : eqOp2ExprsFiltered) {
            ec.addEquivalence(inputRef1, inputRef2);
          }
        }
      }
    }
    if (ec.getEquivalenceClassesMap().isEmpty()) {
      // This may be a cartesian product, we bail out
      return nonRewritable;
    }

    // 3) Gather all tables from the FK side and the table from the
    // non-FK side
    final Set<RelTableRef> leftTables = mq.getTableReferences(join.getLeft());
    final Set<RelTableRef> rightTables =
        Sets.difference(mq.getTableReferences(join), mq.getTableReferences(join.getLeft()));
    final Set<RelTableRef> fkTables = join.getLeft() == fkInput ? leftTables : rightTables;
    final Set<RelTableRef> nonFkTables = join.getLeft() == fkInput ? rightTables : leftTables;
    assert nonFkTables.size() == 1;
    final RelTableRef nonFkTable = nonFkTables.iterator().next();
    final List<String> nonFkTableQName = nonFkTable.getQualifiedName();

    // 4) For each table, check whether there is a matching on the non-FK side.
    // If there is and it is the only condition, we are ready to transform
    boolean canBeRewritten = false;
    List<RexNode> nullableNodes = null;
    for (RelTableRef tRef : fkTables) {
      List<RelReferentialConstraint> constraints = tRef.getTable().getReferentialConstraints();
      for (RelReferentialConstraint constraint : constraints) {
        if (constraint.getTargetQualifiedName().equals(nonFkTableQName)) {
          nullableNodes = new ArrayList<>();
          EquivalenceClasses ecT = EquivalenceClasses.copy(ec);
          boolean allContained = true;
          for (int pos = 0; pos < constraint.getNumColumns(); pos++) {
            int foreignKeyPos = constraint.getColumnPairs().get(pos).source;
            RelDataType foreignKeyColumnType =
                tRef.getTable().getRowType().getFieldList().get(foreignKeyPos).getType();
            RexTableInputRef foreignKeyColumnRef =
                RexTableInputRef.of(tRef, foreignKeyPos, foreignKeyColumnType);
            int uniqueKeyPos = constraint.getColumnPairs().get(pos).target;
            RexTableInputRef uniqueKeyColumnRef = RexTableInputRef.of(nonFkTable, uniqueKeyPos,
                nonFkTable.getTable().getRowType().getFieldList().get(uniqueKeyPos).getType());
            if (ecT.getEquivalenceClassesMap().containsKey(uniqueKeyColumnRef) &&
                ecT.getEquivalenceClassesMap().get(uniqueKeyColumnRef).contains(foreignKeyColumnRef)) {
              if (foreignKeyColumnType.isNullable()) {
                //TODO : Handle Anti Join. https://issues.apache.org/jira/browse/HIVE-23906
                if (joinType == JoinRelType.INNER || join.isSemiJoin()) {
                  // If it is nullable and it is an INNER, we just need a IS NOT NULL filter
                  RexNode originalCondOp = refToRex.get(foreignKeyColumnRef);
                  assert originalCondOp != null;
                  nullableNodes.add(originalCondOp);
                } else {
                  // If it is nullable and this is not an INNER, we cannot execute any transformation
                  allContained = false;
                  break;
                }
              }
              // Remove this condition from eq classes as we have checked that it is present
              // in the join condition
              ecT.removeEquivalence(uniqueKeyColumnRef, foreignKeyColumnRef);
            } else {
              // No relationship, we cannot do anything
              allContained = false;
              break;
            }
          }
          if (allContained && ecT.getEquivalenceClassesMap().isEmpty()) {
            // We made it
            canBeRewritten = true;
            break;
          }
        }
      }
    }

    return RewritablePKFKJoinInfo.of(canBeRewritten, nullableNodes);
  }
Returns a triple where first value represents whether we could extract a FK-PK join
or not, the second value is a pair with the column from left and right input that
are used for the FK-PK join, and the third value are the predicates that are not
part of the FK-PK condition. Currently we can only extract one FK-PK join.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveRelOptUtil.java