This is an automated email from the ASF dual-hosted git repository.
deniskuzZ pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 691b786dc46 HIVE-29785: SkippingTextInputFormat: ClassCastException on
skip.header.line.count files with lone-CR (\r) line endings (#6680)
691b786dc46 is described below
commit 691b786dc4675ee7e5df14258f70e358cf11d58c
Author: srbiswal <[email protected]>
AuthorDate: Thu Aug 20 22:47:26 2026 +0530
HIVE-29785: SkippingTextInputFormat: ClassCastException on
skip.header.line.count files with lone-CR (\r) line endings (#6680)
---
.../hadoop/hive/ql/io/SkippingTextInputFormat.java | 74 +++++----
.../hive/ql/io/TestSkippingTextInputFormat.java | 168 +++++++++++++++++++++
2 files changed, 209 insertions(+), 33 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 2e546117289..fe037b76f2c 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
@@ -22,12 +22,14 @@
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.Text;
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;
@@ -106,29 +108,35 @@ 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);
+ String delimiter = conf.get("textinputformat.record.delimiter");
+ boolean hasDelimiter = delimiter != null && !delimiter.isEmpty();
+ try (FSDataInputStream fis = fileSystem.open(path)) {
+ // LineReader terminates on '\n', '\r\n' and lone '\r' and returns the
number of
+ // bytes it consumed, so offsets come from that count instead of the
+ // readLine()+getPos() idiom, which throws ClassCastException on
lone-CR files.
+ // Deliberately not closed: LineReader.close() would close fis, which
the
+ // try-with-resources already owns.
+ LineReader reader = new LineReader(fis, conf);
+ 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;
}
- if (j == headerCount-1) {
- String delimiter = conf.get("textinputformat.record.delimiter");
- // If record delimiter is defined
- if (delimiter != null && !delimiter.isEmpty()) {
- delimiterIdx = headerLine.indexOf(delimiter);
- } else {
- currPos = fis.getPos();
- }
+ if (j == headerCount - 1 && hasDelimiter) {
+ // Decode one char per byte so indexOf() yields a byte offset
comparable with
+ // currPos. This preserves the old DataInputStream.readLine(),
which
+ // zero-extended each byte to a char; Text.toString() would decode
UTF-8 and
+ // shift the index whenever the header holds multi-byte characters.
+ String lastHeader =
+ new String(headerLine.getBytes(), 0, headerLine.getLength(),
StandardCharsets.ISO_8859_1);
+ delimiterIdx = lastHeader.indexOf(delimiter);
} else {
- currPos = fis.getPos();
+ currPos += bytesRead;
}
}
// Readers skip the entire first row if the start index of the
@@ -137,10 +145,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);
}
@@ -150,7 +154,7 @@ private long getCachedStartIndex(Path path) throws
IOException {
private long getCachedEndIndex(Path path) throws IOException {
Long endIndexForFile = endIndexMap.get(path);
if (endIndexForFile == null) {
- final long bufferSectionSize = 5 * 1024;
+ final int bufferSectionSize = 5 * 1024;
FileSystem fileSystem = path.getFileSystem(conf);
long endOfFile = fileSystem.getFileStatus(path).getLen();
if (footerCount == 0) {
@@ -161,20 +165,28 @@ 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();
+ // A LineReader buffers read-ahead and is not seek-aware, so it is
valid only
+ // for the region following this seek and must not outlive the
section: a
+ // reader carried across the seek would keep serving bytes from
the previously
+ // scanned section while offsets were counted from the new start.
The single
+ // buffer of over-read past bufferSectionEnd is expected and
discarded by
+ // LineBuffer.consume(). Deliberately not closed: that would close
fis, which
+ // the remaining sections still need.
+ LineReader reader = new LineReader(fis, conf);
+ Text line = new Text();
+ 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();
+ pos += bytesRead;
buffer.consume(pos, bufferSectionEnd);
}
if (buffer.getRemainingLineCount() == 0) {
@@ -194,10 +206,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 d53f5363347..272f48d6975 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
@@ -42,10 +42,12 @@
import java.io.File;
import java.io.IOException;
import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
/**
@@ -195,6 +197,161 @@ 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"));
+ }
+
+ /**
+ * Header skipping with a custom textinputformat.record.delimiter must
locate the
+ * delimiter by BYTE offset. The last header line is decoded one char per
byte
+ * (ISO-8859-1) so indexOf() lines up with the byte-counting currPos;
decoding the
+ * header as UTF-8 (Text.toString()) would shrink the index by one for every
+ * multi-byte character preceding the delimiter and shift the split start.
+ */
+ @Test
+ public void testSkipHeaderMultiByteWithRecordDelimiter() throws Exception {
+ // 'é' is two UTF-8 bytes (0xC3 0xA9), so the '~' delimiter sits at byte
offset 7
+ // ("h", "é" [2 bytes], "a", "d", "e", "r", "~") but at char index 6 under
UTF-8.
+ Path file = new Path(dataDir, "multibyte_delim.csv");
+ writeUtf8File(file, "héader~data_row1~data_row2");
+
+ job.set("textinputformat.record.delimiter", "~");
+ SkippingTextInputFormat inputFormat = new SkippingTextInputFormat();
+ inputFormat.configure(job, 1, 0); // skip one header "line", no footer
+ FileInputFormat.setInputPaths(job, file);
+ InputSplit[] splits = inputFormat.getSplits(job, 1);
+
+ assertEquals(1, splits.length);
+ assertTrue(splits[0] instanceof FileSplit);
+ // Byte offset of the delimiter; char-based UTF-8 decoding would wrongly
yield 6.
+ assertEquals(7, ((FileSplit) splits[0]).getStart());
+ }
+
+ /**
+ * 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.
*/
@@ -203,4 +360,15 @@ private void writeTextFile(Path file, String content)
throws IOException {
writer.write(content);
writer.close();
}
+
+ /**
+ * Writes the given string to the given file as UTF-8, so multi-byte
characters
+ * land as a deterministic byte sequence regardless of the platform default
charset.
+ */
+ private void writeUtf8File(Path file, String content) throws IOException {
+ try (OutputStreamWriter writer =
+ new OutputStreamWriter(fileSystem.create(file),
StandardCharsets.UTF_8)) {
+ writer.write(content);
+ }
+ }
}
