[Gluten-995] Fix memory leak for ClickHouse Backend

Fix memory leak for ClickHouse Backend.

There are memory leak after executing BHJ.

Close apache#995.

backends-clickhouse/src/main/java/io/glutenproject/vectorized/OnHeapCopyShuffleInputStream.java

@@ -23,7 +23,7 @@
 
 public class OnHeapCopyShuffleInputStream implements ShuffleInputStream {
 
-  private final InputStream in;
+  private InputStream in;
   private final boolean isCompressed;
   private int bufferSize;
   private long bytesRead = 0L;
@@ -78,6 +78,7 @@ public boolean isCompressed() {
   public void close() {
     try {
       in.close();
+      in = null;
     } catch (IOException e) {
       throw new RuntimeException(e);
     }

backends-clickhouse/src/main/java/io/glutenproject/vectorized/StorageJoinBuilder.java

@@ -32,7 +32,7 @@
 import org.apache.spark.sql.catalyst.expressions.Attribute;
 import org.apache.spark.sql.catalyst.expressions.Expression;
 
-public class StorageJoinBuilder {
+public class StorageJoinBuilder implements AutoCloseable {
   private ShuffleInputStream in;
 
   private int customizeBufferSize;
@@ -107,4 +107,13 @@ public void build() {
         join,
         structure);
   }
+
+  @Override
+  public void close() throws Exception {
+    try {
+      in.close();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
 }

backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/joins/ClickHouseBuildSideRelation.scala

@@ -65,6 +65,7 @@ case class ClickHouseBuildSideRelation(
     )
     // Build the hash table
     storageJoinBuilder.build()
+    storageJoinBuilder.close()
     this
   }
 

backends-clickhouse/src/test/scala/org/apache/spark/sql/execution/benchmarks/CHAggAndShuffleBenchmark.scala

@@ -92,7 +92,7 @@ object CHAggAndShuffleBenchmark extends SqlBasedBenchmark {
       .setIfMissing("spark.executor.memory", memorySize)
       .setIfMissing("spark.sql.files.maxPartitionBytes", "1G")
       .setIfMissing("spark.sql.files.openCostInBytes", "1073741824")
-      .setIfMissing("spark.gluten.sql.columnar.coalesce.batches", "true")
+      .setIfMissing("spark.gluten.sql.columnar.coalesce.batches", "false")
       .setIfMissing("spark.shuffle.manager", "sort")
       .setIfMissing("spark.io.compression.codec", "SNAPPY")