private void bucketingVersionSanityCheck(OptimizeTezProcContext procCtx) throws SemanticException {
    // Fetch all the FileSinkOperators.
    Set<FileSinkOperator> fsOpsAll = new HashSet<>();
    for (TableScanOperator ts : procCtx.parseContext.getTopOps().values()) {
      Set<FileSinkOperator> fsOps = OperatorUtils.findOperators(
          ts, FileSinkOperator.class);
      fsOpsAll.addAll(fsOps);
    }

    Map<Operator<?>, Integer> processedOperators = new IdentityHashMap<>();
    for (FileSinkOperator fsOp : fsOpsAll) {
      // Look for direct parent ReduceSinkOp
      // If there are more than 1 parent, bail out.
      Operator<?> parent = fsOp;
      List<Operator<?>> parentOps = parent.getParentOperators();
      while (parentOps != null && parentOps.size() == 1) {
        parent = parentOps.get(0);
        if (!(parent instanceof ReduceSinkOperator)) {
          parentOps = parent.getParentOperators();
          continue;
        }

        // Found the target RSOp 0
        int bucketingVersion = fsOp.getConf().getTableInfo().getBucketingVersion();
        if (fsOp.getConf().getTableInfo().getBucketingVersion() == -1) {
          break;
        }
        if (fsOp.getConf().getTableInfo().getBucketingVersion() != fsOp.getConf().getBucketingVersion()) {
          throw new RuntimeException("FsOp bucketingVersions is inconsistent with its tableinfo");
        }
        if (processedOperators.containsKey(parent) && processedOperators.get(parent) != bucketingVersion) {
          throw new SemanticException(String.format(
              "Operator (%s) is already processed and is using bucketingVersion(%d); so it can't be changed to %d ",
              parent, processedOperators.get(parent), bucketingVersion));
        }
        processedOperators.put(parent, bucketingVersion);

        break;
      }
    }
  }
Internal class to encapsulate information needed to evaluate stats
about a SJ that will be kept in the tree.
ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java