package com.twitter.elephantbird.cascading2.scheme;

import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.SequenceFileInputFormat;

import com.twitter.elephantbird.mapreduce.input.MapReduceInputFormatWrapper;
import com.twitter.elephantbird.mapreduce.input.combine.DelegateCombineFileInputFormat;

import cascading.flow.FlowProcess;
import cascading.scheme.hadoop.SequenceFile;
import cascading.tap.Tap;
import cascading.tuple.Fields;

/**
 * This scheme allows SequenceFile splits to be combined via the DelegateCombineFileInputFormat
 * before it is read. It can be used to combine inputs for intermediate MR jobs in Cascading.
 *
 * To enable, set cascading.flowconnector.intermediateschemeclass to this class in the Hadoop
 * configuration.
 *
 * @author Akihiro Matsukawa
 */
public class CombinedSequenceFile extends SequenceFile {

  protected CombinedSequenceFile() { super(); }

  public CombinedSequenceFile(Fields fields) { super(fields); }

  @Override
  public void sourceConfInit(
      FlowProcess<JobConf> flowProcess,
      Tap<JobConf, RecordReader, OutputCollector> tap,
      JobConf conf ) {
    super.sourceConfInit(flowProcess, tap, conf);

    // Since the EB combiner works over the mapreduce API while Cascading is on the mapred API,
    // in order to use the EB combiner we must wrap the mapred SequenceFileInputFormat
    // with the MapReduceInputFormatWrapper, then wrap the DelegateCombineFileInputFormat
    // again with DeprecatedInputFormatWrapper to make it compatible with Cascading.
    MapReduceInputFormatWrapper.setWrappedInputFormat(SequenceFileInputFormat.class, conf);
    DelegateCombineFileInputFormat.setDelegateInputFormat(conf, MapReduceInputFormatWrapper.class);
  }

}

@@ -955,114 +955,1414 @@
   pr
-otected CombinedSequenceFile() %7B super(); %7D%0A%0A  public CombinedSequenceFile(Fields fields) %7B super(fields);
+ivate static final String MR_COMPRESS_ENABLE = %22mapreduce.output.fileoutputformat.compress%22;%0A  public static final String COMPRESS_ENABLE = %22elephantbird.cascading.combinedsequencefile.compress.enable%22;%0A%0A  private static final String MR_COMPRESS_TYPE = %22mapreduce.output.fileoutputformat.compress.type%22;%0A  public static final String COMPRESS_TYPE = %22elephantbird.cascading.combinedsequencefile.compress.type%22;%0A%0A  private static final String MR_COMPRESS_CODEC = %22mapreduce.output.fileoutputformat.compress.codec%22;%0A  public static final String COMPRESS_CODEC = %22elephantbird.cascading.combinedsequencefile.compress.codec%22;%0A%0A%0A  protected CombinedSequenceFile() %7B super(); %7D%0A%0A  public CombinedSequenceFile(Fields fields) %7B super(fields); %7D%0A%0A  // We can allow overriding the compression settings for just this scheme here%0A  private void updateJobConfForLocalSettings(JobConf conf) %7B%0A    String localSetCompressionEnabled = conf.get(COMPRESS_ENABLE);%0A    if(localSetCompressionEnabled != null) %7B%0A      conf.set(MR_COMPRESS_ENABLE, localSetCompressionEnabled);%0A    %7D%0A%0A    String localSetCompressionType = conf.get(COMPRESS_TYPE);%0A    if(localSetCompressionType != null) %7B%0A      conf.set(MR_COMPRESS_TYPE, localSetCompressionType);%0A    %7D%0A%0A    String localSetCompressionCodec = conf.get(COMPRESS_CODEC);%0A    if(localSetCompressionCodec != null) %7B%0A      conf.set(MR_COMPRESS_CODEC, localSetCompressionCodec);%0A    %7D%0A 
  %7D%0A%0A
@@ -2572,16 +2572,58 @@
 conf);%0A%0A
+    updateJobConfForLocalSettings(conf);%0A%0A
     // S
@@ -3169,10 +3169,247 @@
 );%0A  %7D%0A%0A
+  @Override%0A  public void sinkConfInit( FlowProcess%3CJobConf%3E flowProcess, Tap%3CJobConf, RecordReader, OutputCollector%3E tap, JobConf conf )%0A  %7B%0A    super.sinkConfInit(flowProcess, tap, conf);%0A%0A    updateJobConfForLocalSettings(conf);%0A  %7D%0A%0A
 %7D%0A

Add some options so we can influence the compression options of intermediate data written by cascading