You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2014/12/10 15:08:34 UTC
[4/5] tajo git commit: TAJO-1235: ByteBufLineReader can not read text
line with CRLF. (jinho)
TAJO-1235: ByteBufLineReader can not read text line with CRLF. (jinho)
Closes #289
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/bebc7801
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/bebc7801
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/bebc7801
Branch: refs/heads/index_support
Commit: bebc78011798cd2fd691e8901141e94f1d445d6e
Parents: 72256fc
Author: jhkim <jh...@apache.org>
Authored: Wed Dec 10 11:25:03 2014 +0900
Committer: jhkim <jh...@apache.org>
Committed: Wed Dec 10 11:25:03 2014 +0900
----------------------------------------------------------------------
CHANGES | 3 +
.../tajo/storage/ByteBufInputChannel.java | 4 --
.../tajo/storage/text/ByteBufLineReader.java | 68 ++++++++++++--------
.../tajo/storage/text/DelimitedLineReader.java | 7 +-
.../org/apache/tajo/storage/TestLineReader.java | 43 ++++++++++---
5 files changed, 81 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/bebc7801/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index b2985ba..710f463 100644
--- a/CHANGES
+++ b/CHANGES
@@ -90,6 +90,9 @@ Release 0.9.1 - unreleased
BUG FIXES
+ TAJO-1235: ByteBufLineReader can not read text line with CRLF.
+ (jinho)
+
TAJO-1237: Fix missing maven-module for pullserver. (jinho)
TAJO-1196: Unit test hangs occasionally and randomly. (jihoon)
http://git-wip-us.apache.org/repos/asf/tajo/blob/bebc7801/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java b/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java
index b1b6d65..45fb1d8 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java
@@ -69,8 +69,4 @@ public class ByteBufInputChannel extends AbstractInterruptibleChannel implements
protected void implCloseChannel() throws IOException {
IOUtils.cleanup(null, channel, inputStream);
}
-
- public int available() throws IOException {
- return inputStream.available();
- }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/bebc7801/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java
index 86319e1..2f742c6 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java
@@ -32,10 +32,11 @@ public class ByteBufLineReader implements Closeable {
private int bufferSize;
private long readBytes;
+ private int startIndex;
private boolean eof = false;
private ByteBuf buffer;
private final ByteBufInputChannel channel;
- private final AtomicInteger tempReadBytes = new AtomicInteger();
+ private final AtomicInteger lineReadBytes = new AtomicInteger();
private final LineSplitProcessor processor = new LineSplitProcessor();
public ByteBufLineReader(ByteBufInputChannel channel) {
@@ -53,10 +54,6 @@ public class ByteBufLineReader implements Closeable {
return readBytes - buffer.readableBytes();
}
- public long available() throws IOException {
- return channel.available() + buffer.readableBytes();
- }
-
@Override
public void close() throws IOException {
if (this.buffer.refCnt() > 0) {
@@ -66,7 +63,7 @@ public class ByteBufLineReader implements Closeable {
}
public String readLine() throws IOException {
- ByteBuf buf = readLineBuf(tempReadBytes);
+ ByteBuf buf = readLineBuf(lineReadBytes);
if (buf != null) {
return buf.toString(CharsetUtil.UTF_8);
}
@@ -77,24 +74,26 @@ public class ByteBufLineReader implements Closeable {
int tailBytes = 0;
if (this.readBytes > 0) {
+ //startIndex = 0, readIndex = tailBytes length, writable = (buffer capacity - tailBytes)
this.buffer.markReaderIndex();
- this.buffer.discardSomeReadBytes(); // compact the buffer
+ this.buffer.discardReadBytes(); // compact the buffer
tailBytes = this.buffer.writerIndex();
if (!this.buffer.isWritable()) {
// a line bytes is large than the buffer
- BufferPool.ensureWritable(buffer, bufferSize);
+ BufferPool.ensureWritable(buffer, bufferSize * 2);
this.bufferSize = buffer.capacity();
}
+ this.startIndex = 0;
}
boolean release = true;
try {
int readBytes = tailBytes;
for (; ; ) {
- int localReadBytes = buffer.writeBytes(channel, bufferSize - readBytes);
+ int localReadBytes = buffer.writeBytes(channel, this.bufferSize - readBytes);
if (localReadBytes < 0) {
- if (tailBytes == readBytes) {
- // no more bytes are in the channel
+ if (buffer.isWritable()) {
+ //if read bytes is less than the buffer capacity, there is no more bytes in the channel
eof = true;
}
break;
@@ -106,9 +105,8 @@ public class ByteBufLineReader implements Closeable {
}
this.readBytes += (readBytes - tailBytes);
release = false;
- if (!eof) {
- this.buffer.readerIndex(this.buffer.readerIndex() + tailBytes); //skip past buffer (tail)
- }
+
+ this.buffer.readerIndex(this.buffer.readerIndex() + tailBytes); //skip past buffer (tail)
} finally {
if (release) {
buffer.release();
@@ -120,24 +118,36 @@ public class ByteBufLineReader implements Closeable {
* Read a line terminated by one of CR, LF, or CRLF.
*/
public ByteBuf readLineBuf(AtomicInteger reads) throws IOException {
- if(eof) return null;
-
- int startIndex = buffer.readerIndex();
- int readBytes;
+ int readBytes = 0; // newline + text line bytes
+ int newlineLength = 0; //length of terminating newline
int readable;
- int newlineLength; //length of terminating newline
+
+ this.startIndex = buffer.readerIndex();
loop:
while (true) {
readable = buffer.readableBytes();
if (readable <= 0) {
- buffer.readerIndex(startIndex);
+ buffer.readerIndex(this.startIndex);
fillBuffer(); //compact and fill buffer
- if (!buffer.isReadable()) {
+
+ //if buffer.writerIndex() is zero, there is no bytes in buffer
+ if (!buffer.isReadable() && buffer.writerIndex() == 0) {
+ reads.set(0);
return null;
} else {
- if (!eof) startIndex = 0; // reset the line start position
- else startIndex = buffer.readerIndex();
+ //skip first newLine
+ if (processor.isPrevCharCR() && buffer.getByte(buffer.readerIndex()) == LineSplitProcessor.LF) {
+ buffer.skipBytes(1);
+ if(eof && !buffer.isReadable()) {
+ reads.set(1);
+ return null;
+ }
+
+ newlineLength++;
+ readBytes++;
+ startIndex = buffer.readerIndex();
+ }
}
readable = buffer.readableBytes();
}
@@ -147,19 +157,21 @@ public class ByteBufLineReader implements Closeable {
//does not appeared terminating newline
buffer.readerIndex(buffer.writerIndex()); // set to end buffer
if(eof){
- readBytes = buffer.readerIndex() - startIndex;
- newlineLength = 0;
+ readBytes += (buffer.readerIndex() - startIndex);
break loop;
}
} else {
buffer.readerIndex(endIndex + 1);
- readBytes = buffer.readerIndex() - startIndex;
+ readBytes += (buffer.readerIndex() - startIndex); //past newline + text line
+
+ //appeared terminating CRLF
if (processor.isPrevCharCR() && buffer.isReadable()
&& buffer.getByte(buffer.readerIndex()) == LineSplitProcessor.LF) {
buffer.skipBytes(1);
- newlineLength = 2;
+ readBytes++;
+ newlineLength += 2;
} else {
- newlineLength = 1;
+ newlineLength += 1;
}
break loop;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/bebc7801/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java
index eb1929e..0efe030 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java
@@ -57,7 +57,7 @@ public class DelimitedLineReader implements Closeable {
private long startOffset, end, pos;
private boolean eof = true;
private ByteBufLineReader lineReader;
- private AtomicInteger tempReadBytes = new AtomicInteger();
+ private AtomicInteger lineReadBytes = new AtomicInteger();
private FileFragment fragment;
private Configuration conf;
@@ -122,11 +122,10 @@ public class DelimitedLineReader implements Closeable {
return null;
}
- ByteBuf buf = lineReader.readLineBuf(tempReadBytes);
+ ByteBuf buf = lineReader.readLineBuf(lineReadBytes);
+ pos += lineReadBytes.get();
if (buf == null) {
eof = true;
- } else {
- pos += tempReadBytes.get();
}
if (!isCompressed() && getCompressedPosition() > end) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/bebc7801/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java
index 4512d00..bfaba04 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java
@@ -19,6 +19,7 @@
package org.apache.tajo.storage;
import io.netty.buffer.ByteBuf;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -43,6 +44,7 @@ import org.junit.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
+import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.*;
@@ -84,18 +86,15 @@ public class TestLineReader {
FileStatus status = fs.getFileStatus(tablePath);
ByteBufInputChannel channel = new ByteBufInputChannel(fs.open(tablePath));
- assertEquals(status.getLen(), channel.available());
ByteBufLineReader reader = new ByteBufLineReader(channel);
- assertEquals(status.getLen(), reader.available());
long totalRead = 0;
int i = 0;
AtomicInteger bytes = new AtomicInteger();
for(;;){
ByteBuf buf = reader.readLineBuf(bytes);
- if(buf == null) break;
-
totalRead += bytes.get();
+ if(buf == null) break;
i++;
}
IOUtils.cleanup(null, reader, channel, fs);
@@ -171,18 +170,15 @@ public class TestLineReader {
String data = FileUtil.readTextFile(file);
ByteBufInputChannel channel = new ByteBufInputChannel(new FileInputStream(file));
-
- assertEquals(file.length(), channel.available());
ByteBufLineReader reader = new ByteBufLineReader(channel);
- assertEquals(file.length(), reader.available());
long totalRead = 0;
int i = 0;
AtomicInteger bytes = new AtomicInteger();
for(;;){
ByteBuf buf = reader.readLineBuf(bytes);
- if(buf == null) break;
totalRead += bytes.get();
+ if(buf == null) break;
i++;
}
IOUtils.cleanup(null, reader);
@@ -190,4 +186,35 @@ public class TestLineReader {
assertEquals(file.length(), reader.readBytes());
assertEquals(data.split("\n").length, i);
}
+
+ @Test
+ public void testCRLFLine() throws IOException {
+ TajoConf conf = new TajoConf();
+ Path testFile = new Path(CommonTestingUtil.getTestDir(TEST_PATH), "testCRLFLineText.txt");
+
+ FileSystem fs = testFile.getFileSystem(conf);
+ FSDataOutputStream outputStream = fs.create(testFile, true);
+ outputStream.write("0\r\n1\r\n".getBytes());
+ outputStream.flush();
+ IOUtils.closeStream(outputStream);
+
+ ByteBufInputChannel channel = new ByteBufInputChannel(fs.open(testFile));
+ ByteBufLineReader reader = new ByteBufLineReader(channel, BufferPool.directBuffer(2));
+ FileStatus status = fs.getFileStatus(testFile);
+
+ long totalRead = 0;
+ int i = 0;
+ AtomicInteger bytes = new AtomicInteger();
+ for(;;){
+ ByteBuf buf = reader.readLineBuf(bytes);
+ totalRead += bytes.get();
+ if(buf == null) break;
+ String row = buf.toString(Charset.defaultCharset());
+ assertEquals(i, Integer.parseInt(row));
+ i++;
+ }
+ IOUtils.cleanup(null, reader);
+ assertEquals(status.getLen(), totalRead);
+ assertEquals(status.getLen(), reader.readBytes());
+ }
}