LineReader implements Closeable { public void close() throws IOException { in.close(); } LineReader(InputStream in);  LineReader(InputStream in, int bufferSize);  LineReader(InputStream in, Configuration conf);  LineReader(InputStream in, byte[] recordDelimiterBytes);  LineReader(InputStream in, int bufferSize,
                    byte[] recordDelimiterBytes);  LineReader(InputStream in, Configuration conf,
                    byte[] recordDelimiterBytes); void close(); void reset(); int readLine(Text str, int maxLineLength,
                      int maxBytesToConsume); int readDefaultLine(NonSyncByteArrayOutputStream str, ArrayList<Integer> offsets, int maxLineLength
      , int maxBytesToConsume); int readLine(Text str, int maxLineLength); int readLine(Text str);  }
@Test public void testSeekableByteBufLineReader() throws IOException { TajoConf conf = new TajoConf(); Path testDir = CommonTestingUtil.getTestDir(TEST_PATH); FileSystem fs = testDir.getFileSystem(conf); Schema schema = new Schema(); schema.addColumn("id", Type.INT4); schema.addColumn("age", Type.INT8); schema.addColumn("comment", Type.TEXT); schema.addColumn("comment2", Type.TEXT); TableMeta meta = CatalogUtil.newTableMeta("TEXT"); Path tablePath = new Path(testDir, "testSeekableByteBufLineReader.data"); FileAppender appender = (FileAppender) StorageManager.getFileStorageManager(conf).getAppender( null, null, meta, schema, tablePath); appender.enableStats(); appender.init(); int tupleNum = 10000; VTuple vTuple; for (int i = 0; i < tupleNum; i++) { vTuple = new VTuple(4); vTuple.put(0, DatumFactory.createInt4(i + 1)); vTuple.put(1, DatumFactory.createInt8(25l)); vTuple.put(2, DatumFactory.createText("emiya muljomdao")); vTuple.put(3, NullDatum.get()); appender.addTuple(vTuple); } appender.close(); FileStatus status = fs.getFileStatus(tablePath); AtomicInteger bytes = new AtomicInteger(); InputChannel channel = new FSDataInputChannel(fs.open(tablePath)); ByteBufLineReader reader = new ByteBufLineReader(channel); reader.seek(status.getLen()); assertNull(reader.readLineBuf(bytes)); assertEquals(0, bytes.get()); reader.seek(0); long totalRead = 0; int i = 0; for(;;){ ByteBuf buf = reader.readLineBuf(bytes); totalRead += bytes.get(); if(buf == null) break; i++; } IOUtils.cleanup(null, reader, channel, fs); assertEquals(tupleNum, i); assertEquals(status.getLen(), totalRead); assertEquals(status.getLen(), reader.readBytes()); }
@Test public void testByteBufLineReader() throws IOException { TajoConf conf = new TajoConf(); Path testDir = CommonTestingUtil.getTestDir(TEST_PATH); FileSystem fs = testDir.getFileSystem(conf); Schema schema = new Schema(); schema.addColumn("id", Type.INT4); schema.addColumn("age", Type.INT8); schema.addColumn("comment", Type.TEXT); schema.addColumn("comment2", Type.TEXT); TableMeta meta = CatalogUtil.newTableMeta("TEXT"); Path tablePath = new Path(testDir, "line.data"); FileAppender appender = (FileAppender) StorageManager.getFileStorageManager(conf).getAppender( null, null, meta, schema, tablePath); appender.enableStats(); appender.init(); int tupleNum = 10000; VTuple vTuple; for (int i = 0; i < tupleNum; i++) { vTuple = new VTuple(4); vTuple.put(0, DatumFactory.createInt4(i + 1)); vTuple.put(1, DatumFactory.createInt8(25l)); vTuple.put(2, DatumFactory.createText("emiya muljomdao")); vTuple.put(3, NullDatum.get()); appender.addTuple(vTuple); } appender.close(); FileStatus status = fs.getFileStatus(tablePath); ByteBufInputChannel channel = new ByteBufInputChannel(fs.open(tablePath)); ByteBufLineReader reader = new ByteBufLineReader(channel); long totalRead = 0; int i = 0; AtomicInteger bytes = new AtomicInteger(); for(;;){ ByteBuf buf = reader.readLineBuf(bytes); totalRead += bytes.get(); if(buf == null) break; i++; } IOUtils.cleanup(null, reader, channel, fs); assertEquals(tupleNum, i); assertEquals(status.getLen(), totalRead); assertEquals(status.getLen(), reader.readBytes()); }