private boolean checkConvertJoinSMBJoin(JoinOperator joinOp, OptimizeTezProcContext context,
      int bigTablePosition, TezBucketJoinProcCtx tezBucketJoinProcCtx) throws SemanticException {

    ReduceSinkOperator bigTableRS =
        (ReduceSinkOperator) joinOp.getParentOperators().get(bigTablePosition);
    int numBuckets = bigTableRS.getParentOperators().get(0).getOpTraits().getNumBuckets();

    int size = -1;
    boolean shouldCheckExternalTables =
        context.conf.getBoolVar(HiveConf.ConfVars.HIVE_DISABLE_UNSAFE_EXTERNALTABLE_OPERATIONS);
    StringBuilder sb = new StringBuilder();
    for (Operator<?> parentOp : joinOp.getParentOperators()) {
      if (shouldCheckExternalTables && !canTableUseStats(parentOp, sb)) {
        LOG.debug("External table {} found in join and also could not provide statistics - disabling SMB join.", sb);
        return false;
      }
      // GBY operators buffers one record. These are processed when ReduceRecordSources flushes the operator tree
      // when end of record stream reached. If the tree has more than two GBY operators CommonMergeJoinOperator can
      // not process all buffered records.
      // HIVE-27788
      if (parentOp.getParentOperators() != null) {
        // Parent operator is RS and hasMoreOperatorsThan traverses until the next RS, so we start from grandparent
        for (Operator<?> grandParent : parentOp.getParentOperators()) {
          if (hasMoreOperatorsThan(grandParent, GroupByOperator.class, 1)) {
            LOG.info("We cannot convert to SMB join " +
                "because one of the join branches has more than one Group by operators in the same reducer.");
            return false;
          }
        }
      }

      // each side better have 0 or more RS. if either side is unbalanced, cannot convert.
      // This is a workaround for now. Right fix would be to refactor code in the
      // MapRecordProcessor and ReduceRecordProcessor with respect to the sources.
      Set<ReduceSinkOperator> set =
          OperatorUtils.findOperatorsUpstream(parentOp.getParentOperators(),
              ReduceSinkOperator.class);
      if (size < 0) {
        size = set.size();
        continue;
      }

      if (((size > 0) && (set.size() > 0)) || ((size == 0) && (set.size() == 0))) {
        continue;
      } else {
        return false;
      }
    }

    // the sort and bucket cols have to match on both sides for this
    // transformation of the join operation
    for (Operator<? extends OperatorDesc> parentOp : joinOp.getParentOperators()) {
      if (!(parentOp instanceof ReduceSinkOperator)) {
        // could be mux/demux operators. Currently not supported
        LOG.debug("Found correlation optimizer operators. Cannot convert to SMB at this time.");
        return false;
      }
      ReduceSinkOperator rsOp = (ReduceSinkOperator) parentOp;
      if (rsOp.getOpTraits().hasCustomBucketFunction()) {
        LOG.info("We don't support SMB with custom bucket functions yet");
        return false;
      }

      List<ExprNodeDesc> keyCols = rsOp.getConf().getKeyCols();

      // For SMB, the key column(s) in RS should be same as bucket column(s) and sort column(s)`
      List<String> sortCols = rsOp.getOpTraits().getSortCols().get(0);
      List<String> bucketCols = rsOp.getOpTraits().getBucketColNames().get(0);
      if (sortCols.size() != keyCols.size() || bucketCols.size() != keyCols.size()) {
        return false;
      }

      // Check columns.
      for (int i = 0; i < sortCols.size(); i++) {
        ExprNodeDesc sortCol = rsOp.getColumnExprMap().get(sortCols.get(i));
        ExprNodeDesc bucketCol = rsOp.getColumnExprMap().get(bucketCols.get(i));
        if (!(sortCol.isSame(keyCols.get(i)) && bucketCol.isSame(keyCols.get(i)))) {
          return false;
        }
      }
      // check Parent's traits are same as rs
      OpTraits parentTraits = rsOp.getParentOperators().get(0).getOpTraits();
      if (null == parentTraits) {
        // programming error - shouldn't be null
        return false;
      }

      if (!checkColEquality(parentTraits.getSortCols(), rsOp
          .getOpTraits().getSortCols(), rsOp.getColumnExprMap(), false)) {
        LOG.info("We cannot convert to SMB because the sort column names do not match.");
        return false;
      }

      if (!checkColEquality(parentTraits.getBucketColNames(), rsOp
          .getOpTraits().getBucketColNames(), rsOp.getColumnExprMap(), true)) {
        LOG.info("We cannot convert to SMB because bucket column names do not match.");
        return false;
      }
    }

    if (numBuckets < 0) {
      numBuckets = bigTableRS.getConf().getNumReducers();
    }
    tezBucketJoinProcCtx.setNumBuckets(numBuckets);

    // With bucketing using two different versions. Version 1 for exiting
    // tables and version 2 for new tables. All the inputs to the SMB must be
    // from same version. This only applies to tables read directly and not
    // intermediate outputs of joins/groupbys
    int bucketingVersion = -1;
    for (Operator<? extends OperatorDesc> parentOp : joinOp.getParentOperators()) {
      // Check if the parent is coming from a table scan, if so, what is the version of it.
      assert parentOp.getParentOperators() != null && parentOp.getParentOperators().size() == 1;
      Operator<?> op = OperatorUtils.findSourceOperatorInSameBranch(parentOp);

      if (op instanceof TableScanOperator) {
        int localVersion = ((TableScanOperator) op).getConf().getTableMetadata().getBucketingVersion();
        if (bucketingVersion == -1) {
          bucketingVersion = localVersion;
        } else if (bucketingVersion != localVersion) {
          // versions dont match, return false.
          LOG.debug("SMB Join can't be performed due to bucketing version mismatch");
          return false;
        }
      }
    }

    /* As SMB replaces last RS op from the joining branches and the JOIN op with MERGEJOIN, we need to ensure
     * the RS before these RS, in both branches, are partitioning using same hash generator. It
     * differs depending on ReducerTraits.UNIFORM i.e. ReduceSinkOperator#computeMurmurHash or
     * ReduceSinkOperator#computeHashCode, leading to different code for same value. Skip SMB join in such cases.
     */
    Boolean prevRsHasUniformTrait = null;
    for (Operator<? extends OperatorDesc> parentOp : joinOp.getParentOperators()) {
      // Assertion of mandatory single parent is already being done in bucket version check earlier
      Operator<?> op = OperatorUtils.findSourceOperatorInSameBranch(parentOp.getParentOperators().get(0));

      if (op instanceof ReduceSinkOperator) {
        boolean hasUniformTrait = ((ReduceSinkOperator) op).getConf()
                .getReducerTraits().contains(ReduceSinkDesc.ReducerTraits.UNIFORM);
        if (prevRsHasUniformTrait == null) {
          prevRsHasUniformTrait = hasUniformTrait;
        } else if (prevRsHasUniformTrait != hasUniformTrait) {
          LOG.debug("SMB Join can't be performed due to partition hash generator mismatch across join branches");
          return false;
        }
      }
    }

    LOG.info("We can convert the join to an SMB join.");
    return true;
  }
Preserves additional information about the operator.

When an operator is replaced by a new one; some of the information of the old have to be retained.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java