[hive] branch master updated: HIVE-24213: Incorrect exception in the Merge MapJoinTask into its child MapRedTask optimizer (reviewed by Zoltan Haindrich)

[ku...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kuczoram pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git


The following commit(s) were added to refs/heads/master by this push:
     new dd5716f  HIVE-24213: Incorrect exception in the Merge MapJoinTask into its child MapRedTask optimizer (reviewed by Zoltan Haindrich)
dd5716f is described below

commit dd5716f9340716872992808461420aed65f3c92b
Author: zmatyus <47...@users.noreply.github.com>
AuthorDate: Fri Oct 2 11:01:42 2020 +0200

    HIVE-24213: Incorrect exception in the Merge MapJoinTask into its child MapRedTask optimizer (reviewed by Zoltan Haindrich)
---
 .../ql/optimizer/physical/CommonJoinTaskDispatcher.java | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java
index 035a5fd..28e8d12 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java
@@ -58,6 +58,8 @@ import org.apache.hadoop.hive.ql.plan.MapredLocalWork;
 import org.apache.hadoop.hive.ql.plan.MapredWork;
 import org.apache.hadoop.hive.ql.plan.OperatorDesc;
 import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /*
  * Convert tasks involving JOIN into MAPJOIN.
@@ -105,6 +107,8 @@ import org.apache.hadoop.hive.ql.plan.ReduceWork;
  */
 public class CommonJoinTaskDispatcher extends AbstractJoinTaskDispatcher implements SemanticDispatcher {
 
+  protected final Logger LOG = LoggerFactory.getLogger(CommonJoinTaskDispatcher.class);
+
   HashMap<String, Long> aliasToSize = null;
 
   public CommonJoinTaskDispatcher(PhysicalContext context) {
@@ -224,14 +228,19 @@ public class CommonJoinTaskDispatcher extends AbstractJoinTaskDispatcher impleme
           " operator as the work associated with alias " + mapJoinAlias +
           ". Found a " + mapJoinAliasToWork.get(mapJoinAlias).getName() + " operator.");
     }
-    FileSinkOperator mapJoinTaskFileSinkOperator =
-        OperatorUtils.findSingleOperator(
-            mapJoinTaskTableScanOperator, FileSinkOperator.class);
-    if (mapJoinTaskFileSinkOperator == null) {
+
+    Set<FileSinkOperator> mapJoinTaskFileSinkOperators = 
+        OperatorUtils.findOperators(mapJoinTaskTableScanOperator, FileSinkOperator.class);
+    if (mapJoinTaskFileSinkOperators.isEmpty()) {
       throw new SemanticException("Cannot find the " + FileSinkOperator.getOperatorName() +
           " operator at the last operator of the MapJoin Task.");
     }
+    if (mapJoinTaskFileSinkOperators.size() > 1) {
+      LOG.warn("Multiple " + FileSinkOperator.getOperatorName() + " operators found at the last operator of the MapJoin Task.");
+      return;
+    }
     // The mapJoinTaskFileSinkOperator writes to a different directory
+    FileSinkOperator mapJoinTaskFileSinkOperator = mapJoinTaskFileSinkOperators.iterator().next();
     Path childMRPath = mapJoinTaskFileSinkOperator.getConf().getDirName();
     List<String> childMRAliases = childMapWork.getPathToAliases().get(childMRPath);
     if (childMRAliases == null || childMRAliases.size() != 1) {