HIVE-9477: No error thrown when global limit optimization failed to f…

…ind enough number of rows [Spark Branch] (Rui via Xuefu)

git-svn-id: https://svn.apache.org/repos/asf/hive/branches/spark@1655468 13f79535-47bb-0310-9956-ffa450edef68

ql/src/java/org/apache/hadoop/hive/ql/exec/spark/SparkPlanGenerator.java

@@ -233,6 +233,7 @@ private JobConf cloneJobConf(BaseWork work) throws Exception {
       throw new IllegalArgumentException(msg, e);
     }
     if (work instanceof MapWork) {
+      cloned.setBoolean("mapred.task.is.map", true);
       List<Path> inputPaths = Utilities.getInputPaths(cloned, (MapWork) work,
           scratchDir, context, false);
       Utilities.setInputPaths(cloned, inputPaths);
@@ -250,6 +251,7 @@ private JobConf cloneJobConf(BaseWork work) throws Exception {
       // remember the JobConf cloned for each MapWork, so we won't clone for it again
       workToJobConf.put(work, cloned);
     } else if (work instanceof ReduceWork) {
+      cloned.setBoolean("mapred.task.is.map", false);
       Utilities.setReduceWork(cloned, (ReduceWork) work, scratchDir, false);
       Utilities.createTmpDirs(cloned, (ReduceWork) work);
       cloned.set(Utilities.MAPRED_REDUCER_CLASS, ExecReducer.class.getName());

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

@@ -41,6 +41,7 @@
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.exec.mr.ExecDriver;
+import org.apache.hadoop.hive.ql.exec.spark.SparkTask;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.metadata.Hive;
@@ -280,6 +281,10 @@ public void compile(final ParseContext pCtx, final List<Task<? extends Serializa
       for (ExecDriver tsk : mrTasks) {
         tsk.setRetryCmdWhenFail(true);
       }
+      List<SparkTask> sparkTasks = Utilities.getSparkTasks(rootTasks);
+      for (SparkTask sparkTask : sparkTasks) {
+        sparkTask.setRetryCmdWhenFail(true);
+      }
     }
 
     Interner<TableDesc> interner = Interners.newStrongInterner();