Address Julien's comment

apache · Feb 27, 2015 · b9abab0 · b9abab0
1 parent 965af7f
commit b9abab0
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 14 deletions.
diff --git a/parquet-encoding/src/main/java/parquet/bytes/ConcatenatingByteArrayCollector.java b/parquet-encoding/src/main/java/parquet/bytes/ConcatenatingByteArrayCollector.java
@@ -11,11 +11,8 @@ public class ConcatenatingByteArrayCollector extends BytesInput {
   private final List<byte[]> slabs = new ArrayList<byte[]>();
   private long size = 0;
 
-  public void collect(BytesInput bytes) throws IOException {
-    collect(bytes.toByteArray());
-  }
-
-  public void collect(byte[] bytes) {
+  public void collect(BytesInput bytesInput) throws IOException {
+    byte[] bytes = bytesInput.toByteArray();
     slabs.add(bytes);
     size += bytes.length;
   }

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageWriteStore.java
@@ -28,7 +28,6 @@
 
 import parquet.Log;
 import parquet.bytes.BytesInput;
-import parquet.bytes.CapacityByteArrayOutputStream;
 import parquet.bytes.ConcatenatingByteArrayCollector;
 import parquet.column.ColumnDescriptor;
 import parquet.column.Encoding;
@@ -102,14 +101,14 @@ public void writePage(BytesInput bytes,
           dlEncoding,
           valuesEncoding,
           tempOutputStream);
-      buf.collect(tempOutputStream.toByteArray());
-      tempOutputStream.reset();
       this.uncompressedLength += uncompressedSize;
       this.compressedLength += compressedSize;
       this.totalValueCount += valueCount;
       this.pageCount += 1;
       this.totalStatistics.mergeStatistics(statistics);
-      buf.collect(compressedBytes);
+      // by concatenating before collecting instead of collecting twice,
+      // we only allocate one buffer to copy into instead of multiple.
+      buf.collect(BytesInput.concat(BytesInput.from(tempOutputStream), compressedBytes));
       encodings.add(rlEncoding);
       encodings.add(dlEncoding);
       encodings.add(valuesEncoding);
@@ -140,16 +139,21 @@ public void writePageV2(
           rlByteLength,
           dlByteLength,
           tempOutputStream);
-      buf.collect(tempOutputStream.toByteArray());
-      tempOutputStream.reset();
       this.uncompressedLength += uncompressedSize;
       this.compressedLength += compressedSize;
       this.totalValueCount += valueCount;
       this.pageCount += 1;
       this.totalStatistics.mergeStatistics(statistics);
-      buf.collect(repetitionLevels);
-      buf.collect(definitionLevels);
-      buf.collect(compressedData);
+
+      // by concatenating before collecting instead of collecting twice,
+      // we only allocate one buffer to copy into instead of multiple.
+      buf.collect(
+          BytesInput.concat(
+            BytesInput.from(tempOutputStream),
+            repetitionLevels,
+            definitionLevels,
+            compressedData)
+      );
       encodings.add(dataEncoding);
     }