diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/SkippingTextInputFormat.java b/ql/src/java/org/apache/hadoop/hive/ql/io/SkippingTextInputFormat.java index 45634a098d68..448af3128de3 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/SkippingTextInputFormat.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/SkippingTextInputFormat.java @@ -21,12 +21,15 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; import org.apache.hadoop.io.compress.CompressionCodecFactory; import org.apache.hadoop.mapred.FileSplit; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.TextInputFormat; +import org.apache.hadoop.util.LineReader; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.ArrayDeque; import java.util.Map; import java.util.Queue; @@ -105,16 +108,18 @@ private long getCachedStartIndex(Path path) throws IOException { } Long startIndexForFile = startIndexMap.get(path); if (startIndexForFile == null) { - FileSystem fileSystem; - FSDataInputStream fis = null; - fileSystem = path.getFileSystem(conf); - try { - fis = fileSystem.open(path); - long currPos = fis.getPos(); + FileSystem fileSystem = path.getFileSystem(conf); + // Hadoop's LineReader counts bytes internally and handles '\n', '\r\n' and + // lone '\r', so we avoid the unreliable readLine()+getPos() idiom that throws + // ClassCastException on lone-CR files. + try (FSDataInputStream fis = fileSystem.open(path)) { + LineReader reader = new LineReader(fis); + Text headerLine = new Text(); + long currPos = 0; int delimiterIdx = -1; for (int j = 0; j < headerCount; j++) { - String headerLine = fis.readLine(); - if (headerLine == null) { + int bytesRead = reader.readLine(headerLine); + if (bytesRead == 0) { startIndexMap.put(path, Long.MAX_VALUE); return Long.MAX_VALUE; } @@ -122,12 +127,17 @@ private long getCachedStartIndex(Path path) throws IOException { String delimiter = conf.get("textinputformat.record.delimiter"); // If record delimiter is defined if (delimiter != null && !delimiter.isEmpty()) { - delimiterIdx = headerLine.indexOf(delimiter); + // Decode as ISO-8859-1 (one char per byte) so the delimiter's index is a + // byte offset, matching currPos. Text.toString() would decode UTF-8 and + // shift the index for multi-byte header bytes preceding the delimiter. + String lastHeader = + new String(headerLine.getBytes(), 0, headerLine.getLength(), StandardCharsets.ISO_8859_1); + delimiterIdx = lastHeader.indexOf(delimiter); } else { - currPos = fis.getPos(); + currPos += bytesRead; } } else { - currPos = fis.getPos(); + currPos += bytesRead; } } // Readers skip the entire first row if the start index of the @@ -136,10 +146,6 @@ private long getCachedStartIndex(Path path) throws IOException { // is discarded instead of the first valid input row. // We consider record delimiters if they exist. startIndexForFile = currPos + delimiterIdx; - } finally { - if (fis != null) { - fis.close(); - } } startIndexMap.put(path, startIndexForFile); } @@ -160,20 +166,24 @@ private long getCachedEndIndex(Path path) throws IOException { // we need 'footer count' lines and one space for EOF LineBuffer buffer = new LineBuffer(footerCount + 1); - FSDataInputStream fis = null; - try { - fis = fileSystem.open(path); + try (FSDataInputStream fis = fileSystem.open(path)) { + Text line = new Text(); while (bufferSectionEnd > bufferSectionStart) { fis.seek(bufferSectionStart); - long pos = fis.getPos(); + // Fresh reader per seek; offsets are seek position + bytes consumed. + LineReader reader = new LineReader(fis); + long consumed = 0; + long pos = bufferSectionStart; while (pos < bufferSectionEnd) { - if (fis.readLine() == null) { + int bytesRead = reader.readLine(line); + if (bytesRead == 0) { // if there is not enough lines in this section, check the previous // section. If this is the beginning section, there are simply not // enough lines in the file. break; } - pos = fis.getPos(); + consumed += bytesRead; + pos = bufferSectionStart + consumed; buffer.consume(pos, bufferSectionEnd); } if (buffer.getRemainingLineCount() == 0) { @@ -193,10 +203,6 @@ private long getCachedEndIndex(Path path) throws IOException { // there were not enough lines in the file to consume all footer rows. endIndexForFile = Long.MIN_VALUE; } - } finally { - if (fis != null) { - fis.close(); - } } } endIndexMap.put(path, endIndexForFile); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/TestSkippingTextInputFormat.java b/ql/src/test/org/apache/hadoop/hive/ql/io/TestSkippingTextInputFormat.java index 9607706e7950..5afe024ed228 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/io/TestSkippingTextInputFormat.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/TestSkippingTextInputFormat.java @@ -45,6 +45,7 @@ import java.util.LinkedHashMap; import java.util.List; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; /** @@ -194,6 +195,135 @@ public void testSkipCompressedFileSplits() throws Exception { } } + /** + * Reproduces ClassCastException (PushbackInputStream cannot be cast to Seekable) + * during split generation when a header line ends in a lone CR ('\r') followed + * by a non-'\n' byte (classic-Mac line endings). readLine() swaps the stream's + * inner input for a non-Seekable PushbackInputStream; the following getPos() casts. + * With the byte-counting reader this now succeeds instead of throwing. + */ + @Test + public void testSkipFileSplitsLoneCR() throws Exception { + FileInputFormat.setInputPaths(job, dataDir); + Path loneCrFile = new Path(dataDir, "data1_cr_only.csv"); + // Exact bytes from HIVE-29785: + // printf "id;name;place;\n1;smruti;ctc;\n2;biswal;bbsr;\n3;'';NULL;\n" | tr '\n' '\r' + writeTextFile(loneCrFile, + "id;name;place;\r" + + "1;smruti;ctc;\r" + + "2;biswal;bbsr;\r" + + "3;'';NULL;\r"); + + SkippingTextInputFormat inputFormat = new SkippingTextInputFormat(); + inputFormat.configure(job, 1, 0); // skip.header.line.count = 1, no footer + FileInputFormat.setInputPaths(job, loneCrFile); + + // On unmodified master this throws ClassCastException during split generation. + InputSplit[] splits = inputFormat.getSplits(job, 1); + assertTrue(splits.length >= 1); + + // Read every row back: the header must be skipped and no data row truncated, + // i.e. SELECT COUNT(*) would return the 3 data rows. + List received = new ArrayList(); + for (InputSplit split : splits) { + RecordReader reader = + inputFormat.getRecordReader(split, job, reporter); + LongWritable key = reader.createKey(); + Text value = reader.createValue(); + while (reader.next(key, value)) { + received.add(value.toString()); + } + reader.close(); + } + assertEquals(3, received.size()); + assertEquals("1;smruti;ctc;", received.get(0)); + assertEquals("2;biswal;bbsr;", received.get(1)); + assertEquals("3;'';NULL;", received.get(2)); + } + + /** + * The lone-CR fix must not move split boundaries for the well-behaved cases. + * The same logical content is written with LF, lone-CR and CRLF terminators; + * LF and lone-CR share a one-byte terminator so their (start, length) must be + * identical, while CRLF's two-byte terminator shifts the header boundary by one. + */ + @Test + public void testSkipHeaderSplitOffsetsAcrossLineEndings() throws Exception { + Path lf = new Path(dataDir, "lf.csv"); + writeTextFile(lf, + "id;name;place;\n1;smruti;ctc;\n2;biswal;bbsr;\n3;'';NULL;\n"); + Path cr = new Path(dataDir, "cr.csv"); + writeTextFile(cr, + "id;name;place;\r1;smruti;ctc;\r2;biswal;bbsr;\r3;'';NULL;\r"); + Path crlf = new Path(dataDir, "crlf.csv"); + writeTextFile(crlf, + "id;name;place;\r\n1;smruti;ctc;\r\n2;biswal;bbsr;\r\n3;'';NULL;\r\n"); + + FileSplit lfSplit = singleHeaderSplit(lf); + FileSplit crSplit = singleHeaderSplit(cr); + FileSplit crlfSplit = singleHeaderSplit(crlf); + + // LF and lone-CR: identical byte layout (1-byte terminator) => identical split. + assertEquals(14, lfSplit.getStart()); + assertEquals(41, lfSplit.getLength()); + assertEquals(lfSplit.getStart(), crSplit.getStart()); + assertEquals(lfSplit.getLength(), crSplit.getLength()); + + // CRLF: 2-byte terminator shifts the header boundary by one; file is 4 bytes longer. + assertEquals(15, crlfSplit.getStart()); + assertEquals(44, crlfSplit.getLength()); + } + + /** + * Exercises the footer detection path (getCachedEndIndex) with lone-CR line + * endings, which throws the same ClassCastException on unmodified master. + * The header and footer rows must be skipped and the two data rows read back. + */ + @Test + public void testSkipFileSplitsLoneCRHeaderFooter() throws Exception { + FileInputFormat.setInputPaths(job, dataDir); + Path file = new Path(dataDir, "cr_header_footer.csv"); + writeTextFile(file, + "dir1_header\r" + + "dir1_file1_line1\r" + + "dir1_file1_line2\r" + + "dir1_footer"); + + SkippingTextInputFormat inputFormat = new SkippingTextInputFormat(); + inputFormat.configure(job, 1, 1); // skip one header and one footer line + FileInputFormat.setInputPaths(job, file); + InputSplit[] splits = inputFormat.getSplits(job, 2); + + List received = new ArrayList(); + for (int i = 0; i < splits.length; i++) { + RecordReader reader = + inputFormat.getRecordReader(splits[i], job, reporter); + LongWritable key = reader.createKey(); + Text value = reader.createValue(); + while (reader.next(key, value)) { + received.add(value.toString()); + } + reader.close(); + } + assertEquals(2, received.size()); + assertTrue(!received.get(0).contains("header")); + assertTrue(!received.get(received.size() - 1).contains("footer")); + } + + /** + * Generates the single (header-adjusted) split for the given file with + * skip.header.line.count=1 and no footer. + */ + private FileSplit singleHeaderSplit(Path file) throws Exception { + SkippingTextInputFormat inputFormat = new SkippingTextInputFormat(); + inputFormat.configure(job, 1, 0); + FileInputFormat.setInputPaths(job, file); + InputSplit[] splits = inputFormat.getSplits(job, 1); + assertEquals(1, splits.length); + assertTrue(splits[0] instanceof FileSplit); + return (FileSplit) splits[0]; + } + /** * Writes the given string to the given file. */