From 392377a62a134a6109308727945f688b505a0a65 Mon Sep 17 00:00:00 2001 From: srbiswal Date: Thu, 6 Aug 2026 06:40:20 -0700 Subject: [PATCH 1/2] HIVE-29785: SkippingTextInputFormat: ClassCastException on skip.header.line.count files with lone-CR (\r) line endings SkippingTextInputFormat.getCachedStartIndex (header path) and getCachedEndIndex (footer path) called FSDataInputStream.readLine() and then getPos() on the same stream. On a lone '\r' not followed by '\n', DataInputStream.readLine() pushes its look-ahead byte back by replacing the stream's inner input with a non-Seekable PushbackInputStream, so the subsequent getPos() throws ClassCastException during Tez split generation. LF, CRLF and '\r'-at-EOF were unaffected. Replace readLine()+getPos() in both methods with a ByteCountingLineReader that recognizes '\n', '\r\n' and lone '\r' and derives offsets from bytes consumed, so getPos() is never called on a readLine()-mutated stream. Offset semantics are preserved exactly. As a safety net, makeSplitInternal now converts any unexpected RuntimeException from header/footer detection into a clear, file-contextual error instead of a cryptic cast. Adds tests: a lone-CR repro, exact split-offset assertions across LF/CR/CRLF proving no boundary regression, and a footer-path lone-CR case. --- .../hive/ql/io/SkippingTextInputFormat.java | 101 ++++++++++---- .../ql/io/TestSkippingTextInputFormat.java | 130 ++++++++++++++++++ 2 files changed, 208 insertions(+), 23 deletions(-) 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..1f23d5ad53d7 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 @@ -27,6 +27,7 @@ import org.apache.hadoop.mapred.TextInputFormat; import java.io.IOException; +import java.io.InputStream; import java.util.ArrayDeque; import java.util.Map; import java.util.Queue; @@ -78,6 +79,10 @@ private FileSplit makeSplitInternal(Path file, long start, long length, String[] } catch (IOException e) { LOG.warn("Could not detect header/footer", e); return new NullRowsInputFormat.DummyInputSplit(file); + } catch (RuntimeException e) { + // Report unexpected detection failures clearly instead of a cryptic cast. + throw new RuntimeException("Failed to detect header/footer boundaries for file " + + file + " during split generation", e); } if (cachedStart > start + length) { return new NullRowsInputFormat.DummyInputSplit(file); @@ -105,15 +110,14 @@ 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); + // ByteCountingLineReader avoids the unreliable readLine()+getPos() idiom. + try (FSDataInputStream fis = fileSystem.open(path)) { + ByteCountingLineReader reader = new ByteCountingLineReader(fis); + long currPos = 0; int delimiterIdx = -1; for (int j = 0; j < headerCount; j++) { - String headerLine = fis.readLine(); + String headerLine = reader.readLine(); if (headerLine == null) { startIndexMap.put(path, Long.MAX_VALUE); return Long.MAX_VALUE; @@ -124,10 +128,10 @@ private long getCachedStartIndex(Path path) throws IOException { if (delimiter != null && !delimiter.isEmpty()) { delimiterIdx = headerLine.indexOf(delimiter); } else { - currPos = fis.getPos(); + currPos = reader.getBytesConsumed(); } } else { - currPos = fis.getPos(); + currPos = reader.getBytesConsumed(); } } // Readers skip the entire first row if the start index of the @@ -136,10 +140,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 +160,20 @@ 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)) { while (bufferSectionEnd > bufferSectionStart) { fis.seek(bufferSectionStart); - long pos = fis.getPos(); + // Fresh reader per seek; offsets are seek position + bytes consumed. + ByteCountingLineReader reader = new ByteCountingLineReader(fis); + long pos = bufferSectionStart; while (pos < bufferSectionEnd) { - if (fis.readLine() == null) { + if (reader.readLine() == null) { // 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(); + pos = bufferSectionStart + reader.getBytesConsumed(); buffer.consume(pos, bufferSectionEnd); } if (buffer.getRemainingLineCount() == 0) { @@ -193,10 +193,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); @@ -204,6 +200,65 @@ private long getCachedEndIndex(Path path) throws IOException { return endIndexForFile; } + /** + * Reads lines while counting bytes consumed, so offsets can be computed without + * {@link FSDataInputStream#getPos()} -- which is unreliable after {@code readLine()}: + * a lone {@code '\r'} makes it swap in a non-Seekable {@code PushbackInputStream}, so + * the next {@code getPos()} throws {@code ClassCastException}. Handles {@code '\n'}, + * {@code '\r\n'} and lone {@code '\r'}; after {@link #readLine()}, + * {@link #getBytesConsumed()} is the offset just past the line's terminator. + */ + static final class ByteCountingLineReader { + private final InputStream in; + private long bytesConsumed; + // Look-ahead byte past a '\r' (belongs to the next line, not yet counted); -1 = empty. + private int pushedBack = -1; + + ByteCountingLineReader(InputStream in) { + this.in = in; + } + + long getBytesConsumed() { + return bytesConsumed; + } + + private int nextByte() throws IOException { + if (pushedBack != -1) { + int b = pushedBack; + pushedBack = -1; + return b; + } + return in.read(); + } + + /** Returns the next line without its terminator, or {@code null} at end of stream. */ + String readLine() throws IOException { + int c = nextByte(); + if (c == -1) { + return null; + } + StringBuilder sb = new StringBuilder(); + while (c != -1) { + bytesConsumed++; + if (c == '\n') { + return sb.toString(); + } + if (c == '\r') { + int next = nextByte(); + if (next == '\n') { + bytesConsumed++; + } else if (next != -1) { + pushedBack = next; // belongs to the next line, not yet counted + } + return sb.toString(); + } + sb.append((char) c); + c = nextByte(); + } + return sb.toString(); + } + } + static class LineBuffer { private final Queue queue = new ArrayDeque(); private int remainingLineEnds; 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. */ From 837cb65d3075cd58423e7a101f40148c0d590a88 Mon Sep 17 00:00:00 2001 From: srbiswal Date: Wed, 12 Aug 2026 07:36:01 -0700 Subject: [PATCH 2/2] HIVE-29785: Use Hadoop LineReader for header/footer detection and drop redundant RuntimeException wrap Address review feedback on the lone-CR ClassCastException fix: - Replace the hand-rolled ByteCountingLineReader with Hadoop's org.apache.hadoop.util.LineReader, which counts bytes internally via readLine(Text) (no getPos()) and handles '\n', '\r\n' and lone '\r'. It is the standard utility, already used in TextRecordReader. Constructed as new LineReader(fis) so the existing textinputformat.record.delimiter index logic is preserved. - Decode the last header line as ISO-8859-1 (one char per byte) before the delimiter indexOf so the index remains a byte offset, matching the prior implementation exactly for non-ASCII header bytes. - Drop the catch (RuntimeException) re-throw in makeSplitInternal: it wrapped a RuntimeException in the same type and only added the file path, while the reader change already removes the ClassCastException it guarded against. Offset semantics unchanged; existing TestSkippingTextInputFormat and TestLineBuffer pass. --- .../hive/ql/io/SkippingTextInputFormat.java | 99 +++++-------------- 1 file changed, 25 insertions(+), 74 deletions(-) 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 1f23d5ad53d7..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,13 +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.io.InputStream; +import java.nio.charset.StandardCharsets; import java.util.ArrayDeque; import java.util.Map; import java.util.Queue; @@ -79,10 +81,6 @@ private FileSplit makeSplitInternal(Path file, long start, long length, String[] } catch (IOException e) { LOG.warn("Could not detect header/footer", e); return new NullRowsInputFormat.DummyInputSplit(file); - } catch (RuntimeException e) { - // Report unexpected detection failures clearly instead of a cryptic cast. - throw new RuntimeException("Failed to detect header/footer boundaries for file " - + file + " during split generation", e); } if (cachedStart > start + length) { return new NullRowsInputFormat.DummyInputSplit(file); @@ -111,14 +109,17 @@ private long getCachedStartIndex(Path path) throws IOException { Long startIndexForFile = startIndexMap.get(path); if (startIndexForFile == null) { FileSystem fileSystem = path.getFileSystem(conf); - // ByteCountingLineReader avoids the unreliable readLine()+getPos() idiom. + // 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)) { - ByteCountingLineReader reader = new ByteCountingLineReader(fis); + LineReader reader = new LineReader(fis); + Text headerLine = new Text(); long currPos = 0; int delimiterIdx = -1; for (int j = 0; j < headerCount; j++) { - String headerLine = reader.readLine(); - if (headerLine == null) { + int bytesRead = reader.readLine(headerLine); + if (bytesRead == 0) { startIndexMap.put(path, Long.MAX_VALUE); return Long.MAX_VALUE; } @@ -126,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 = reader.getBytesConsumed(); + currPos += bytesRead; } } else { - currPos = reader.getBytesConsumed(); + currPos += bytesRead; } } // Readers skip the entire first row if the start index of the @@ -161,19 +167,23 @@ private long getCachedEndIndex(Path path) throws IOException { // we need 'footer count' lines and one space for EOF LineBuffer buffer = new LineBuffer(footerCount + 1); try (FSDataInputStream fis = fileSystem.open(path)) { + Text line = new Text(); while (bufferSectionEnd > bufferSectionStart) { fis.seek(bufferSectionStart); // Fresh reader per seek; offsets are seek position + bytes consumed. - ByteCountingLineReader reader = new ByteCountingLineReader(fis); + LineReader reader = new LineReader(fis); + long consumed = 0; long pos = bufferSectionStart; while (pos < bufferSectionEnd) { - if (reader.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 = bufferSectionStart + reader.getBytesConsumed(); + consumed += bytesRead; + pos = bufferSectionStart + consumed; buffer.consume(pos, bufferSectionEnd); } if (buffer.getRemainingLineCount() == 0) { @@ -200,65 +210,6 @@ private long getCachedEndIndex(Path path) throws IOException { return endIndexForFile; } - /** - * Reads lines while counting bytes consumed, so offsets can be computed without - * {@link FSDataInputStream#getPos()} -- which is unreliable after {@code readLine()}: - * a lone {@code '\r'} makes it swap in a non-Seekable {@code PushbackInputStream}, so - * the next {@code getPos()} throws {@code ClassCastException}. Handles {@code '\n'}, - * {@code '\r\n'} and lone {@code '\r'}; after {@link #readLine()}, - * {@link #getBytesConsumed()} is the offset just past the line's terminator. - */ - static final class ByteCountingLineReader { - private final InputStream in; - private long bytesConsumed; - // Look-ahead byte past a '\r' (belongs to the next line, not yet counted); -1 = empty. - private int pushedBack = -1; - - ByteCountingLineReader(InputStream in) { - this.in = in; - } - - long getBytesConsumed() { - return bytesConsumed; - } - - private int nextByte() throws IOException { - if (pushedBack != -1) { - int b = pushedBack; - pushedBack = -1; - return b; - } - return in.read(); - } - - /** Returns the next line without its terminator, or {@code null} at end of stream. */ - String readLine() throws IOException { - int c = nextByte(); - if (c == -1) { - return null; - } - StringBuilder sb = new StringBuilder(); - while (c != -1) { - bytesConsumed++; - if (c == '\n') { - return sb.toString(); - } - if (c == '\r') { - int next = nextByte(); - if (next == '\n') { - bytesConsumed++; - } else if (next != -1) { - pushedBack = next; // belongs to the next line, not yet counted - } - return sb.toString(); - } - sb.append((char) c); - c = nextByte(); - } - return sb.toString(); - } - } - static class LineBuffer { private final Queue queue = new ArrayDeque(); private int remainingLineEnds;