public static PKFKJoinInfo extractPKFKJoin(
      Join join, List<RexNode> joinFilters, boolean leftInputPotentialFK, RelMetadataQuery mq) {
    final List<RexNode> residualPreds = new ArrayList<>();
    final JoinRelType joinType = join.getJoinType();
    final RelNode fkInput = leftInputPotentialFK ? join.getLeft() : join.getRight();
    final PKFKJoinInfo cannotExtract =
        PKFKJoinInfo.of(false, null, null);

    if (joinType != JoinRelType.INNER) {
      // 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(), join.getCluster().getRexBuilder().makeLiteral(true),
          join.getLeft(), join.getRight(), JoinRelType.INNER, false);
    }

    // 1) Gather all tables from the FK side and the table from the
    // non-FK side
    final Set<RelTableRef> leftTables = mq.getTableReferences(join.getLeft());
    if (leftTables == null) {
      // Could not infer, bail out
      return cannotExtract;
    }
    final Set<RelTableRef> joinTables = mq.getTableReferences(join);
    if (joinTables == null) {
      // Could not infer, bail out
      return cannotExtract;
    }
    final Set<RelTableRef> rightTables = Sets.difference(joinTables, leftTables);
    final Set<RelTableRef> fkTables = join.getLeft() == fkInput ? leftTables : rightTables;
    final Set<RelTableRef> nonFkTables = join.getLeft() == fkInput ? rightTables : leftTables;

    // 2) Check whether there is a FK relationship
    Set<RexCall> candidatePredicates = new HashSet<>();
    EquivalenceClasses ec = new EquivalenceClasses();
    for (RexNode conj : joinFilters) {
      if (!conj.isA(SqlKind.EQUALS)) {
        // Not an equality, continue
        residualPreds.add(conj);
        continue;
      }
      RexCall equiCond = (RexCall) conj;
      RexNode eqOp1 = equiCond.getOperands().get(0);
      if (!RexUtil.isReferenceOrAccess(eqOp1, true)) {
        // Ignore
        residualPreds.add(conj);
        continue;
      }
      Set<RexNode> eqOp1ExprsLineage = mq.getExpressionLineage(join, eqOp1);
      if (eqOp1ExprsLineage == null) {
        // Cannot be mapped, continue
        residualPreds.add(conj);
        continue;
      }
      RexNode eqOp2 = equiCond.getOperands().get(1);
      if (!RexUtil.isReferenceOrAccess(eqOp2, true)) {
        // Ignore
        residualPreds.add(conj);
        continue;
      }
      Set<RexNode> eqOp2ExprsLineage = mq.getExpressionLineage(join, eqOp2);
      if (eqOp2ExprsLineage == null) {
        // Cannot be mapped, continue
        residualPreds.add(conj);
        continue;
      }
      List<RexTableInputRef> eqOp2ExprsFiltered = null;
      for (RexNode eqOpExprLineage1 : eqOp1ExprsLineage) {
        RexTableInputRef inputRef1 = extractTableInputRef(eqOpExprLineage1);
        if (inputRef1 == null) {
          // This condition could not be map into an input reference
          continue;
        }
        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
              continue;
            }
            // Add to list of expressions for follow-up iterations
            eqOp2ExprsFiltered.add(inputRef2);
            // Add to equivalence classes and backwards mapping
            ec.addEquivalence(inputRef1, inputRef2, equiCond);
            candidatePredicates.add(equiCond);
          }
        } else {
          // Rest of iterations, only adding, no checking
          for (RexTableInputRef inputRef2 : eqOp2ExprsFiltered) {
            ec.addEquivalence(inputRef1, inputRef2, equiCond);
          }
        }
      }
      if (!candidatePredicates.contains(conj)) {
        // We add it to residual already
        residualPreds.add(conj);
      }
    }
    if (ec.getEquivalenceClassesMap().isEmpty()) {
      // This may be a cartesian product, we bail out
      return cannotExtract;
    }

    // 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
    for (final RelTableRef nonFkTable : nonFkTables) {
      final List<String> nonFkTableQName = nonFkTable.getQualifiedName();
      for (RelTableRef tRef : fkTables) {
        List<RelReferentialConstraint> constraints = tRef.getTable().getReferentialConstraints();
        for (RelReferentialConstraint constraint : constraints) {
          if (constraint.getTargetQualifiedName().equals(nonFkTableQName)) {
            EquivalenceClasses ecT = EquivalenceClasses.copy(ec);
            Set<RexNode> removedOriginalPredicates = new HashSet<>();
            ImmutableBitSet.Builder lBitSet = ImmutableBitSet.builder();
            ImmutableBitSet.Builder rBitSet = ImmutableBitSet.builder();
            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)) {
                // Remove this condition from eq classes as we have checked that it is present
                // in the join condition. In turn, populate the columns that are referenced
                // from the join inputs
                for (RexCall originalPred : ecT.removeEquivalence(uniqueKeyColumnRef, foreignKeyColumnRef)) {
                  ImmutableBitSet leftCols = RelOptUtil.InputFinder.bits(originalPred.getOperands().get(0));
                  ImmutableBitSet rightCols = RelOptUtil.InputFinder.bits(originalPred.getOperands().get(1));
                  // Get length and flip column references if join condition specified in
                  // reverse order to join sources
                  int nFieldsLeft = join.getLeft().getRowType().getFieldList().size();
                  int nFieldsRight = join.getRight().getRowType().getFieldList().size();
                  int nSysFields = join.getSystemFieldList().size();
                  ImmutableBitSet rightFieldsBitSet = ImmutableBitSet.range(nSysFields + nFieldsLeft,
                      nSysFields + nFieldsLeft + nFieldsRight);
                  if (rightFieldsBitSet.contains(leftCols)) {
                    ImmutableBitSet t = leftCols;
                    leftCols = rightCols;
                    rightCols = t;
                  }
                  lBitSet.set(leftCols.nextSetBit(0) - nSysFields);
                  rBitSet.set(rightCols.nextSetBit(0) - (nSysFields + nFieldsLeft));
                  removedOriginalPredicates.add(originalPred);
                }
              } else {
                // No relationship, we cannot do anything
                allContained = false;
                break;
              }
            }
            if (allContained) {
              // This is a PK-FK, reassign equivalence classes and remove conditions
              // TODO: Support inference of multiple PK-FK relationships

              // 4.1) Add to residual whatever is remaining
              candidatePredicates.removeAll(removedOriginalPredicates);
              residualPreds.addAll(candidatePredicates);
              // 4.2) Return result
              return PKFKJoinInfo.of(true, Pair.of(lBitSet.build(), rBitSet.build()), residualPreds);
            }
          }
        }
      }
    }

    return cannotExtract;
  }
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