diff --git ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java index ed303d1..62fc770 100644 --- ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java +++ ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.ql.io; import java.io.BufferedOutputStream; +import java.io.Closeable; import java.io.DataInput; import java.io.DataInputStream; import java.io.DataOutput; @@ -912,7 +913,7 @@ public class RCFile { * Read KeyBuffer/ValueBuffer pairs from a RCFile. * */ - public static class Reader { + public static class Reader implements Closeable { private final Path file; private final FSDataInputStream in; diff --git ql/src/java/org/apache/hadoop/hive/ql/io/RCFileInputFormat.java ql/src/java/org/apache/hadoop/hive/ql/io/RCFileInputFormat.java index 4c36eff..9f3ecf6 100644 --- ql/src/java/org/apache/hadoop/hive/ql/io/RCFileInputFormat.java +++ ql/src/java/org/apache/hadoop/hive/ql/io/RCFileInputFormat.java @@ -21,10 +21,13 @@ package org.apache.hadoop.hive.ql.io; import java.io.IOException; import java.util.ArrayList; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.serde2.columnar.BytesRefArrayWritable; +import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.mapred.FileInputFormat; @@ -43,6 +46,8 @@ import org.apache.hadoop.mapred.Reporter; public class RCFileInputFormat extends FileInputFormat implements InputFormatChecker { + private static final Log LOG = LogFactory.getLog(RCFileInputFormat.class); + public RCFileInputFormat() { setMinSplitSize(SequenceFile.SYNC_INTERVAL); } @@ -64,12 +69,14 @@ public class RCFileInputFormat files) throws IOException { @@ -39,12 +44,14 @@ public class SequenceFileInputFormatChecker implements InputFormatChecker { return false; } for (int fileId = 0; fileId < files.size(); fileId++) { + SequenceFile.Reader reader = null; try { - SequenceFile.Reader reader = new SequenceFile.Reader(fs, files.get( + reader = new SequenceFile.Reader(fs, files.get( fileId).getPath(), conf); - reader.close(); } catch (IOException e) { return false; + } finally { + IOUtils.cleanup(LOG, reader); } } return true;