PARQUET-353: Release compression resources.

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java

@@ -323,9 +323,7 @@ private static int getMaxPaddingSize(Configuration conf) {
 
   /**
    * constructor used when this OutputFormat in wrapped in another one (In Pig for example)
-   * @param writeSupportClass the class used to convert the incoming records
-   * @param schema the schema of the records
-   * @param extraMetaData extra meta data to be stored in the footer of the file
+   * @param writeSupport the class used to convert the incoming records
    */
   public <S extends WriteSupport<T>> ParquetOutputFormat(S writeSupport) {
     this.writeSupport = writeSupport;
@@ -387,8 +385,6 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
     if (INFO) LOG.info("Min row count for page size check is: " + props.getMinRowCountForPageSizeCheck());
     if (INFO) LOG.info("Min row count for page size check is: " + props.getMaxRowCountForPageSizeCheck());
 
-    CodecFactory codecFactory = new CodecFactory(conf, props.getPageSizeThreshold());
-
     WriteContext init = writeSupport.init(conf);
     ParquetFileWriter w = new ParquetFileWriter(
         conf, init.getSchema(), file, Mode.CREATE, blockSize, maxPaddingSize);
@@ -411,10 +407,11 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
         init.getSchema(),
         init.getExtraMetaData(),
         blockSize,
-        codecFactory.getCompressor(codec),
+        codec,
         validating,
         props,
-        memoryManager);
+        memoryManager,
+        conf);
   }
 
   /**

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java

@@ -21,13 +21,15 @@
 import java.io.IOException;
 import java.util.Map;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 
 import org.apache.parquet.column.ParquetProperties;
 import org.apache.parquet.column.ParquetProperties.WriterVersion;
 import org.apache.parquet.hadoop.CodecFactory.BytesCompressor;
 import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
 import org.apache.parquet.schema.MessageType;
 
 import static org.apache.parquet.Preconditions.checkNotNull;
@@ -43,8 +45,9 @@
  */
 public class ParquetRecordWriter<T> extends RecordWriter<Void, T> {
 
-  private InternalParquetRecordWriter<T> internalWriter;
-  private MemoryManager memoryManager;
+  private final InternalParquetRecordWriter<T> internalWriter;
+  private final MemoryManager memoryManager;
+  private final CodecFactory codecFactory;
 
   /**
    *
@@ -79,6 +82,7 @@ public ParquetRecordWriter(
     internalWriter = new InternalParquetRecordWriter<T>(w, writeSupport, schema,
         extraMetaData, blockSize, compressor, validating, props);
     this.memoryManager = null;
+    this.codecFactory = null;
   }
 
   /**
@@ -106,14 +110,17 @@ public ParquetRecordWriter(
       boolean validating,
       WriterVersion writerVersion,
       MemoryManager memoryManager) {
-    this(w, writeSupport, schema, extraMetaData, blockSize, compressor,
-        validating, ParquetProperties.builder()
-            .withPageSize(pageSize)
-            .withDictionaryPageSize(dictionaryPageSize)
-            .withDictionaryEncoding(enableDictionary)
-            .withWriterVersion(writerVersion)
-            .build(),
-        memoryManager);
+    ParquetProperties props = ParquetProperties.builder()
+        .withPageSize(pageSize)
+        .withDictionaryPageSize(dictionaryPageSize)
+        .withDictionaryEncoding(enableDictionary)
+        .withWriterVersion(writerVersion)
+        .build();
+    internalWriter = new InternalParquetRecordWriter<T>(w, writeSupport, schema,
+        extraMetaData, blockSize, compressor, validating, props);
+    this.memoryManager = checkNotNull(memoryManager, "memoryManager");
+    memoryManager.addWriter(internalWriter, blockSize);
+    this.codecFactory = null;
   }
 
   /**
@@ -123,7 +130,7 @@ public ParquetRecordWriter(
    * @param schema the schema of the records
    * @param extraMetaData extra meta data to write in the footer of the file
    * @param blockSize the size of a block in the file (this will be approximate)
-   * @param compressor the compressor used to compress the pages
+   * @param codec the compression codec used to compress the pages
    * @param validating if schema validation should be turned on
    * @param props parquet encoding properties
    */
@@ -133,12 +140,15 @@ public ParquetRecordWriter(
       MessageType schema,
       Map<String, String> extraMetaData,
       long blockSize,
-      BytesCompressor compressor,
+      CompressionCodecName codec,
       boolean validating,
       ParquetProperties props,
-      MemoryManager memoryManager) {
+      MemoryManager memoryManager,
+      Configuration conf) {
+    this.codecFactory = new CodecFactory(conf, props.getPageSizeThreshold());
     internalWriter = new InternalParquetRecordWriter<T>(w, writeSupport, schema,
-        extraMetaData, blockSize, compressor, validating, props);
+        extraMetaData, blockSize, codecFactory.getCompressor(codec), validating,
+        props);
     this.memoryManager = checkNotNull(memoryManager, "memoryManager");
     memoryManager.addWriter(internalWriter, blockSize);
   }
@@ -148,9 +158,16 @@ public ParquetRecordWriter(
    */
   @Override
   public void close(TaskAttemptContext context) throws IOException, InterruptedException {
-    internalWriter.close();
-    if (memoryManager != null) {
-      memoryManager.removeWriter(internalWriter);
+    try {
+      internalWriter.close();
+      // release after the writer closes in case it is used for a last flush
+    } finally {
+      if (codecFactory != null) {
+        codecFactory.release();
+      }
+      if (memoryManager != null) {
+        memoryManager.removeWriter(internalWriter);
+      }
     }
   }
 

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java

@@ -52,6 +52,7 @@ public class ParquetWriter<T> implements Closeable {
   public static final int MAX_PADDING_SIZE_DEFAULT = 0;
 
   private final InternalParquetRecordWriter<T> writer;
+  private final CodecFactory codecFactory;
 
   /**
    * Create a new ParquetWriter.
@@ -273,7 +274,7 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport<T> writeSupport
         conf, schema, file, mode, blockSize, maxPaddingSize);
     fileWriter.start();
 
-    CodecFactory codecFactory = new CodecFactory(conf, encodingProps.getPageSizeThreshold());
+    this.codecFactory = new CodecFactory(conf, encodingProps.getPageSizeThreshold());
     CodecFactory.BytesCompressor compressor =	codecFactory.getCompressor(compressionCodecName);
     this.writer = new InternalParquetRecordWriter<T>(
         fileWriter,
@@ -300,6 +301,9 @@ public void close() throws IOException {
       writer.close();
     } catch (InterruptedException e) {
       throw new IOException(e);
+    } finally {
+      // release after the writer closes in case it is used for a last flush
+      codecFactory.release();
     }
   }