MAPREDUCE-5049. CombineFileInputFormat counts all compressed files non-splitable. (sandyr via tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1@1454137 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -531,6 +531,9 @@ Release 1.2.0 - unreleased
     MAPREDUCE-5038. old API CombineFileInputFormat missing fixes that are in 
     new API. (sandyr via tucu)
 
+    MAPREDUCE-5049. CombineFileInputFormat counts all compressed files 
+    non-splitable. (sandyr via tucu)
+
 Release 1.1.2 - 2013.01.30
 
   INCOMPATIBLE CHANGES

src/mapred/org/apache/hadoop/mapred/lib/CombineFileInputFormat.java

@@ -37,6 +37,7 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.SplittableCompressionCodec;
 import org.apache.hadoop.net.NodeBase;
 import org.apache.hadoop.net.NetworkTopology;
 
@@ -135,7 +136,10 @@ public abstract class CombineFileInputFormat<K, V>
   protected boolean isSplitable(FileSystem fs, Path file) {
     final CompressionCodec codec =
       new CompressionCodecFactory(fs.getConf()).getCodec(file);
-    return codec == null;
+    if (null == codec) {
+      return true;
+    }
+    return codec instanceof SplittableCompressionCodec;
   }
   
   /**

src/mapred/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.SplittableCompressionCodec;
 import org.apache.hadoop.mapreduce.InputFormat;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -147,7 +148,10 @@ public abstract class CombineFileInputFormat<K, V>
   protected boolean isSplitable(JobContext context, Path file) {
     final CompressionCodec codec =
       new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
-    return codec == null;
+    if (null == codec) {
+      return true;
+    }
+    return codec instanceof SplittableCompressionCodec;
   }
 
   /**