HIVE-29124: Avoid committing files when a task is aborted even though some source has completed (#6011)

Author: zhengchenyu

ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java

@@ -305,6 +305,14 @@ public void abort() {
     } else {
       LOG.info("reducer not setup yet. abort not being forwarded");
     }
+    if (mergeWorkList != null) {
+      for (BaseWork redWork : mergeWorkList) {
+        redWork.abort();
+      }
+    }
+    if (reduceWork != null) {
+      reduceWork.abort();
+    }
   }
 
   /**
@@ -343,16 +351,13 @@ void close() {
     }
 
     try {
-      if (isAborted()) {
-        for (ReduceRecordSource rs : sources) {
-          if (!rs.close()) {
-            setAborted(false); // Preserving the old logic. Hmm...
-            break;
-          }
-        }
-      }
-
       boolean abort = isAborted();
+      for (ReduceRecordSource rs : sources) {
+        abort |= !rs.close();
+      }
+      if (abort) {
+        setAborted(true);
+      }
       reducer.close(abort);
       if (mergeWorkList != null) {
         for (BaseWork redWork : mergeWorkList) {

ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java

@@ -520,6 +520,14 @@ private void processVectorGroup(BytesWritable keyWritable,
     }
   }
 
+
+  /**
+   * Closes resources and returns whether the records were successfully processed.
+   * @return boolean indicating the success status:
+   * - true: All data has been processed successfully without exceptions.
+   * - false: Exceptions were encountered during data processing.
+   * @throws Exception unexpected errors occur during closing.
+   */
   boolean close() throws Exception {
     try {
       if (handleGroupKey && groupKey != null) {
@@ -533,7 +541,7 @@ boolean close() throws Exception {
             + e.getMessage(), e);
       }
     }
-    return abort;
+    return !abort;
   }
 
   public ObjectInspector getObjectInspector() {

ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java

@@ -545,4 +545,12 @@ public void setInputSourceToRuntimeValuesInfo(
           String workName, RuntimeValuesInfo runtimeValuesInfo) {
     inputSourceToRuntimeValuesInfo.put(workName, runtimeValuesInfo);
   }
+
+  public void abort() {
+    if (dummyOps != null) {
+      for (Operator<?> dummyOp : dummyOps) {
+        dummyOp.abort();
+      }
+    }
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java

@@ -381,4 +381,12 @@ public void setEdgePropRef(TezEdgeProperty edgeProp) {
   public TezEdgeProperty getEdgePropRef() {
     return edgeProp;
   }
+
+  @Override
+  public void abort() {
+    super.abort();
+    if (reducer != null) {
+      reducer.abort();
+    }
+  }
 }