Skip to content

HIVE-29785: SkippingTextInputFormat: ClassCastException on skip.header.line.count files with lone-CR (\r) line endings #6680

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
srbiswal wants to merge 1 commit into
base: master
Choose a base branch
from
+208 −23
Open

HIVE-29785: SkippingTextInputFormat: ClassCastException on skip.header.line.count files with lone-CR (\r) line endings #6680

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 78 additions & 23 deletions ql/src/java/org/apache/hadoop/hive/ql/io/SkippingTextInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand All @@ -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
Expand All @@ -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);
}
Expand All @@ -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) {
Expand All @@ -193,17 +193,72 @@ 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);
}
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<Long> queue = new ArrayDeque<Long>();
private int remainingLineEnds;
Expand Down
130 changes: 130 additions & 0 deletions ql/src/test/org/apache/hadoop/hive/ql/io/TestSkippingTextInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import java.util.LinkedHashMap;
import java.util.List;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

/**
Expand Down Expand Up @@ -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<String> received = new ArrayList<String>();
for (InputSplit split : splits) {
RecordReader<LongWritable, Text> 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<String> received = new ArrayList<String>();
for (int i = 0; i < splits.length; i++) {
RecordReader<LongWritable, Text> 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.
*/
Expand Down