[WIP] [SQL] Experiment with speculatively running without reset() in HiveCompatibilitySuite

core/src/main/java/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriter.java

@@ -107,6 +107,14 @@ public void insertAll(Iterator<Product2<K, V>> records) throws IOException {
         blockManager.diskBlockManager().createTempShuffleBlock();
       final File file = tempShuffleBlockIdPlusFile._2();
       final BlockId blockId = tempShuffleBlockIdPlusFile._1();
+      // Note that we purposely do not call open() on the disk writers here; DiskBlockObjectWriter
+      // will automatically open() itself if necessary. This is an optimization to avoid file
+      // creation and truncation for empty partitions; this optimization probably doesn't make sense
+      // for most realistic production workloads, but it can make a large difference when playing
+      // around with Spark SQL queries in spark-shell on toy datasets: if you performed a query over
+      // an extremely small number of records then Spark SQL's default parallelism of 200 would
+      // result in slower out-of-the-box performance due to these constant-factor overheads. This
+      // optimization speeds up local microbenchmarking and SQL unit tests.
       partitionWriters[i] =
         blockManager.getDiskWriter(blockId, file, serInstance, fileBufferSize, writeMetrics);
     }
@@ -144,6 +152,10 @@ public long[] writePartitionedFile(
     try {
       for (int i = 0; i < numPartitions; i++) {
         if (partitionWriters[i].fileSegment().length() == 0) {
+          // In insertAll(), we didn't create empty files for empty reduce partitions; this branch
+          // handles that case. Since we'll be skipping deletion of these files, verify that they
+          // don't exist:
+          assert(!partitionWriters[i].fileSegment().file().exists());
           continue;
         }
         final FileInputStream in = new FileInputStream(partitionWriters[i].fileSegment().file());
@@ -175,7 +187,8 @@ public void stop() throws IOException {
         for (BlockObjectWriter writer : partitionWriters) {
           // This method explicitly does _not_ throw exceptions:
           writer.revertPartialWritesAndClose();
-          if (!diskBlockManager.getFile(writer.blockId()).delete()) {
+          final File file = diskBlockManager.getFile(writer.blockId());
+          if (file.exists() && !file.delete()) {
             logger.error("Error while deleting file for block {}", writer.blockId());
           }
         }

core/src/main/scala/org/apache/spark/storage/BlockObjectWriter.scala

@@ -187,13 +187,12 @@ private[spark] class DiskBlockObjectWriter(
         objOut.flush()
         bs.flush()
         close()
-      }
-
-      val truncateStream = new FileOutputStream(file, true)
-      try {
-        truncateStream.getChannel.truncate(initialPosition)
-      } finally {
-        truncateStream.close()
+        val truncateStream = new FileOutputStream(file, true)
+        try {
+          truncateStream.getChannel.truncate(initialPosition)
+        } finally {
+          truncateStream.close()
+        }
       }
     } catch {
       case e: Exception =>