DRILL-8478. HashPartition memory leak when it allocate memory excepti…

…on with OutOfMemoryException (#2874)

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/common/HashPartition.java

@@ -147,6 +147,11 @@ public HashPartition(FragmentContext context, BufferAllocator allocator, Chained
 
     try {
       this.hashTable = baseHashTable.createAndSetupHashTable(null);
+      this.hjHelper = semiJoin ? null : new HashJoinHelper(context, allocator);
+      tmpBatchesList = new ArrayList<>();
+      if (numPartitions > 1) {
+        allocateNewCurrentBatchAndHV();
+      }
     } catch (ClassTransformationException e) {
       throw UserException.unsupportedError(e)
         .message("Code generation error - likely an error in the code.")
@@ -157,11 +162,11 @@ public HashPartition(FragmentContext context, BufferAllocator allocator, Chained
         .build(logger);
     } catch (SchemaChangeException sce) {
       throw new IllegalStateException("Unexpected Schema Change while creating a hash table",sce);
-    }
-    this.hjHelper = semiJoin ? null : new HashJoinHelper(context, allocator);
-    tmpBatchesList = new ArrayList<>();
-    if (numPartitions > 1) {
-      allocateNewCurrentBatchAndHV();
+    } catch (OutOfMemoryException oom) {
+      close();
+      throw UserException.memoryError(oom)
+          .message("OutOfMemory while allocate memory for hash partition.")
+          .build(logger);
     }
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/join/AbstractHashBinaryRecordBatch.java

@@ -81,6 +81,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 import static org.apache.drill.exec.record.RecordBatch.IterOutcome.EMIT;
@@ -1312,7 +1313,9 @@ private void cleanup() {
     }
     // clean (and deallocate) each partition, and delete its spill file
     for (HashPartition partn : partitions) {
-      partn.close();
+      if (Objects.nonNull(partn)) {
+        partn.close();
+      }
     }
 
     // delete any spill file left in unread spilled partitions