fix comments

Author: Indhumathi27

ql/src/java/org/apache/hadoop/hive/ql/optimizer/LimitPushdownOptimizer.java

@@ -198,6 +198,11 @@ public Object process(Node nd, Stack<Node> stack,
           }
           currentOp = currentOp.getChildOperators().get(0);
         }
+        // Skip RS copy for plans that are already TopNKey-optimized to avoid re-propagating
+        // TopNKey configuration into the same ReduceSink.
+        if (((LimitPushdownContext)procCtx).disallowRSCopy && foundGroupByOperator) {
+          return false;
+        }
         List<ExprNodeDesc> cKeys = cRS.getConf().getKeyCols();
         List<ExprNodeDesc> pKeys = pRS.getConf().getKeyCols();
         if (pRS.getChildren().get(0) instanceof GroupByOperator &&
@@ -250,8 +255,12 @@ private static class LimitPushdownContext implements NodeProcessorCtx {
 
     private final float threshold;
 
+    private final boolean disallowRSCopy;
+
     public LimitPushdownContext(HiveConf conf) throws SemanticException {
       threshold = conf.getFloatVar(HiveConf.ConfVars.HIVE_LIMIT_PUSHDOWN_MEMORY_USAGE);
+      disallowRSCopy = conf.getBoolVar(HiveConf.ConfVars.HIVE_OPTIMIZE_TOPNKEY) &&
+              !conf.getBoolVar(HiveConf.ConfVars.HIVE_MAPSIDE_AGGREGATE);
       if (threshold <= 0 || threshold >= 1) {
         throw new SemanticException("Invalid memory usage value " + threshold +
             " for " + HiveConf.ConfVars.HIVE_LIMIT_PUSHDOWN_MEMORY_USAGE);

ql/src/java/org/apache/hadoop/hive/ql/optimizer/topnkey/TopNKeyProcessor.java

@@ -17,16 +17,10 @@
  */
 package org.apache.hadoop.hive.ql.optimizer.topnkey;
 
-import java.util.ArrayDeque;
 import java.util.Collections;
-import java.util.Deque;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.Stack;
 
-import org.apache.hadoop.hive.ql.exec.GroupByOperator;
-import org.apache.hadoop.hive.ql.exec.JoinOperator;
 import org.apache.hadoop.hive.ql.exec.Operator;
 import org.apache.hadoop.hive.ql.exec.OperatorFactory;
 import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
@@ -77,12 +71,6 @@ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
       return null;
     }
 
-    // Skip the current optimization when a simple global ORDER BY...LIMIT is present
-    // This plan structure is handled more efficiently by the specialized 'TopN In Reducer' optimization.
-    if (!reduceSinkDesc.isPTFReduceSink() && reduceSinkDesc.getTopN() > -1 && isOrderByLimitPath(reduceSinkOperator)) {
-      return null;
-    }
-
     if (reduceSinkDesc.getTopN() > maxTopNAllowed) {
       return null;
     }
@@ -123,50 +111,6 @@ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
     return null;
   }
 
-  /**
-   * Returns true if the ReduceSink is only under an ORDER BY + LIMIT plan
-   * and has no GroupBy or Join operators in its upstream ancestry.
-   * This is used to disable TopNKey for pure ORDER BY LIMIT queries where
-   * LIMIT pushdown must take precedence.
-   */
-  public static boolean isOrderByLimitPath(ReduceSinkOperator rs) {
-    if (rs == null) {
-      return false;
-    }
-
-    Deque<Operator<?>> stack = new ArrayDeque<>();
-    Set<Operator<?>> visited = new HashSet<>();
-    stack.push(rs);
-
-    while (!stack.isEmpty()) {
-      Operator<?> current = stack.pop();
-
-      if (!visited.add(current)) {
-        continue;
-      }
-
-      List<Operator<?>> parents = current.getParentOperators();
-      if (parents == null || parents.isEmpty()) {
-        continue;
-      }
-
-      for (Operator<?> parent : parents) {
-
-        // If any GroupBy or Join or topNkey operator exists in the upstream path,
-        // this is NOT a pure ORDER BY LIMIT plan.
-        if (parent instanceof GroupByOperator || parent instanceof JoinOperator || parent instanceof TopNKeyOperator) {
-          return false;
-        }
-
-        stack.push(parent);
-      }
-    }
-
-    // Reached root without seeing GBY or JOIN or topNkey=> pure ORDER BY + LIMIT
-    return true;
-  }
-
-
   public static Operator<? extends OperatorDesc> copyDown(Operator<? extends OperatorDesc> child, OperatorDesc operatorDesc) {
     final List<Operator<? extends OperatorDesc>> parents = child.getParentOperators();
 

ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java

@@ -1300,9 +1300,17 @@ private static void runTopNKeyOptimization(OptimizeTezProcContext procCtx)
       return;
     }
 
+    // Restrict TopNKey matching to GROUP BY / JOIN ReduceSinks, except for PTF queries.
+    String reduceSinkOp = ReduceSinkOperator.getOperatorName() + "%";
+    String topNKeyRegexPattern = hasPTFReduceSink(procCtx) ? reduceSinkOp :
+            ".*(" + GroupByOperator.getOperatorName() + "|" +
+                    JoinOperator.getOperatorName() + "|" +
+                    MapJoinOperator.getOperatorName() + "|" +
+                    CommonMergeJoinOperator.getOperatorName() + "|" +
+            ")%.*" + reduceSinkOp;
     Map<SemanticRule, SemanticNodeProcessor> opRules = new LinkedHashMap<SemanticRule, SemanticNodeProcessor>();
     opRules.put(
-        new RuleRegExp("Top n key optimization", ReduceSinkOperator.getOperatorName() + "%"),
+        new RuleRegExp("Top n key optimization", topNKeyRegexPattern),
         new TopNKeyProcessor(
           HiveConf.getIntVar(procCtx.conf, HiveConf.ConfVars.HIVE_MAX_TOPN_ALLOWED),
           HiveConf.getFloatVar(procCtx.conf, ConfVars.HIVE_TOPN_EFFICIENCY_THRESHOLD),
@@ -1322,6 +1330,15 @@ private static void runTopNKeyOptimization(OptimizeTezProcContext procCtx)
     ogw.startWalking(topNodes, null);
   }
 
+  private static boolean hasPTFReduceSink(OptimizeTezProcContext ctx) {
+    for (ReduceSinkOperator rs : ctx.visitedReduceSinks) {
+      if (rs.getConf().isPTFReduceSink()) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   private boolean findParallelSemiJoinBranch(Operator<?> mapjoin, TableScanOperator bigTableTS,
                                              ParseContext parseContext,
                                              Map<ReduceSinkOperator, TableScanOperator> semijoins,