/**
 * Copyright (c) 2011, Cloudera, Inc. All Rights Reserved.
 *
 * Cloudera, Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"). You may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied. See the License for
 * the specific language governing permissions and limitations under the
 * License.
 */
package com.cloudera.crunch.io.text;

import java.io.IOException;

import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import com.cloudera.crunch.io.CompositePathIterable;
import com.cloudera.crunch.io.ReadableSourceTarget;
import com.cloudera.crunch.io.SourceTargetHelper;
import com.cloudera.crunch.type.PType;
import com.cloudera.crunch.type.avro.AvroTypeFamily;
import com.cloudera.crunch.type.avro.AvroUtf8InputFormat;

public class TextFileSourceTarget<T> extends TextFileTarget implements ReadableSourceTarget<T> {

  private static final Log LOG = LogFactory.getLog(TextFileSourceTarget.class);
  
  private final PType<T> ptype;
  
  public TextFileSourceTarget(String path, PType<T> ptype) {
    this(new Path(path), ptype);
  }
  
  public TextFileSourceTarget(Path path, PType<T> ptype) {
    super(path);
    this.ptype = ptype;
  }
  
  @Override
  public PType<T> getType() {
    return ptype;
  }

  @Override
  public boolean equals(Object other) {
    if (other == null || !(other instanceof TextFileSourceTarget)) {
      return false;
    }
    TextFileSourceTarget o = (TextFileSourceTarget) other;
    return path.equals(o.path) && ptype.equals(o.ptype);
  }
  
  @Override
  public int hashCode() {
    return new HashCodeBuilder().append(path).toHashCode();
  }
  
  @Override
  public String toString() {
    return "TextFile(" + path + ")";
  }
  
  @Override
  public void configureSource(Job job, int inputId) throws IOException {
	if (ptype.getFamily().equals(AvroTypeFamily.getInstance())) {
      SourceTargetHelper.configureSource(job, inputId, AvroUtf8InputFormat.class, path);
	} else {
	  if (path.toString().endsWith(".bz2")) {
	    SourceTargetHelper.configureSource(job, inputId, BZip2TextInputFormat.class, path);
	  } else {
        SourceTargetHelper.configureSource(job, inputId, TextInputFormat.class, path);
	  }
	}
  }

  @Override
  public long getSize(Configuration conf) {
    try {
      long sz = SourceTargetHelper.getPathSize(conf, path);
      if (path.toString().endsWith(".bz")) {
        sz *= 10;
      }
      return sz;
    } catch (IOException e) {
      LOG.info(String.format("Exception thrown looking up size of: %s", path), e);
    }
    return 1L;
  }

  @Override
  public Iterable<T> read(Configuration conf) throws IOException {
	return CompositePathIterable.create(FileSystem.get(conf), path,
	    new TextFileReaderFactory<T>(ptype));
  }
}

@@ -2240,32 +2240,171 @@
 h + %22)%22;%0A  %7D%0A  %0A
+  private boolean isBZip2() %7B%0A    String strPath = path.toString();%0A    return strPath.endsWith(%22.bz%22) %7C%7C strPath.endsWith(%22.bz2%22);%0A  %7D%0A  %0A
   @Override%0A  pu
@@ -2391,32 +2391,32 @@
 %0A  %0A  @Override%0A
-
   public void co
@@ -2645,39 +2645,16 @@
 if (
-path.toString().endsWith(%22.bz2%22
+isBZip2(
 )) %7B
@@ -2995,60 +2995,70 @@
 if (
-path.toString().endsWith(%22.bz%22)) %7B%0A        sz *= 10;
+isBZip2()) %7B%0A        sz *= 10; // Arbitrary compression factor
 %0A   

Refactor bz2 logic into its own function