package parquet.hadoop.mapred;

import static parquet.Log.INFO;
import parquet.Log;
import parquet.hadoop.ParquetOutputFormat;
import parquet.hadoop.ParquetRecordWriter;
import parquet.hadoop.metadata.CompressionCodecName;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Progressable;

@SuppressWarnings("deprecation")
public class DeprecatedParquetOutputFormat<V> extends org.apache.hadoop.mapred.FileOutputFormat<Void, V> {
  private static final Log LOG = Log.getLog(ParquetOutputFormat.class);

  protected ParquetOutputFormat<V> realOutputFormat = new ParquetOutputFormat<V>();

  @Override
  public RecordWriter<Void, V> getRecordWriter(FileSystem fs,
      JobConf conf, String name, Progressable progress) throws IOException {
    return new RecordWriterWrapper<V>(realOutputFormat, fs, conf, name, progress);
  }

  private static CompressionCodecName getCodec(JobConf conf) {
    CompressionCodecName codec;

    if (ParquetOutputFormat.isCompressionSet(conf)) { // explicit parquet config
      codec = ParquetOutputFormat.getCompression(conf);
    } else if (getCompressOutput(conf)) { // from hadoop config
      // find the right codec
      Class<?> codecClass = getOutputCompressorClass(conf, DefaultCodec.class);
      if (INFO) LOG.info("Compression set through hadoop codec: " + codecClass.getName());
      codec = CompressionCodecName.fromCompressionCodec(codecClass);
    } else {
      if (INFO) LOG.info("Compression set to false");
      codec = CompressionCodecName.UNCOMPRESSED;
    }

    if (INFO) LOG.info("Compression: " + codec.name());
    return codec;
  }

  private static Path getDefaultWorkFile(JobConf conf, String name, String extension) {
    String file = getUniqueName(conf, name) + extension;
    return new Path(getWorkOutputPath(conf), file);
  }

  private static class RecordWriterWrapper<V> implements RecordWriter<Void, V> {

    private ParquetRecordWriter<V> realWriter;

    public RecordWriterWrapper(ParquetOutputFormat<V> realOutputFormat,
        FileSystem fs, JobConf conf, String name, Progressable progress) throws IOException {

      CompressionCodecName codec = getCodec(conf);
      String extension = codec.getExtension() + ".parquet";
      Path file = getDefaultWorkFile(conf, name, extension);

      try {
        realWriter = (ParquetRecordWriter<V>) realOutputFormat.getRecordWriter(conf, file, codec);
      } catch (InterruptedException e) {
        Thread.interrupted();
        throw new IOException(e);
      }
    }

    @Override
    public void close(Reporter reporter) throws IOException {
      try {
        realWriter.close(null);
      } catch (InterruptedException e) {
        Thread.interrupted();
        throw new IOException(e);
      }
    }

    @Override
    public void write(Void key, V value) throws IOException {
      try {
        realWriter.write(key, value);
      } catch (InterruptedException e) {
        Thread.interrupted();
        throw new IOException(e);
      }
    }
  }
}

@@ -246,16 +246,61 @@
 ption;%0A%0A
+import org.apache.hadoop.conf.Configuration;%0A
 import o
@@ -810,101 +810,358 @@
 %0A  p
-rotected ParquetOutputFormat%3CV%3E realOutputFormat = new ParquetOutputFormat%3CV%3E(
+ublic static void setWriteSupportClass(Configuration configuration,  Class%3C?%3E writeSupportClass) %7B%0A    configuration.set(ParquetOutputFormat.WRITE_SUPPORT_CLASS, writeSupportClass.getName());%0A  %7D%0A%0A  public static void setBlockSize(Configuration configuration, int blockSize) %7B%0A    configuration.setInt(ParquetOutputFormat.BLOCK_SIZE, blockSize
 );%0A
-%0A
   
-@Override
+%7D%0A
 %0A  p
@@ -1170,218 +1170,513 @@
 lic 
-RecordWriter%3CVoid, V%3E getRecordWriter(FileSystem fs,%0A      JobConf conf, String name, Progressable progress) throws IOException %7B%0A    return new RecordWriterWrapper%3CV%3E(realOutputFormat, fs, conf, name, progress
+static void setPageSize(Configuration configuration, int pageSize) %7B%0A    configuration.setInt(ParquetOutputFormat.PAGE_SIZE, pageSize);%0A  %7D%0A%0A  public static void setCompression(Configuration configuration, CompressionCodecName compression) %7B%0A    configuration.set(ParquetOutputFormat.COMPRESSION, compression.name());%0A  %7D%0A%0A  public static void setEnableDictionary(Configuration configuration, boolean enableDictionary) %7B%0A    configuration.setBoolean(ParquetOutputFormat.ENABLE_DICTIONARY, enableDictionary
 );%0A 
@@ -2642,32 +2642,356 @@
 f), file);%0A  %7D%0A%0A
+  protected ParquetOutputFormat%3CV%3E realOutputFormat = new ParquetOutputFormat%3CV%3E();%0A%0A  @Override%0A  public RecordWriter%3CVoid, V%3E getRecordWriter(FileSystem fs,%0A      JobConf conf, String name, Progressable progress) throws IOException %7B%0A    return new RecordWriterWrapper%3CV%3E(realOutputFormat, fs, conf, name, progress);%0A  %7D%0A%0A
   private static

add some convenience methods (from ParquetOutputFormat)