You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by jh...@apache.org on 2014/11/10 08:46:52 UTC
[1/2] tajo git commit: TAJO-1149: Implement direct read of
DelimitedTextFile. (jinho)
Repository: tajo
Updated Branches:
refs/heads/master 5a72e2f62 -> 3e305b15f
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/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
new file mode 100644
index 0000000..eb1929e
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedLineReader.java
@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.hadoop.io.compress.SplittableCompressionCodec;
+import org.apache.tajo.common.exception.NotImplementedException;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.storage.ByteBufInputChannel;
+import org.apache.tajo.storage.FileScanner;
+import org.apache.tajo.storage.BufferPool;
+import org.apache.tajo.storage.compress.CodecPool;
+import org.apache.tajo.storage.fragment.FileFragment;
+
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class DelimitedLineReader implements Closeable {
+ private static final Log LOG = LogFactory.getLog(DelimitedLineReader.class);
+ private final static int DEFAULT_PAGE_SIZE = 128 * 1024;
+
+ private FileSystem fs;
+ private FSDataInputStream fis;
+ private InputStream is; //decompressd stream
+ private CompressionCodecFactory factory;
+ private CompressionCodec codec;
+ private Decompressor decompressor;
+
+ private long startOffset, end, pos;
+ private boolean eof = true;
+ private ByteBufLineReader lineReader;
+ private AtomicInteger tempReadBytes = new AtomicInteger();
+ private FileFragment fragment;
+ private Configuration conf;
+
+ public DelimitedLineReader(Configuration conf, final FileFragment fragment) throws IOException {
+ this.fragment = fragment;
+ this.conf = conf;
+ this.factory = new CompressionCodecFactory(conf);
+ this.codec = factory.getCodec(fragment.getPath());
+ if (this.codec instanceof SplittableCompressionCodec) {
+ throw new NotImplementedException(); // bzip2 does not support multi-thread model
+ }
+ }
+
+ public void init() throws IOException {
+ if (fs == null) {
+ fs = FileScanner.getFileSystem((TajoConf) conf, fragment.getPath());
+ }
+ if (fis == null) fis = fs.open(fragment.getPath());
+ pos = startOffset = fragment.getStartKey();
+ end = startOffset + fragment.getEndKey();
+
+ if (codec != null) {
+ decompressor = CodecPool.getDecompressor(codec);
+ is = new DataInputStream(codec.createInputStream(fis, decompressor));
+ ByteBufInputChannel channel = new ByteBufInputChannel(is);
+ lineReader = new ByteBufLineReader(channel, BufferPool.directBuffer(DEFAULT_PAGE_SIZE));
+ } else {
+ fis.seek(startOffset);
+ is = fis;
+
+ ByteBufInputChannel channel = new ByteBufInputChannel(is);
+ lineReader = new ByteBufLineReader(channel,
+ BufferPool.directBuffer((int) Math.min(DEFAULT_PAGE_SIZE, end)));
+ }
+ eof = false;
+ }
+
+ public long getCompressedPosition() throws IOException {
+ long retVal;
+ if (isCompressed()) {
+ retVal = fis.getPos();
+ } else {
+ retVal = pos;
+ }
+ return retVal;
+ }
+
+ public long getUnCompressedPosition() throws IOException {
+ return pos;
+ }
+
+ public long getReadBytes() {
+ return pos - startOffset;
+ }
+
+ public boolean isReadable() {
+ return !eof;
+ }
+
+ public ByteBuf readLine() throws IOException {
+ if (eof) {
+ return null;
+ }
+
+ ByteBuf buf = lineReader.readLineBuf(tempReadBytes);
+ if (buf == null) {
+ eof = true;
+ } else {
+ pos += tempReadBytes.get();
+ }
+
+ if (!isCompressed() && getCompressedPosition() > end) {
+ eof = true;
+ }
+ return buf;
+ }
+
+ public boolean isCompressed() {
+ return codec != null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ IOUtils.cleanup(LOG, lineReader, is, fis);
+ fs = null;
+ is = null;
+ fis = null;
+ lineReader = null;
+ } finally {
+ if (decompressor != null) {
+ CodecPool.returnDecompressor(decompressor);
+ decompressor = null;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
new file mode 100644
index 0000000..dbf8435
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
@@ -0,0 +1,483 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang.StringEscapeUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.CompressionOutputStream;
+import org.apache.hadoop.io.compress.Compressor;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.compress.CodecPool;
+import org.apache.tajo.storage.exception.AlreadyExistsStorageException;
+import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.rcfile.NonSyncByteArrayOutputStream;
+
+import java.io.BufferedOutputStream;
+import java.io.DataOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+
+public class DelimitedTextFile {
+
+ public static final byte LF = '\n';
+ public static int EOF = -1;
+
+ private static final Log LOG = LogFactory.getLog(DelimitedTextFile.class);
+
+ public static class DelimitedTextFileAppender extends FileAppender {
+ private final TableMeta meta;
+ private final Schema schema;
+ private final int columnNum;
+ private final FileSystem fs;
+ private FSDataOutputStream fos;
+ private DataOutputStream outputStream;
+ private CompressionOutputStream deflateFilter;
+ private char delimiter;
+ private TableStatistics stats = null;
+ private Compressor compressor;
+ private CompressionCodecFactory codecFactory;
+ private CompressionCodec codec;
+ private Path compressedPath;
+ private byte[] nullChars;
+ private int BUFFER_SIZE = 128 * 1024;
+ private int bufferedBytes = 0;
+ private long pos = 0;
+
+ private NonSyncByteArrayOutputStream os;
+ private FieldSerializerDeserializer serde;
+
+ public DelimitedTextFileAppender(Configuration conf, final Schema schema, final TableMeta meta, final Path path)
+ throws IOException {
+ super(conf, schema, meta, path);
+ this.fs = path.getFileSystem(conf);
+ this.meta = meta;
+ this.schema = schema;
+ this.delimiter = StringEscapeUtils.unescapeJava(this.meta.getOption(StorageConstants.TEXT_DELIMITER,
+ StorageConstants.DEFAULT_FIELD_DELIMITER)).charAt(0);
+ this.columnNum = schema.size();
+
+ String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(StorageConstants.TEXT_NULL,
+ NullDatum.DEFAULT_TEXT));
+ if (StringUtils.isEmpty(nullCharacters)) {
+ nullChars = NullDatum.get().asTextBytes();
+ } else {
+ nullChars = nullCharacters.getBytes();
+ }
+ }
+
+ @Override
+ public void init() throws IOException {
+ if (!fs.exists(path.getParent())) {
+ throw new FileNotFoundException(path.toString());
+ }
+
+ if (this.meta.containsOption(StorageConstants.COMPRESSION_CODEC)) {
+ String codecName = this.meta.getOption(StorageConstants.COMPRESSION_CODEC);
+ codecFactory = new CompressionCodecFactory(conf);
+ codec = codecFactory.getCodecByClassName(codecName);
+ compressor = CodecPool.getCompressor(codec);
+ if (compressor != null) compressor.reset(); //builtin gzip is null
+
+ String extension = codec.getDefaultExtension();
+ compressedPath = path.suffix(extension);
+
+ if (fs.exists(compressedPath)) {
+ throw new AlreadyExistsStorageException(compressedPath);
+ }
+
+ fos = fs.create(compressedPath);
+ deflateFilter = codec.createOutputStream(fos, compressor);
+ outputStream = new DataOutputStream(deflateFilter);
+
+ } else {
+ if (fs.exists(path)) {
+ throw new AlreadyExistsStorageException(path);
+ }
+ fos = fs.create(path);
+ outputStream = new DataOutputStream(new BufferedOutputStream(fos));
+ }
+
+ if (enabledStats) {
+ this.stats = new TableStatistics(this.schema);
+ }
+
+ try {
+ // we need to discuss the De/Serializer interface. so custom serde is to disable
+ String serdeClass = this.meta.getOption(StorageConstants.TEXTFILE_SERDE,
+ TextFieldSerializerDeserializer.class.getName());
+ serde = (TextFieldSerializerDeserializer) ReflectionUtils.newInstance(Class.forName(serdeClass), conf);
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e);
+ }
+
+ if (os == null) {
+ os = new NonSyncByteArrayOutputStream(BUFFER_SIZE);
+ }
+
+ os.reset();
+ pos = fos.getPos();
+ bufferedBytes = 0;
+ super.init();
+ }
+
+
+ @Override
+ public void addTuple(Tuple tuple) throws IOException {
+ Datum datum;
+ int rowBytes = 0;
+
+ for (int i = 0; i < columnNum; i++) {
+ datum = tuple.get(i);
+ rowBytes += serde.serialize(os, datum, schema.getColumn(i), i, nullChars);
+
+ if (columnNum - 1 > i) {
+ os.write((byte) delimiter);
+ rowBytes += 1;
+ }
+ }
+ os.write(LF);
+ rowBytes += 1;
+
+ pos += rowBytes;
+ bufferedBytes += rowBytes;
+ if (bufferedBytes > BUFFER_SIZE) {
+ flushBuffer();
+ }
+ // Statistical section
+ if (enabledStats) {
+ stats.incrementRow();
+ }
+ }
+
+ private void flushBuffer() throws IOException {
+ if (os.getLength() > 0) {
+ os.writeTo(outputStream);
+ os.reset();
+ bufferedBytes = 0;
+ }
+ }
+
+ @Override
+ public long getOffset() throws IOException {
+ return pos;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushBuffer();
+ outputStream.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ try {
+ if(outputStream != null){
+ flush();
+ }
+
+ // Statistical section
+ if (enabledStats) {
+ stats.setNumBytes(getOffset());
+ }
+
+ if (deflateFilter != null) {
+ deflateFilter.finish();
+ deflateFilter.resetState();
+ deflateFilter = null;
+ }
+
+ os.close();
+ } finally {
+ IOUtils.cleanup(LOG, fos);
+ if (compressor != null) {
+ CodecPool.returnCompressor(compressor);
+ compressor = null;
+ }
+ }
+ }
+
+ @Override
+ public TableStats getStats() {
+ if (enabledStats) {
+ return stats.getTableStat();
+ } else {
+ return null;
+ }
+ }
+
+ public boolean isCompress() {
+ return compressor != null;
+ }
+
+ public String getExtension() {
+ return codec != null ? codec.getDefaultExtension() : "";
+ }
+ }
+
+ public static class DelimitedTextFileScanner extends FileScanner {
+
+ private boolean splittable = false;
+ private final long startOffset;
+ private final long endOffset;
+
+ private int recordCount = 0;
+ private int[] targetColumnIndexes;
+
+ private ByteBuf nullChars;
+ private FieldSerializerDeserializer serde;
+ private DelimitedLineReader reader;
+ private FieldSplitProcessor processor;
+
+ public DelimitedTextFileScanner(Configuration conf, final Schema schema, final TableMeta meta,
+ final FileFragment fragment)
+ throws IOException {
+ super(conf, schema, meta, fragment);
+ reader = new DelimitedLineReader(conf, fragment);
+ if (!reader.isCompressed()) {
+ splittable = true;
+ }
+
+ startOffset = fragment.getStartKey();
+ endOffset = startOffset + fragment.getEndKey();
+
+ //Delimiter
+ String delim = meta.getOption(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ this.processor = new FieldSplitProcessor(StringEscapeUtils.unescapeJava(delim).charAt(0));
+ }
+
+ @Override
+ public void init() throws IOException {
+ if (nullChars != null) {
+ nullChars.release();
+ }
+
+ String nullCharacters = StringEscapeUtils.unescapeJava(meta.getOption(StorageConstants.TEXT_NULL,
+ NullDatum.DEFAULT_TEXT));
+ byte[] bytes;
+ if (StringUtils.isEmpty(nullCharacters)) {
+ bytes = NullDatum.get().asTextBytes();
+ } else {
+ bytes = nullCharacters.getBytes();
+ }
+
+ nullChars = BufferPool.directBuffer(bytes.length, bytes.length);
+ nullChars.writeBytes(bytes);
+
+ if (reader != null) {
+ reader.close();
+ }
+ reader = new DelimitedLineReader(conf, fragment);
+ reader.init();
+ recordCount = 0;
+
+ if (targets == null) {
+ targets = schema.toArray();
+ }
+
+ targetColumnIndexes = new int[targets.length];
+ for (int i = 0; i < targets.length; i++) {
+ targetColumnIndexes[i] = schema.getColumnId(targets[i].getQualifiedName());
+ }
+
+ try {
+ // we need to discuss the De/Serializer interface. so custom serde is to disable
+ String serdeClass = this.meta.getOption(StorageConstants.TEXTFILE_SERDE,
+ TextFieldSerializerDeserializer.class.getName());
+ serde = (TextFieldSerializerDeserializer) ReflectionUtils.newInstance(Class.forName(serdeClass), conf);
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e);
+ }
+
+ super.init();
+ Arrays.sort(targetColumnIndexes);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DelimitedTextFileScanner open:" + fragment.getPath() + "," + startOffset + "," + endOffset);
+ }
+
+ if (startOffset > 0) {
+ reader.readLine(); // skip first line;
+ }
+ }
+
+ public ByteBuf readLine() throws IOException {
+ ByteBuf buf = reader.readLine();
+ if (buf == null) {
+ return null;
+ } else {
+ recordCount++;
+ }
+
+ return buf;
+ }
+
+ @Override
+ public float getProgress() {
+ try {
+ if (!reader.isReadable()) {
+ return 1.0f;
+ }
+ long filePos = reader.getCompressedPosition();
+ if (startOffset == filePos) {
+ return 0.0f;
+ } else {
+ long readBytes = filePos - startOffset;
+ long remainingBytes = Math.max(endOffset - filePos, 0);
+ return Math.min(1.0f, (float) (readBytes) / (float) (readBytes + remainingBytes));
+ }
+ } catch (IOException e) {
+ LOG.error(e.getMessage(), e);
+ return 0.0f;
+ }
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ try {
+ if (!reader.isReadable()) return null;
+
+ ByteBuf buf = readLine();
+ if (buf == null) return null;
+
+ if (targets.length == 0) {
+ return EmptyTuple.get();
+ }
+
+ VTuple tuple = new VTuple(schema.size());
+ fillTuple(schema, tuple, buf, targetColumnIndexes);
+ return tuple;
+ } catch (Throwable t) {
+ LOG.error("Tuple list current index: " + recordCount + " file offset:" + reader.getCompressedPosition(), t);
+ throw new IOException(t);
+ }
+ }
+
+ private void fillTuple(Schema schema, Tuple dst, ByteBuf lineBuf, int[] target) throws IOException {
+ int[] projection = target;
+ if (lineBuf == null || target == null || target.length == 0) {
+ return;
+ }
+
+ final int rowLength = lineBuf.readableBytes();
+ int start = 0, fieldLength = 0, end = 0;
+
+ //Projection
+ int currentTarget = 0;
+ int currentIndex = 0;
+
+ while (end != -1) {
+ end = lineBuf.forEachByte(start, rowLength - start, processor);
+
+ if (end < 0) {
+ fieldLength = rowLength - start;
+ } else {
+ fieldLength = end - start;
+ }
+
+ if (projection.length > currentTarget && currentIndex == projection[currentTarget]) {
+ Datum datum = serde.deserialize(lineBuf.slice(start, fieldLength),
+ schema.getColumn(currentIndex), currentIndex, nullChars);
+ dst.put(currentIndex, datum);
+ currentTarget++;
+ }
+
+ if (projection.length == currentTarget) {
+ break;
+ }
+
+ start = end + 1;
+ currentIndex++;
+ }
+ }
+
+ @Override
+ public void reset() throws IOException {
+ init();
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ if (nullChars != null) {
+ nullChars.release();
+ nullChars = null;
+ }
+
+ if (tableStats != null && reader != null) {
+ tableStats.setReadBytes(reader.getReadBytes()); //Actual Processed Bytes. (decompressed bytes + overhead)
+ tableStats.setNumRows(recordCount);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DelimitedTextFileScanner processed record:" + recordCount);
+ }
+ } finally {
+ IOUtils.cleanup(LOG, reader);
+ reader = null;
+ }
+ }
+
+ @Override
+ public boolean isProjectable() {
+ return true;
+ }
+
+ @Override
+ public boolean isSelectable() {
+ return false;
+ }
+
+ @Override
+ public void setSearchCondition(Object expr) {
+ }
+
+ @Override
+ public boolean isSplittable() {
+ return splittable;
+ }
+
+ @Override
+ public TableStats getInputStats() {
+ if (tableStats != null && reader != null) {
+ tableStats.setReadBytes(reader.getReadBytes()); //Actual Processed Bytes. (decompressed bytes + overhead)
+ tableStats.setNumRows(recordCount);
+ tableStats.setNumBytes(fragment.getEndKey());
+ }
+ return tableStats;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/text/FieldSplitProcessor.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/FieldSplitProcessor.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/FieldSplitProcessor.java
new file mode 100644
index 0000000..a5ac142
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/FieldSplitProcessor.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import io.netty.buffer.ByteBufProcessor;
+
+public class FieldSplitProcessor implements ByteBufProcessor {
+ private char delimiter; //the ascii separate character
+
+ public FieldSplitProcessor(char recordDelimiterByte) {
+ this.delimiter = recordDelimiterByte;
+ }
+
+ @Override
+ public boolean process(byte value) throws Exception {
+ return delimiter != value;
+ }
+
+ public char getDelimiter() {
+ return delimiter;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/text/LineSplitProcessor.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/LineSplitProcessor.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/LineSplitProcessor.java
new file mode 100644
index 0000000..a130527
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/LineSplitProcessor.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import io.netty.buffer.ByteBufProcessor;
+
+public class LineSplitProcessor implements ByteBufProcessor {
+ public static final byte CR = '\r';
+ public static final byte LF = '\n';
+ private boolean prevCharCR = false; //true of prev char was CR
+
+ @Override
+ public boolean process(byte value) throws Exception {
+ switch (value) {
+ case LF:
+ return false;
+ case CR:
+ prevCharCR = true;
+ return false;
+ default:
+ prevCharCR = false;
+ return true;
+ }
+ }
+
+ public boolean isPrevCharCR() {
+ return prevCharCR;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/text/TextFieldSerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/text/TextFieldSerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/text/TextFieldSerializerDeserializer.java
new file mode 100644
index 0000000..0057b54
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/TextFieldSerializerDeserializer.java
@@ -0,0 +1,227 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import com.google.protobuf.Message;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.common.TajoDataTypes;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.*;
+import org.apache.tajo.datum.protobuf.ProtobufJsonFormat;
+import org.apache.tajo.storage.FieldSerializerDeserializer;
+import org.apache.tajo.util.NumberUtil;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+//Compatibility with Apache Hive
+public class TextFieldSerializerDeserializer implements FieldSerializerDeserializer {
+ public static final byte[] trueBytes = "true".getBytes();
+ public static final byte[] falseBytes = "false".getBytes();
+ private ProtobufJsonFormat protobufJsonFormat = ProtobufJsonFormat.getInstance();
+
+ private static boolean isNull(ByteBuf val, ByteBuf nullBytes) {
+ return !val.isReadable() || nullBytes.equals(val);
+ }
+
+ private static boolean isNullText(ByteBuf val, ByteBuf nullBytes) {
+ return val.readableBytes() > 0 && nullBytes.equals(val);
+ }
+
+ @Override
+ public int serialize(OutputStream out, Datum datum, Column col, int columnIndex, byte[] nullChars) throws IOException {
+ byte[] bytes;
+ int length = 0;
+ TajoDataTypes.DataType dataType = col.getDataType();
+
+ if (datum == null || datum instanceof NullDatum) {
+ switch (dataType.getType()) {
+ case CHAR:
+ case TEXT:
+ length = nullChars.length;
+ out.write(nullChars);
+ break;
+ default:
+ break;
+ }
+ return length;
+ }
+
+ switch (dataType.getType()) {
+ case BOOLEAN:
+ out.write(datum.asBool() ? trueBytes : falseBytes);
+ length = trueBytes.length;
+ break;
+ case CHAR:
+ byte[] pad = new byte[dataType.getLength() - datum.size()];
+ bytes = datum.asTextBytes();
+ out.write(bytes);
+ out.write(pad);
+ length = bytes.length + pad.length;
+ break;
+ case TEXT:
+ case BIT:
+ case INT2:
+ case INT4:
+ case INT8:
+ case FLOAT4:
+ case FLOAT8:
+ case INET4:
+ case DATE:
+ case INTERVAL:
+ bytes = datum.asTextBytes();
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case TIME:
+ bytes = ((TimeDatum) datum).asChars(TajoConf.getCurrentTimeZone(), true).getBytes();
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case TIMESTAMP:
+ bytes = ((TimestampDatum) datum).asChars(TajoConf.getCurrentTimeZone(), true).getBytes();
+ length = bytes.length;
+ out.write(bytes);
+ break;
+ case INET6:
+ case BLOB:
+ bytes = Base64.encodeBase64(datum.asByteArray(), false);
+ length = bytes.length;
+ out.write(bytes, 0, length);
+ break;
+ case PROTOBUF:
+ ProtobufDatum protobuf = (ProtobufDatum) datum;
+ byte[] protoBytes = protobufJsonFormat.printToString(protobuf.get()).getBytes();
+ length = protoBytes.length;
+ out.write(protoBytes, 0, protoBytes.length);
+ break;
+ case NULL_TYPE:
+ default:
+ break;
+ }
+ return length;
+ }
+
+ @Override
+ public Datum deserialize(ByteBuf buf, Column col, int columnIndex, ByteBuf nullChars) throws IOException{
+ Datum datum;
+ TajoDataTypes.Type type = col.getDataType().getType();
+ boolean nullField;
+ if (type == TajoDataTypes.Type.TEXT || type == TajoDataTypes.Type.CHAR) {
+ nullField = isNullText(buf, nullChars);
+ } else {
+ nullField = isNull(buf, nullChars);
+ }
+
+ if (nullField) {
+ datum = NullDatum.get();
+ } else {
+ switch (type) {
+ case BOOLEAN:
+ byte bool = buf.readByte();
+ datum = DatumFactory.createBool(bool == 't' || bool == 'T');
+ break;
+ case BIT:
+ datum = DatumFactory.createBit(Byte.parseByte(buf.toString(CharsetUtil.UTF_8)));
+ break;
+ case CHAR:
+ datum = DatumFactory.createChar(buf.toString(CharsetUtil.UTF_8).trim());
+ break;
+ case INT1:
+ case INT2: {
+ //TODO zero-copy
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createInt2((short) NumberUtil.parseInt(bytes, 0, bytes.length));
+ break;
+ }
+ case INT4: {
+ //TODO zero-copy
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createInt4(NumberUtil.parseInt(bytes, 0, bytes.length));
+ break;
+ }
+ case INT8:
+ //TODO zero-copy
+ datum = DatumFactory.createInt8(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case FLOAT4:
+ //TODO zero-copy
+ datum = DatumFactory.createFloat4(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case FLOAT8: {
+ //TODO zero-copy
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createFloat8(NumberUtil.parseDouble(bytes, 0, bytes.length));
+ break;
+ }
+ case TEXT: {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createText(bytes);
+ break;
+ }
+ case DATE:
+ datum = DatumFactory.createDate(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case TIME:
+ datum = DatumFactory.createTime(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case TIMESTAMP:
+ datum = DatumFactory.createTimestamp(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case INTERVAL:
+ datum = DatumFactory.createInterval(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case PROTOBUF: {
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(col.getDataType());
+ Message.Builder builder = factory.newBuilder();
+ try {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ protobufJsonFormat.merge(bytes, builder);
+ datum = factory.createDatum(builder.build());
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ break;
+ }
+ case INET4:
+ datum = DatumFactory.createInet4(buf.toString(CharsetUtil.UTF_8));
+ break;
+ case BLOB: {
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ datum = DatumFactory.createBlob(Base64.decodeBase64(bytes));
+ break;
+ }
+ default:
+ datum = NullDatum.get();
+ break;
+ }
+ }
+ return datum;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/resources/storage-default.xml
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/resources/storage-default.xml b/tajo-storage/src/main/resources/storage-default.xml
index 4669477..f262585 100644
--- a/tajo-storage/src/main/resources/storage-default.xml
+++ b/tajo-storage/src/main/resources/storage-default.xml
@@ -21,11 +21,6 @@
<configuration>
<property>
- <name>tajo.storage.manager.v2</name>
- <value>false</value>
- </property>
-
- <property>
<name>tajo.storage.manager.maxReadBytes</name>
<value>8388608</value>
<description></description>
@@ -40,11 +35,15 @@
<!--- Registered Scanner Handler -->
<property>
<name>tajo.storage.scanner-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
+ <value>textfile,csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
</property>
<!--- Fragment Class Configurations -->
<property>
+ <name>tajo.storage.fragment.textfile.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
<name>tajo.storage.fragment.csv.class</name>
<value>org.apache.tajo.storage.fragment.FileFragment</value>
</property>
@@ -79,13 +78,13 @@
<!--- Scanner Handler -->
<property>
- <name>tajo.storage.scanner-handler.csv.class</name>
- <value>org.apache.tajo.storage.CSVFile$CSVScanner</value>
+ <name>tajo.storage.scanner-handler.textfile.class</name>
+ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.csv.class</name>
- <value>org.apache.tajo.storage.v2.CSVFileScanner</value>
+ <name>tajo.storage.scanner-handler.csv.class</name>
+ <value>org.apache.tajo.storage.CSVFile$CSVScanner</value>
</property>
<property>
@@ -94,74 +93,44 @@
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.raw.class</name>
- <value>org.apache.tajo.storage.RawFile$RawFileScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.rcfile.class</name>
<value>org.apache.tajo.storage.rcfile.RCFile$RCFileScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.rcfile.class</name>
- <value>org.apache.tajo.storage.v2.RCFileScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.rowfile.class</name>
<value>org.apache.tajo.storage.RowFile$RowFileScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.rowfile.class</name>
- <value>org.apache.tajo.storage.RowFile$RowFileScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.trevni.class</name>
<value>org.apache.tajo.storage.trevni.TrevniScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.trevni.class</name>
- <value>org.apache.tajo.storage.trevni.TrevniScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.parquet.class</name>
<value>org.apache.tajo.storage.parquet.ParquetScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.parquet.class</name>
- <value>org.apache.tajo.storage.parquet.ParquetScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.sequencefile.class</name>
<value>org.apache.tajo.storage.sequencefile.SequenceFileScanner</value>
</property>
<property>
- <name>tajo.storage.scanner-handler.v2.sequencefile.class</name>
- <value>org.apache.tajo.storage.sequencefile.SequenceFileScanner</value>
- </property>
-
- <property>
<name>tajo.storage.scanner-handler.avro.class</name>
<value>org.apache.tajo.storage.avro.AvroScanner</value>
</property>
+ <!--- Appender Handler -->
<property>
- <name>tajo.storage.scanner-handler.v2.avro.class</name>
- <value>org.apache.tajo.storage.avro.AvroScanner</value>
+ <name>tajo.storage.appender-handler</name>
+ <value>textfile,csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
</property>
- <!--- Appender Handler -->
<property>
- <name>tajo.storage.appender-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
+ <name>tajo.storage.appender-handler.textfile.class</name>
+ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileAppender</value>
</property>
<property>
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/test/java/org/apache/tajo/storage/TestCompressionStorages.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestCompressionStorages.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestCompressionStorages.java
index 212f374..fd5a63e 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestCompressionStorages.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestCompressionStorages.java
@@ -37,6 +37,7 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.storage.sequencefile.SequenceFileScanner;
+import org.apache.tajo.storage.text.DelimitedTextFile;
import org.apache.tajo.util.CommonTestingUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -70,7 +71,8 @@ public class TestCompressionStorages {
return Arrays.asList(new Object[][]{
{StoreType.CSV},
{StoreType.RCFILE},
- {StoreType.SEQUENCEFILE}
+ {StoreType.SEQUENCEFILE},
+ {StoreType.TEXTFILE}
});
}
@@ -102,81 +104,11 @@ public class TestCompressionStorages {
}
@Test
- public void testBzip2CodecCompressionData() throws IOException {
- storageCompressionTest(storeType, BZip2Codec.class);
- }
-
- @Test
public void testLz4CodecCompressionData() throws IOException {
if(NativeCodeLoader.isNativeCodeLoaded() && Lz4Codec.isNativeCodeLoaded())
storageCompressionTest(storeType, Lz4Codec.class);
}
- @Test
- public void testSplitCompressionData() throws IOException {
- if(StoreType.CSV != storeType) return;
-
- Schema schema = new Schema();
- schema.addColumn("id", Type.INT4);
- schema.addColumn("age", Type.INT8);
-
- TableMeta meta = CatalogUtil.newTableMeta(storeType);
- meta.putOption("compression.codec", BZip2Codec.class.getCanonicalName());
-
- Path tablePath = new Path(testDir, "SplitCompression");
- Appender appender = StorageManager.getStorageManager(conf).getAppender(meta, schema, tablePath);
- appender.enableStats();
- appender.init();
-
- String extention = "";
- if (appender instanceof CSVFile.CSVAppender) {
- extention = ((CSVFile.CSVAppender) appender).getExtension();
- }
-
- int tupleNum = 100000;
- VTuple vTuple;
-
- for (int i = 0; i < tupleNum; i++) {
- vTuple = new VTuple(2);
- vTuple.put(0, DatumFactory.createInt4(i + 1));
- vTuple.put(1, DatumFactory.createInt8(25l));
- appender.addTuple(vTuple);
- }
- appender.close();
-
- TableStats stat = appender.getStats();
- assertEquals(tupleNum, stat.getNumRows().longValue());
- tablePath = tablePath.suffix(extention);
-
- FileStatus status = fs.getFileStatus(tablePath);
- long fileLen = status.getLen();
- long randomNum = (long) (Math.random() * fileLen) + 1;
-
- FileFragment[] tablets = new FileFragment[2];
- tablets[0] = new FileFragment("SplitCompression", tablePath, 0, randomNum);
- tablets[1] = new FileFragment("SplitCompression", tablePath, randomNum, (fileLen - randomNum));
-
- Scanner scanner = StorageManager.getStorageManager(conf).getScanner(meta, schema, tablets[0], schema);
- assertTrue(scanner.isSplittable());
- scanner.init();
- int tupleCnt = 0;
- Tuple tuple;
- while ((tuple = scanner.next()) != null) {
- tupleCnt++;
- }
- scanner.close();
-
- scanner = StorageManager.getStorageManager(conf).getScanner(meta, schema, tablets[1], schema);
- assertTrue(scanner.isSplittable());
- scanner.init();
- while ((tuple = scanner.next()) != null) {
- tupleCnt++;
- }
-
- scanner.close();
- assertEquals(tupleNum, tupleCnt);
- }
-
private void storageCompressionTest(StoreType storeType, Class<? extends CompressionCodec> codec) throws IOException {
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
@@ -199,6 +131,8 @@ public class TestCompressionStorages {
String extension = "";
if (appender instanceof CSVFile.CSVAppender) {
extension = ((CSVFile.CSVAppender) appender).getExtension();
+ } else if (appender instanceof DelimitedTextFile.DelimitedTextFileAppender) {
+ extension = ((DelimitedTextFile.DelimitedTextFileAppender) appender).getExtension();
}
int tupleNum = 100000;
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/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
new file mode 100644
index 0000000..ef6efdf
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestLineReader.java
@@ -0,0 +1,163 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.compress.DeflateCodec;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.common.TajoDataTypes.Type;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.storage.text.ByteBufLineReader;
+import org.apache.tajo.storage.text.DelimitedTextFile;
+import org.apache.tajo.storage.text.DelimitedLineReader;
+import org.apache.tajo.util.CommonTestingUtil;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.*;
+
+public class TestLineReader {
+ private static String TEST_PATH = "target/test-data/TestLineReader";
+
+ @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(StoreType.TEXTFILE);
+ Path tablePath = new Path(testDir, "line.data");
+ FileAppender appender = (FileAppender) StorageManager.getStorageManager(conf).getAppender(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));
+ 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();
+ i++;
+ }
+ IOUtils.cleanup(null, reader, channel, fs);
+ assertEquals(tupleNum, i);
+ assertEquals(status.getLen(), totalRead);
+ assertEquals(status.getLen(), reader.readBytes());
+ }
+
+ @Test
+ public void testLineDelimitedReader() 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(StoreType.TEXTFILE);
+ meta.putOption("compression.codec", DeflateCodec.class.getCanonicalName());
+
+ Path tablePath = new Path(testDir, "line1." + DeflateCodec.class.getSimpleName());
+ FileAppender appender = (FileAppender) StorageManager.getStorageManager(conf).getAppender(meta, schema,
+ tablePath);
+ appender.enableStats();
+ appender.init();
+ int tupleNum = 10000;
+ VTuple vTuple;
+
+ long splitOffset = 0;
+ 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);
+
+ if(i == (tupleNum / 2)){
+ splitOffset = appender.getOffset();
+ }
+ }
+ String extension = ((DelimitedTextFile.DelimitedTextFileAppender) appender).getExtension();
+ appender.close();
+
+ tablePath = tablePath.suffix(extension);
+ FileFragment fragment = new FileFragment("table", tablePath, 0, splitOffset);
+ DelimitedLineReader reader = new DelimitedLineReader(conf, fragment); // if file is compressed, will read to EOF
+ assertTrue(reader.isCompressed());
+ assertFalse(reader.isReadable());
+ reader.init();
+ assertTrue(reader.isReadable());
+
+
+ int i = 0;
+ while(reader.isReadable()){
+ ByteBuf buf = reader.readLine();
+ if(buf == null) break;
+ i++;
+ }
+
+ IOUtils.cleanup(null, reader, fs);
+ assertEquals(tupleNum, i);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/test/java/org/apache/tajo/storage/TestSplitProcessor.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestSplitProcessor.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestSplitProcessor.java
new file mode 100644
index 0000000..12ea551
--- /dev/null
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestSplitProcessor.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.util.CharsetUtil;
+import org.apache.tajo.storage.text.FieldSplitProcessor;
+import org.apache.tajo.storage.text.LineSplitProcessor;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static io.netty.util.ReferenceCountUtil.releaseLater;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestSplitProcessor {
+
+ @Test
+ public void testFieldSplitProcessor() throws IOException {
+ String data = "abc||de";
+ final ByteBuf buf = releaseLater(
+ Unpooled.copiedBuffer(data, CharsetUtil.ISO_8859_1));
+
+ final int len = buf.readableBytes();
+ FieldSplitProcessor processor = new FieldSplitProcessor('|');
+
+ assertEquals(3, buf.forEachByte(0, len, processor));
+ assertEquals(4, buf.forEachByte(4, len - 4, processor));
+ assertEquals(-1, buf.forEachByte(5, len - 5, processor));
+
+ }
+
+ @Test
+ public void testLineSplitProcessor() throws IOException {
+ String data = "abc\r\n\n";
+ final ByteBuf buf = releaseLater(
+ Unpooled.copiedBuffer(data, CharsetUtil.ISO_8859_1));
+
+ final int len = buf.readableBytes();
+ LineSplitProcessor processor = new LineSplitProcessor();
+
+ //find CR
+ assertEquals(3, buf.forEachByte(0, len, processor));
+
+ // find CRLF
+ assertEquals(4, buf.forEachByte(4, len - 4, processor));
+ assertEquals(buf.getByte(4), '\n');
+ // need to skip LF
+ assertTrue(processor.isPrevCharCR());
+
+ // find LF
+ assertEquals(5, buf.forEachByte(5, len - 5, processor)); //line length is zero
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
index dca21af..56cef77 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
@@ -121,12 +121,14 @@ public class TestStorages {
@Parameterized.Parameters
public static Collection<Object[]> generateParameters() {
return Arrays.asList(new Object[][] {
+ //type, splitable, statsable, seekable
{StoreType.CSV, true, true, true},
{StoreType.RAW, false, false, true},
{StoreType.RCFILE, true, true, false},
{StoreType.PARQUET, false, false, false},
{StoreType.SEQUENCEFILE, true, true, false},
{StoreType.AVRO, false, false, false},
+ {StoreType.TEXTFILE, true, true, false},
});
}
@@ -381,7 +383,7 @@ public class TestStorages {
KeyValueSet options = new KeyValueSet();
TableMeta meta = CatalogUtil.newTableMeta(storeType, options);
meta.setOptions(CatalogUtil.newPhysicalProperties(storeType));
- meta.putOption(StorageConstants.CSVFILE_NULL, "\\\\N");
+ meta.putOption(StorageConstants.TEXT_NULL, "\\\\N");
meta.putOption(StorageConstants.RCFILE_NULL, "\\\\N");
meta.putOption(StorageConstants.RCFILE_SERDE, TextSerializerDeserializer.class.getName());
meta.putOption(StorageConstants.SEQUENCEFILE_NULL, "\\");
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/test/resources/storage-default.xml
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/resources/storage-default.xml b/tajo-storage/src/test/resources/storage-default.xml
index 6bfc902..a81f3d6 100644
--- a/tajo-storage/src/test/resources/storage-default.xml
+++ b/tajo-storage/src/test/resources/storage-default.xml
@@ -28,11 +28,15 @@
<!--- Registered Scanner Handler -->
<property>
<name>tajo.storage.scanner-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
+ <value>textfile,csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
</property>
<!--- Fragment Class Configurations -->
<property>
+ <name>tajo.storage.fragment.textfile.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
<name>tajo.storage.fragment.csv.class</name>
<value>org.apache.tajo.storage.fragment.FileFragment</value>
</property>
@@ -67,6 +71,11 @@
<!--- Scanner Handler -->
<property>
+ <name>tajo.storage.scanner-handler.textfile.class</name>
+ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileScanner</value>
+ </property>
+
+ <property>
<name>tajo.storage.scanner-handler.csv.class</name>
<value>org.apache.tajo.storage.CSVFile$CSVScanner</value>
</property>
@@ -109,7 +118,12 @@
<!--- Appender Handler -->
<property>
<name>tajo.storage.appender-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
+ <value>textfile,csv,raw,rcfile,row,trevni,parquet,sequencefile,avro</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.textfile.class</name>
+ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileAppender</value>
</property>
<property>
[2/2] tajo git commit: TAJO-1149: Implement direct read of
DelimitedTextFile. (jinho)
Posted by jh...@apache.org.
TAJO-1149: Implement direct read of DelimitedTextFile. (jinho)
Closes #221
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/3e305b15
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/3e305b15
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/3e305b15
Branch: refs/heads/master
Commit: 3e305b15fa94b1ed76dfa0e769825137ba1dfc1a
Parents: 5a72e2f
Author: jhkim <jh...@apache.org>
Authored: Mon Nov 10 16:45:29 2014 +0900
Committer: jhkim <jh...@apache.org>
Committed: Mon Nov 10 16:45:29 2014 +0900
----------------------------------------------------------------------
CHANGES | 5 +
.../org/apache/tajo/catalog/CatalogUtil.java | 12 +-
.../src/main/proto/CatalogProtos.proto | 1 +
.../tajo/catalog/store/HCatalogStore.java | 16 +-
.../apache/tajo/catalog/store/HCatalogUtil.java | 2 +-
.../tajo/catalog/store/TestHCatalogStore.java | 18 +-
.../main/java/org/apache/tajo/SessionVars.java | 7 +-
.../java/org/apache/tajo/conf/TajoConf.java | 2 +-
.../org/apache/tajo/storage/EmptyTuple.java | 176 +++++++
.../apache/tajo/storage/StorageConstants.java | 9 +
.../java/org/apache/tajo/benchmark/TPCH.java | 2 +-
.../apache/tajo/engine/parser/SQLAnalyzer.java | 8 +-
.../engine/planner/PhysicalPlannerImpl.java | 2 +-
.../planner/physical/PhysicalPlanUtil.java | 41 +-
.../apache/tajo/master/querymaster/Query.java | 2 +-
.../test/java/org/apache/tajo/TpchTestBase.java | 2 +-
.../apache/tajo/cli/tools/TestDDLBuilder.java | 2 +-
.../org/apache/tajo/client/TestTajoClient.java | 2 +-
.../planner/physical/TestBSTIndexExec.java | 3 -
.../apache/tajo/engine/query/TestCTASQuery.java | 2 +-
.../tajo/engine/query/TestGroupByQuery.java | 17 +-
.../tajo/engine/query/TestJoinBroadcast.java | 4 +-
.../apache/tajo/engine/query/TestJoinQuery.java | 16 +-
.../tajo/engine/query/TestNullValues.java | 14 +-
.../tajo/engine/query/TestSelectQuery.java | 4 +-
.../apache/tajo/engine/query/TestSortQuery.java | 12 +-
.../tajo/engine/query/TestTablePartitions.java | 4 +-
.../org/apache/tajo/jdbc/TestResultSet.java | 2 +-
.../results/TestTajoCli/testDescTable.result | 4 +-
.../results/TestTajoDump/testDump1.result | 2 +-
.../testBuildDDLForBaseTable.result | 2 +-
.../testBuildDDLForExternalTable.result | 2 +-
.../testBuildDDLQuotedTableName1.result | 2 +-
.../testBuildDDLQuotedTableName2.result | 2 +-
.../main/sphinx/getting_started/first_query.rst | 2 +-
tajo-docs/src/main/sphinx/sql_language/ddl.rst | 2 +-
.../src/main/sphinx/table_management/csv.rst | 10 +-
tajo-docs/src/main/sphinx/tsql/variables.rst | 2 +-
tajo-project/pom.xml | 5 +
tajo-storage/pom.xml | 4 +
.../storage/BinarySerializerDeserializer.java | 1 +
.../org/apache/tajo/storage/BufferPool.java | 74 +++
.../tajo/storage/ByteBufInputChannel.java | 76 +++
.../java/org/apache/tajo/storage/CSVFile.java | 23 +-
.../storage/FieldSerializerDeserializer.java | 35 ++
.../tajo/storage/SerializerDeserializer.java | 2 +-
.../storage/TextSerializerDeserializer.java | 1 +
.../tajo/storage/text/ByteBufLineReader.java | 154 ++++++
.../tajo/storage/text/DelimitedLineReader.java | 157 ++++++
.../tajo/storage/text/DelimitedTextFile.java | 483 +++++++++++++++++++
.../tajo/storage/text/FieldSplitProcessor.java | 38 ++
.../tajo/storage/text/LineSplitProcessor.java | 45 ++
.../text/TextFieldSerializerDeserializer.java | 227 +++++++++
.../src/main/resources/storage-default.xml | 59 +--
.../tajo/storage/TestCompressionStorages.java | 76 +--
.../org/apache/tajo/storage/TestLineReader.java | 163 +++++++
.../apache/tajo/storage/TestSplitProcessor.java | 72 +++
.../org/apache/tajo/storage/TestStorages.java | 4 +-
.../src/test/resources/storage-default.xml | 18 +-
59 files changed, 1897 insertions(+), 237 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 2c1e817..c699a34 100644
--- a/CHANGES
+++ b/CHANGES
@@ -60,6 +60,11 @@ Release 0.9.1 - unreleased
TAJO-1141: Refactor the packages hierarchy of tajo-client. (hyunsik)
+ SUB TASKS
+
+ TAJO-1149: Implement direct read of DelimitedTextFile. (jinho)
+
+
Release 0.9.0
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
index 8be4a8d..6e66d2a 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java
@@ -31,7 +31,6 @@ import org.apache.tajo.common.TajoDataTypes;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.exception.InvalidOperationException;
import org.apache.tajo.storage.StorageConstants;
-import org.apache.tajo.unit.StorageUnit;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.StringUtils;
import org.apache.tajo.util.TUtil;
@@ -40,7 +39,10 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
-import java.util.*;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import static org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import static org.apache.tajo.common.TajoDataTypes.Type;
@@ -276,6 +278,8 @@ public class CatalogUtil {
return StoreType.SEQUENCEFILE;
} else if (typeStr.equalsIgnoreCase(StoreType.AVRO.name())) {
return StoreType.AVRO;
+ } else if (typeStr.equalsIgnoreCase(StoreType.TEXTFILE.name())) {
+ return StoreType.TEXTFILE;
} else {
return null;
}
@@ -817,8 +821,8 @@ public class CatalogUtil {
public static KeyValueSet newPhysicalProperties(StoreType type) {
KeyValueSet options = new KeyValueSet();
- if (StoreType.CSV == type) {
- options.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ if (StoreType.CSV == type || StoreType.TEXTFILE == type) {
+ options.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
} else if (StoreType.RCFILE == type) {
options.set(StorageConstants.RCFILE_SERDE, StorageConstants.DEFAULT_BINARY_SERDE);
} else if (StoreType.SEQUENCEFILE == type) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index db402ae..31a7446 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -36,6 +36,7 @@ enum StoreType {
PARQUET = 7;
SEQUENCEFILE = 8;
AVRO = 9;
+ TEXTFILE = 10;
}
enum OrderType {
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
index fb8f4c4..4c37d75 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogStore.java
@@ -193,9 +193,9 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
String fileOutputformat = properties.getProperty(hive_metastoreConstants.FILE_OUTPUT_FORMAT);
storeType = CatalogUtil.getStoreType(HCatalogUtil.getStoreType(fileOutputformat));
- if (storeType.equals(CatalogProtos.StoreType.CSV)) {
- options.set(StorageConstants.CSVFILE_DELIMITER, StringEscapeUtils.escapeJava(fieldDelimiter));
- options.set(StorageConstants.CSVFILE_NULL, StringEscapeUtils.escapeJava(nullFormat));
+ if (storeType.equals(CatalogProtos.StoreType.TEXTFILE)) {
+ options.set(StorageConstants.TEXT_DELIMITER, StringEscapeUtils.escapeJava(fieldDelimiter));
+ options.set(StorageConstants.TEXT_NULL, StringEscapeUtils.escapeJava(nullFormat));
} else if (storeType.equals(CatalogProtos.StoreType.RCFILE)) {
options.set(StorageConstants.RCFILE_NULL, StringEscapeUtils.escapeJava(nullFormat));
String serde = properties.getProperty(serdeConstants.SERIALIZATION_LIB);
@@ -491,7 +491,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
sd.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class.getName());
sd.setOutputFormat(org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat.class.getName());
- String fieldDelimiter = tableDesc.getMeta().getOption(StorageConstants.CSVFILE_DELIMITER,
+ String fieldDelimiter = tableDesc.getMeta().getOption(StorageConstants.TEXT_DELIMITER,
StorageConstants.DEFAULT_FIELD_DELIMITER);
// User can use an unicode for filed delimiter such as \u0001, \001.
@@ -503,12 +503,12 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
StringEscapeUtils.unescapeJava(fieldDelimiter));
sd.getSerdeInfo().putToParameters(serdeConstants.FIELD_DELIM,
StringEscapeUtils.unescapeJava(fieldDelimiter));
- table.getParameters().remove(StorageConstants.CSVFILE_DELIMITER);
+ table.getParameters().remove(StorageConstants.TEXT_DELIMITER);
- if (tableDesc.getMeta().containsOption(StorageConstants.CSVFILE_NULL)) {
+ if (tableDesc.getMeta().containsOption(StorageConstants.TEXT_NULL)) {
table.putToParameters(serdeConstants.SERIALIZATION_NULL_FORMAT,
- StringEscapeUtils.unescapeJava(tableDesc.getMeta().getOption(StorageConstants.CSVFILE_NULL)));
- table.getParameters().remove(StorageConstants.CSVFILE_NULL);
+ StringEscapeUtils.unescapeJava(tableDesc.getMeta().getOption(StorageConstants.TEXT_NULL)));
+ table.getParameters().remove(StorageConstants.TEXT_NULL);
}
} else if (tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.SEQUENCEFILE)) {
String serde = tableDesc.getMeta().getOption(StorageConstants.SEQUENCEFILE_SERDE);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
index 54fdb9d..c7f2590 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/main/java/org/apache/tajo/catalog/store/HCatalogUtil.java
@@ -128,7 +128,7 @@ public class HCatalogUtil {
String outputFormatClass = fileFormatArrary[fileFormatArrary.length-1];
if(outputFormatClass.equals(HiveIgnoreKeyTextOutputFormat.class.getSimpleName())) {
- return CatalogProtos.StoreType.CSV.name();
+ return CatalogProtos.StoreType.TEXTFILE.name();
} else if(outputFormatClass.equals(HiveSequenceFileOutputFormat.class.getSimpleName())) {
return CatalogProtos.StoreType.SEQUENCEFILE.name();
} else if(outputFormatClass.equals(RCFileOutputFormat.class.getSimpleName())) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/test/java/org/apache/tajo/catalog/store/TestHCatalogStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/test/java/org/apache/tajo/catalog/store/TestHCatalogStore.java b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/test/java/org/apache/tajo/catalog/store/TestHCatalogStore.java
index f225424..5f4ad45 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/test/java/org/apache/tajo/catalog/store/TestHCatalogStore.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hcatalog/src/test/java/org/apache/tajo/catalog/store/TestHCatalogStore.java
@@ -110,7 +110,7 @@ public class TestHCatalogStore {
}
assertEquals(StringEscapeUtils.escapeJava(StorageConstants.DEFAULT_FIELD_DELIMITER),
- table1.getMeta().getOption(StorageConstants.CSVFILE_DELIMITER));
+ table1.getMeta().getOption(StorageConstants.TEXT_DELIMITER));
store.dropTable(DB_NAME, CUSTOMER);
}
@@ -174,8 +174,8 @@ public class TestHCatalogStore {
@Test
public void testTableWithNullValue() throws Exception {
KeyValueSet options = new KeyValueSet();
- options.set(StorageConstants.CSVFILE_DELIMITER, StringEscapeUtils.escapeJava("\u0002"));
- options.set(StorageConstants.CSVFILE_NULL, StringEscapeUtils.escapeJava("\u0003"));
+ options.set(StorageConstants.TEXT_DELIMITER, StringEscapeUtils.escapeJava("\u0002"));
+ options.set(StorageConstants.TEXT_NULL, StringEscapeUtils.escapeJava("\u0003"));
TableMeta meta = new TableMeta(CatalogProtos.StoreType.CSV, options);
org.apache.tajo.catalog.Schema schema = new org.apache.tajo.catalog.Schema();
@@ -201,16 +201,16 @@ public class TestHCatalogStore {
assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
}
- assertEquals(table.getMeta().getOption(StorageConstants.CSVFILE_DELIMITER),
- table1.getMeta().getOption(StorageConstants.CSVFILE_DELIMITER));
+ assertEquals(table.getMeta().getOption(StorageConstants.TEXT_DELIMITER),
+ table1.getMeta().getOption(StorageConstants.TEXT_DELIMITER));
- assertEquals(table.getMeta().getOption(StorageConstants.CSVFILE_NULL),
- table1.getMeta().getOption(StorageConstants.CSVFILE_NULL));
+ assertEquals(table.getMeta().getOption(StorageConstants.TEXT_NULL),
+ table1.getMeta().getOption(StorageConstants.TEXT_NULL));
- assertEquals(table1.getMeta().getOption(StorageConstants.CSVFILE_DELIMITER),
+ assertEquals(table1.getMeta().getOption(StorageConstants.TEXT_DELIMITER),
StringEscapeUtils.escapeJava("\u0002"));
- assertEquals(table1.getMeta().getOption(StorageConstants.CSVFILE_NULL),
+ assertEquals(table1.getMeta().getOption(StorageConstants.TEXT_NULL),
StringEscapeUtils.escapeJava("\u0003"));
store.dropTable(DB_NAME, SUPPLIER);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
index dbe949a..dd33e12 100644
--- a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
+++ b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
@@ -19,12 +19,11 @@
package org.apache.tajo;
import com.google.common.collect.Maps;
-
-import java.util.Map;
-
import org.apache.tajo.validation.Validator;
import org.apache.tajo.validation.Validators;
+import java.util.Map;
+
import static org.apache.tajo.SessionVars.VariableMode.*;
import static org.apache.tajo.conf.TajoConf.ConfVars;
@@ -121,7 +120,7 @@ public enum SessionVars implements ConfigKey {
DEFAULT, Long.class, Validators.min("0")),
MAX_OUTPUT_FILE_SIZE(ConfVars.$MAX_OUTPUT_FILE_SIZE, "Maximum per-output file size (mb). 0 means infinite.", DEFAULT,
Long.class, Validators.min("0")),
- NULL_CHAR(ConfVars.$CSVFILE_NULL, "null char of text file output", DEFAULT),
+ NULL_CHAR(ConfVars.$TEXT_NULL, "null char of text file output", DEFAULT),
CODEGEN(ConfVars.$CODEGEN, "Runtime code generation enabled (experiment)", DEFAULT),
// Behavior Control ---------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 62f5007..786aed0 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -359,7 +359,7 @@ public class TajoConf extends Configuration {
$DATE_ORDER("tajo.date.order", "YMD"),
// FILE FORMAT
- $CSVFILE_NULL("tajo.csvfile.null", "\\\\N"),
+ $TEXT_NULL("tajo.text.null", "\\\\N"),
// Only for Debug and Testing
$DEBUG_ENABLED("tajo.debug.enabled", false),
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-common/src/main/java/org/apache/tajo/storage/EmptyTuple.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/EmptyTuple.java b/tajo-common/src/main/java/org/apache/tajo/storage/EmptyTuple.java
new file mode 100644
index 0000000..ec0095b
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/EmptyTuple.java
@@ -0,0 +1,176 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+
+/* This class doesn’t have content datum. if selected column is zero, this is useful
+* e.g. select count(*) from table
+* */
+public class EmptyTuple implements Tuple, Cloneable {
+
+ private static EmptyTuple tuple;
+ private static Datum[] EMPTY_VALUES = new Datum[0];
+
+ static {
+ tuple = new EmptyTuple();
+ }
+
+ public static EmptyTuple get() {
+ return tuple;
+ }
+
+ private EmptyTuple() {
+ }
+
+ @Override
+ public int size() {
+ return 0;
+ }
+
+ public boolean contains(int fieldId) {
+ return false;
+ }
+
+ @Override
+ public boolean isNull(int fieldid) {
+ return true;
+ }
+
+ @Override
+ public boolean isNotNull(int fieldid) {
+ return false;
+ }
+
+ @Override
+ public void clear() {
+ }
+
+ @Override
+ public void put(int fieldId, Datum value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void put(int fieldId, Datum[] values) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void put(int fieldId, Tuple tuple) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void put(Datum[] values) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Datum get(int fieldId) {
+ return NullDatum.get();
+ }
+
+ @Override
+ public void setOffset(long offset) {
+
+ }
+
+ @Override
+ public long getOffset() {
+ return -1;
+ }
+
+ @Override
+ public boolean getBool(int fieldId) {
+ return NullDatum.get().asBool();
+ }
+
+ @Override
+ public byte getByte(int fieldId) {
+ return NullDatum.get().asByte();
+ }
+
+ @Override
+ public char getChar(int fieldId) {
+ return NullDatum.get().asChar();
+ }
+
+ @Override
+ public byte[] getBytes(int fieldId) {
+ return NullDatum.get().asByteArray();
+ }
+
+ @Override
+ public short getInt2(int fieldId) {
+ return NullDatum.get().asInt2();
+ }
+
+ @Override
+ public int getInt4(int fieldId) {
+ return NullDatum.get().asInt4();
+ }
+
+ @Override
+ public long getInt8(int fieldId) {
+ return NullDatum.get().asInt8();
+ }
+
+ @Override
+ public float getFloat4(int fieldId) {
+ return NullDatum.get().asFloat4();
+ }
+
+ @Override
+ public double getFloat8(int fieldId) {
+ return NullDatum.get().asFloat8();
+ }
+
+ @Override
+ public String getText(int fieldId) {
+ return NullDatum.get().asChars();
+ }
+
+ @Override
+ public ProtobufDatum getProtobufDatum(int fieldId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Datum getInterval(int fieldId) {
+ return NullDatum.get();
+ }
+
+ @Override
+ public char[] getUnicodeChars(int fieldId) {
+ return NullDatum.get().asUnicodeChars();
+ }
+
+ @Override
+ public Tuple clone() throws CloneNotSupportedException {
+ return this.clone();
+ }
+
+ @Override
+ public Datum[] getValues() {
+ return EMPTY_VALUES;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-common/src/main/java/org/apache/tajo/storage/StorageConstants.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/storage/StorageConstants.java b/tajo-common/src/main/java/org/apache/tajo/storage/StorageConstants.java
index a76e9ef..11ac9b7 100644
--- a/tajo-common/src/main/java/org/apache/tajo/storage/StorageConstants.java
+++ b/tajo-common/src/main/java/org/apache/tajo/storage/StorageConstants.java
@@ -23,15 +23,24 @@ public class StorageConstants {
public static final String COMPRESSION_CODEC = "compression.codec";
public static final String COMPRESSION_TYPE = "compression.type";
+ @Deprecated
public static final String CSVFILE_DELIMITER = "csvfile.delimiter";
+ @Deprecated
public static final String CSVFILE_NULL = "csvfile.null";
+ @Deprecated
public static final String CSVFILE_SERDE = "csvfile.serde";
+ public static final String TEXT_DELIMITER = "text.delimiter";
+ public static final String TEXT_NULL = "text.null";
+ public static final String TEXTFILE_SERDE = "textfile.serde";
+ @Deprecated
public static final String SEQUENCEFILE_DELIMITER = "sequencefile.delimiter";
+ @Deprecated
public static final String SEQUENCEFILE_NULL = "sequencefile.null";
public static final String SEQUENCEFILE_SERDE = "sequencefile.serde";
+ @Deprecated
public static final String RCFILE_NULL = "rcfile.null";
public static final String RCFILE_SERDE = "rcfile.serde";
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/main/java/org/apache/tajo/benchmark/TPCH.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/benchmark/TPCH.java b/tajo-core/src/main/java/org/apache/tajo/benchmark/TPCH.java
index 71d930f..f4b4d6a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/benchmark/TPCH.java
+++ b/tajo-core/src/main/java/org/apache/tajo/benchmark/TPCH.java
@@ -205,7 +205,7 @@ public class TPCH extends BenchmarkSet {
public void loadTable(String tableName) throws ServiceException {
TableMeta meta = CatalogUtil.newTableMeta(StoreType.CSV);
- meta.putOption(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ meta.putOption(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
PartitionMethodDesc partitionMethodDesc = null;
if (tableName.equals(CUSTOMER_PARTS)) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
index 580ec61..03b10c9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
@@ -1484,8 +1484,12 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
public Map<String, String> escapeTableMeta(Map<String, String> map) {
Map<String, String> params = new HashMap<String, String>();
for (Map.Entry<String, String> entry : map.entrySet()) {
- if (entry.getKey().equals(StorageConstants.CSVFILE_DELIMITER)) {
- params.put(entry.getKey(), StringUtils.unicodeEscapedDelimiter(entry.getValue()));
+ if (entry.getKey().equals(StorageConstants.CSVFILE_DELIMITER)
+ || entry.getKey().equals(StorageConstants.TEXT_DELIMITER)) { //backward compatibility
+ params.put(StorageConstants.TEXT_DELIMITER, StringUtils.unicodeEscapedDelimiter(entry.getValue()));
+ } else if (entry.getKey().equals(StorageConstants.CSVFILE_NULL)
+ || entry.getKey().equals(StorageConstants.TEXT_NULL)) { //backward compatibility
+ params.put(StorageConstants.TEXT_NULL, entry.getValue());
} else {
params.put(entry.getKey(), entry.getValue());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index 99b79ec..6806609 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -789,7 +789,7 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
// if there is no given NULL CHAR property in the table property and the query is neither CTAS or INSERT,
// we set DEFAULT NULL CHAR to the table property.
if (!ctx.getQueryContext().containsKey(SessionVars.NULL_CHAR)) {
- plan.getOptions().set(StorageConstants.CSVFILE_NULL, TajoConf.ConfVars.$CSVFILE_NULL.defaultVal);
+ plan.getOptions().set(StorageConstants.TEXT_NULL, TajoConf.ConfVars.$TEXT_NULL.defaultVal);
}
return new StoreTableExec(ctx, plan, subOp);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
index ec5915f..da5e073 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/PhysicalPlanUtil.java
@@ -173,16 +173,19 @@ public class PhysicalPlanUtil {
*/
private static void setNullCharForTextSerializer(TableMeta meta, String nullChar) {
switch (meta.getStoreType()) {
- case CSV:
- meta.putOption(StorageConstants.CSVFILE_NULL, nullChar);
- break;
- case RCFILE:
- meta.putOption(StorageConstants.RCFILE_NULL, nullChar);
- break;
- case SEQUENCEFILE:
- meta.putOption(StorageConstants.SEQUENCEFILE_NULL, nullChar);
- break;
- default: // nothing to do
+ case CSV:
+ meta.putOption(StorageConstants.TEXT_NULL, nullChar);
+ break;
+ case TEXTFILE:
+ meta.putOption(StorageConstants.TEXT_NULL, nullChar);
+ break;
+ case RCFILE:
+ meta.putOption(StorageConstants.RCFILE_NULL, nullChar);
+ break;
+ case SEQUENCEFILE:
+ meta.putOption(StorageConstants.SEQUENCEFILE_NULL, nullChar);
+ break;
+ default: // nothing to do
}
}
@@ -194,14 +197,16 @@ public class PhysicalPlanUtil {
*/
public static boolean containsNullChar(TableMeta meta) {
switch (meta.getStoreType()) {
- case CSV:
- return meta.containsOption(StorageConstants.CSVFILE_NULL);
- case RCFILE:
- return meta.containsOption(StorageConstants.RCFILE_NULL);
- case SEQUENCEFILE:
- return meta.containsOption(StorageConstants.SEQUENCEFILE_NULL);
- default: // nothing to do
- return false;
+ case CSV:
+ return meta.containsOption(StorageConstants.TEXT_NULL);
+ case TEXTFILE:
+ return meta.containsOption(StorageConstants.TEXT_NULL);
+ case RCFILE:
+ return meta.containsOption(StorageConstants.RCFILE_NULL);
+ case SEQUENCEFILE:
+ return meta.containsOption(StorageConstants.SEQUENCEFILE_NULL);
+ default: // nothing to do
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
index 1f445ec..a16d36a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -746,7 +746,7 @@ public class Query implements EventHandler<QueryEvent> {
TableMeta meta = lastStage.getTableMeta();
String nullChar = queryContext.get(SessionVars.NULL_CHAR);
- meta.putOption(StorageConstants.CSVFILE_NULL, nullChar);
+ meta.putOption(StorageConstants.TEXT_NULL, nullChar);
TableStats stats = lastStage.getResultStats();
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/TpchTestBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/TpchTestBase.java b/tajo-core/src/test/java/org/apache/tajo/TpchTestBase.java
index aacae36..0f713e5 100644
--- a/tajo-core/src/test/java/org/apache/tajo/TpchTestBase.java
+++ b/tajo-core/src/test/java/org/apache/tajo/TpchTestBase.java
@@ -91,7 +91,7 @@ public class TpchTestBase {
private void setUp() throws Exception {
util = new LocalTajoTestingUtility();
KeyValueSet opt = new KeyValueSet();
- opt.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ opt.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
util.setup(names, paths, schemas, opt);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/cli/tools/TestDDLBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/cli/tools/TestDDLBuilder.java b/tajo-core/src/test/java/org/apache/tajo/cli/tools/TestDDLBuilder.java
index 0ad3af1..83ce459 100644
--- a/tajo-core/src/test/java/org/apache/tajo/cli/tools/TestDDLBuilder.java
+++ b/tajo-core/src/test/java/org/apache/tajo/cli/tools/TestDDLBuilder.java
@@ -42,7 +42,7 @@ public class TestDDLBuilder {
schema1.addColumn("addr", TajoDataTypes.Type.TEXT);
meta1 = CatalogUtil.newTableMeta(CatalogProtos.StoreType.CSV);
- meta1.putOption(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ meta1.putOption(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
meta1.putOption(StorageConstants.COMPRESSION_CODEC, GzipCodec.class.getName());
Schema expressionSchema = new Schema();
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
index 719a775..faff5fb 100644
--- a/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
+++ b/tajo-core/src/test/java/org/apache/tajo/client/TestTajoClient.java
@@ -695,7 +695,7 @@ public class TestTajoClient {
TajoResultSet res = (TajoResultSet)client.executeQueryAndGetResult(sql);
- assertEquals(res.getTableDesc().getMeta().getOption(StorageConstants.CSVFILE_NULL), "\\\\T");
+ assertEquals(res.getTableDesc().getMeta().getOption(StorageConstants.TEXT_NULL), "\\\\T");
Path path = res.getTableDesc().getPath();
FileSystem fs = path.getFileSystem(tajoConf);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
index c7fbf39..e12d6eb 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestBSTIndexExec.java
@@ -161,9 +161,6 @@ public class TestBSTIndexExec {
@Test
public void testEqual() throws Exception {
- if(conf.getBoolean("tajo.storage.manager.v2", false)) {
- return;
- }
this.rndKey = rnd.nextInt(250);
final String QUERY = "select * from employee where managerId = " + rndKey;
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
index c2db595..27d59a9 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
@@ -211,7 +211,7 @@ public class TestCTASQuery extends QueryTestCaseBase {
KeyValueSet options = desc.getMeta().getOptions();
assertNotNull(options);
- assertEquals(StringEscapeUtils.escapeJava("\u0001"), options.get(StorageConstants.CSVFILE_DELIMITER));
+ assertEquals(StringEscapeUtils.escapeJava("\u0001"), options.get(StorageConstants.TEXT_DELIMITER));
}
@Test
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
index 8b9f9f7..cccbf2f 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestGroupByQuery.java
@@ -24,7 +24,6 @@ import org.apache.tajo.*;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf.ConfVars;
-import org.apache.tajo.ipc.TajoWorkerProtocol.ShuffleFileOutput;
import org.apache.tajo.master.querymaster.Query;
import org.apache.tajo.master.querymaster.QueryMasterTask;
import org.apache.tajo.master.querymaster.QueryUnit;
@@ -395,8 +394,8 @@ public class TestGroupByQuery extends QueryTestCaseBase {
// case9
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.TEXT);
@@ -449,8 +448,8 @@ public class TestGroupByQuery extends QueryTestCaseBase {
public final void testDistinctAggregationCaseByCase3() throws Exception {
// first distinct is smaller than second distinct.
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", Type.TEXT);
@@ -479,8 +478,8 @@ public class TestGroupByQuery extends QueryTestCaseBase {
public final void testDistinctAggregationCaseByCase4() throws Exception {
// Reproduction case for TAJO-994
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", Type.TEXT);
@@ -666,8 +665,8 @@ public class TestGroupByQuery extends QueryTestCaseBase {
@Test
public final void testNumShufflePartition() throws Exception {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", Type.TEXT);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
index 222b508..3d413b7 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinBroadcast.java
@@ -580,8 +580,8 @@ public class TestJoinBroadcast extends QueryTestCaseBase {
@Test
public final void testLeftOuterJoinLeftSideSmallTable() throws Exception {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
index 4afa9c3..223ea8e 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestJoinQuery.java
@@ -420,8 +420,8 @@ public class TestJoinQuery extends QueryTestCaseBase {
public final void testLeftOuterJoinWithEmptySubquery1() throws Exception {
// Empty Null Supplying table
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
@@ -462,8 +462,8 @@ public class TestJoinQuery extends QueryTestCaseBase {
public final void testLeftOuterJoinWithEmptySubquery2() throws Exception {
//Empty Preserved Row table
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
@@ -951,8 +951,8 @@ public class TestJoinQuery extends QueryTestCaseBase {
private void createOuterJoinTestTable() throws Exception {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
@@ -1085,8 +1085,8 @@ public class TestJoinQuery extends QueryTestCaseBase {
@Test
public void testJoinWithDifferentShuffleKey() throws Exception {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestNullValues.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
index a492d6e..4981e89 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
@@ -54,7 +54,7 @@ public class TestNullValues {
"3|filled|0.2"
};
KeyValueSet opts = new KeyValueSet();
- opts.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ opts.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable1 where col3 is null");
@@ -81,7 +81,7 @@ public class TestNullValues {
"3|filled|"
};
KeyValueSet opts = new KeyValueSet();
- opts.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ opts.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable2 where col1 is not null");
@@ -116,7 +116,7 @@ public class TestNullValues {
",,,43578,19,13,6,3581,2557,1024"
};
KeyValueSet opts = new KeyValueSet();
- opts.set(StorageConstants.CSVFILE_DELIMITER, ",");
+ opts.set(StorageConstants.TEXT_DELIMITER, ",");
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable3 where col1 is null and col2 is null and col3 is null and col4 = 43578");
@@ -149,8 +149,8 @@ public class TestNullValues {
",\\N,,43578"
};
KeyValueSet opts = new KeyValueSet();
- opts.set(StorageConstants.CSVFILE_DELIMITER, ",");
- opts.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ opts.set(StorageConstants.TEXT_DELIMITER, ",");
+ opts.set(StorageConstants.TEXT_NULL, "\\\\N");
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable4 where col1 is null and col2 is null and col3 is null and col5 is null and col4 = 43578");
@@ -266,8 +266,8 @@ public class TestNullValues {
"4|d|4.0|\\N"
};
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
if (client == null) {
return TajoTestingCluster.run(table, schemas, tableOptions, new String[][]{data}, query);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
index 41e2f3e..5e6d05b 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
@@ -456,8 +456,8 @@ public class TestSelectQuery extends QueryTestCaseBase {
@Test
public final void testNowInMultipleTasks() throws Exception {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
index df02708..37454e6 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestSortQuery.java
@@ -181,8 +181,8 @@ public class TestSortQuery extends QueryTestCaseBase {
try {
testingCluster.setAllTajoDaemonConfValue(ConfVars.$TEST_MIN_TASK_NUM.varname, "2");
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("id", Type.INT4);
@@ -209,8 +209,8 @@ public class TestSortQuery extends QueryTestCaseBase {
try {
testingCluster.setAllTajoDaemonConfValue(ConfVars.$TEST_MIN_TASK_NUM.varname, "2");
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", Type.INT4);
@@ -237,8 +237,8 @@ public class TestSortQuery extends QueryTestCaseBase {
try {
testingCluster.setAllTajoDaemonConfValue(ConfVars.$TEST_MIN_TASK_NUM.varname, "2");
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", Type.INT4);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
index 828d2a3..6378814 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
@@ -839,8 +839,8 @@ public class TestTablePartitions extends QueryTestCaseBase {
testingCluster.setAllTajoDaemonConfValue(TajoConf.ConfVars.SHUFFLE_HASH_APPENDER_PAGE_VOLUME.varname, "1");
try {
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- tableOptions.set(StorageConstants.CSVFILE_NULL, "\\\\N");
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
Schema schema = new Schema();
schema.addColumn("col1", TajoDataTypes.Type.TEXT);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
index 1edf239..0c92ac1 100644
--- a/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
+++ b/tajo-core/src/test/java/org/apache/tajo/jdbc/TestResultSet.java
@@ -155,7 +155,7 @@ public class TestResultSet {
"2014-01-01|01:00:00|2014-01-01 01:00:00"
};
KeyValueSet tableOptions = new KeyValueSet();
- tableOptions.set(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+ tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
res = TajoTestingCluster
.run(table, schemas, tableOptions, new String[][]{data}, query, client);
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/TestTajoCli/testDescTable.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestTajoCli/testDescTable.result b/tajo-core/src/test/resources/results/TestTajoCli/testDescTable.result
index d9319eb..ae2af45 100644
--- a/tajo-core/src/test/resources/results/TestTajoCli/testDescTable.result
+++ b/tajo-core/src/test/resources/results/TestTajoCli/testDescTable.result
@@ -6,7 +6,7 @@ store type: CSV
number of rows: 0
volume: 0 B
Options:
- 'csvfile.delimiter'='|'
+ 'text.delimiter'='|'
schema:
col1 INT4
@@ -20,7 +20,7 @@ store type: CSV
number of rows: 0
volume: 0 B
Options:
- 'csvfile.delimiter'='|'
+ 'text.delimiter'='|'
schema:
col1 INT4
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/TestTajoDump/testDump1.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/TestTajoDump/testDump1.result b/tajo-core/src/test/resources/results/TestTajoDump/testDump1.result
index fa68435..326ba3f 100644
--- a/tajo-core/src/test/resources/results/TestTajoDump/testDump1.result
+++ b/tajo-core/src/test/resources/results/TestTajoDump/testDump1.result
@@ -12,5 +12,5 @@ CREATE DATABASE IF NOT EXISTS "TestTajoDump";
--
-- Name: "TestTajoDump"."TableName1"; Type: TABLE; Storage: CSV
--
-CREATE TABLE "TestTajoDump"."TableName1" ("Age" INT4, "FirstName" TEXT, lastname TEXT) USING CSV WITH ('csvfile.delimiter'='|');
+CREATE TABLE "TestTajoDump"."TableName1" ("Age" INT4, "FirstName" TEXT, lastname TEXT) USING CSV WITH ('text.delimiter'='|');
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForBaseTable.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForBaseTable.result b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForBaseTable.result
index a6e2d05..b1172a7 100644
--- a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForBaseTable.result
+++ b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForBaseTable.result
@@ -1,4 +1,4 @@
--
-- Name: db1.table2; Type: TABLE; Storage: CSV
--
-CREATE TABLE db1.table2 (name BLOB, addr TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'csvfile.delimiter'='|');
\ No newline at end of file
+CREATE TABLE db1.table2 (name BLOB, addr TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'text.delimiter'='|');
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForExternalTable.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForExternalTable.result b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForExternalTable.result
index 78c64ff..a5b5f63 100644
--- a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForExternalTable.result
+++ b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLForExternalTable.result
@@ -2,4 +2,4 @@
-- Name: db1.table1; Type: TABLE; Storage: CSV
-- Path: /table1
--
-CREATE EXTERNAL TABLE db1.table1 (name BLOB, addr TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'csvfile.delimiter'='|') PARTITION BY COLUMN(key INT4, key2 TEXT) LOCATION '/table1';
\ No newline at end of file
+CREATE EXTERNAL TABLE db1.table1 (name BLOB, addr TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'text.delimiter'='|') PARTITION BY COLUMN(key INT4, key2 TEXT) LOCATION '/table1';
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName1.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName1.result b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName1.result
index 3ff5134..43be915 100644
--- a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName1.result
+++ b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName1.result
@@ -2,4 +2,4 @@
-- Name: db1."TABLE2"; Type: TABLE; Storage: CSV
-- Path: /table1
--
-CREATE EXTERNAL TABLE db1."TABLE2" (name BLOB, addr TEXT, "FirstName" TEXT, "LastName" TEXT, "with" TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'csvfile.delimiter'='|') PARTITION BY COLUMN("BirthYear" INT4) LOCATION '/table1';
\ No newline at end of file
+CREATE EXTERNAL TABLE db1."TABLE2" (name BLOB, addr TEXT, "FirstName" TEXT, "LastName" TEXT, "with" TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'text.delimiter'='|') PARTITION BY COLUMN("BirthYear" INT4) LOCATION '/table1';
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName2.result
----------------------------------------------------------------------
diff --git a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName2.result b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName2.result
index eeb9b59..764892f 100644
--- a/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName2.result
+++ b/tajo-core/src/test/resources/results/testDDLBuilder/testBuildDDLQuotedTableName2.result
@@ -1,4 +1,4 @@
--
-- Name: db1."TABLE1"; Type: TABLE; Storage: CSV
--
-CREATE TABLE db1."TABLE1" (name BLOB, addr TEXT, "FirstName" TEXT, "LastName" TEXT, "with" TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'csvfile.delimiter'='|') PARTITION BY COLUMN("BirthYear" INT4);
\ No newline at end of file
+CREATE TABLE db1."TABLE1" (name BLOB, addr TEXT, "FirstName" TEXT, "LastName" TEXT, "with" TEXT) USING CSV WITH ('compression.codec'='org.apache.hadoop.io.compress.GzipCodec', 'text.delimiter'='|') PARTITION BY COLUMN("BirthYear" INT4);
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-docs/src/main/sphinx/getting_started/first_query.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/getting_started/first_query.rst b/tajo-docs/src/main/sphinx/getting_started/first_query.rst
index 80ee3b5..9498ef6 100644
--- a/tajo-docs/src/main/sphinx/getting_started/first_query.rst
+++ b/tajo-docs/src/main/sphinx/getting_started/first_query.rst
@@ -31,7 +31,7 @@ Here, we assume the schema as (int, text, float, text). ::
name text,
score float,
type text)
- using csv with ('csvfile.delimiter'='|') location 'file:/home/x/table1';
+ using csv with ('text.delimiter'='|') location 'file:/home/x/table1';
To load an external table, you need to use ‘create external table’ statement.
In the location clause, you should use the absolute directory path with an appropriate scheme.
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-docs/src/main/sphinx/sql_language/ddl.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/sql_language/ddl.rst b/tajo-docs/src/main/sphinx/sql_language/ddl.rst
index e2e64f6..3fba6be 100644
--- a/tajo-docs/src/main/sphinx/sql_language/ddl.rst
+++ b/tajo-docs/src/main/sphinx/sql_language/ddl.rst
@@ -56,7 +56,7 @@ If you want to add an external table that contains compressed data, you should g
...
L_COMMENT text)
- USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec')
+ USING csv WITH ('text.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec')
LOCATION 'hdfs://localhost:9010/tajo/warehouse/lineitem_100_snappy';
`compression.codec` parameter can have one of the following compression codecs:
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-docs/src/main/sphinx/table_management/csv.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/table_management/csv.rst b/tajo-docs/src/main/sphinx/table_management/csv.rst
index 2d3ee9e..3aba2ba 100644
--- a/tajo-docs/src/main/sphinx/table_management/csv.rst
+++ b/tajo-docs/src/main/sphinx/table_management/csv.rst
@@ -36,8 +36,8 @@ The ``WITH`` clause in the CREATE TABLE statement allows users to set those para
Now, the CSV storage format provides the following physical properties.
-* ``csvfile.delimiter``: delimiter character. ``|`` or ``\u0001`` is usually used, and the default field delimiter is ``|``.
-* ``csvfile.null``: NULL character. The default NULL character is an empty string ``''``. Hive's default NULL character is ``'\\N'``.
+* ``text.delimiter``: delimiter character. ``|`` or ``\u0001`` is usually used, and the default field delimiter is ``|``.
+* ``text.null``: NULL character. The default NULL character is an empty string ``''``. Hive's default NULL character is ``'\\N'``.
* ``compression.codec``: Compression codec. You can enable compression feature and set specified compression algorithm. The compression algorithm used to compress files. The compression codec name should be the fully qualified class name inherited from `org.apache.hadoop.io.compress.CompressionCodec <https://hadoop.apache.org/docs/current/api/org/apache/hadoop/io/compress/CompressionCodec.html>`_. By default, compression is disabled.
* ``csvfile.serde``: custom (De)serializer class. ``org.apache.tajo.storage.TextSerializerDeserializer`` is the default (De)serializer class.
@@ -50,8 +50,8 @@ The following example is to set a custom field delimiter, NULL character, and co
name text,
score float,
type text
- ) USING CSV WITH('csvfile.delimiter'='\u0001',
- 'csvfile.null'='\\N',
+ ) USING CSV WITH('text.delimiter'='\u0001',
+ 'text.null'='\\N',
'compression.codec'='org.apache.hadoop.io.compress.SnappyCodec');
.. warning::
@@ -88,7 +88,7 @@ Null Value Handling Issues
In default, NULL character in CSV files is an empty string ``''``.
In other words, an empty field is basically recognized as a NULL value in Tajo.
If a field domain is ``TEXT``, an empty field is recognized as a string value ``''`` instead of NULL value.
-Besides, You can also use your own NULL character by specifying a physical property ``csvfile.null``.
+Besides, You can also use your own NULL character by specifying a physical property ``text.null``.
=========================================
Compatibility Issues with Apache Hive™
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-docs/src/main/sphinx/tsql/variables.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/tsql/variables.rst b/tajo-docs/src/main/sphinx/tsql/variables.rst
index e859451..a4dd409 100644
--- a/tajo-docs/src/main/sphinx/tsql/variables.rst
+++ b/tajo-docs/src/main/sphinx/tsql/variables.rst
@@ -54,7 +54,7 @@ Now, tajo provides the following session variables.
* ``CLI_ERROR_STOP``
* ``TIMEZONE``
* ``DATE_ORDER``
-* ``CSVFILE_NULL``
+* ``TEXT_NULL``
* ``DEBUG_ENABLED``
* ``TEST_BROADCAST_JOIN_ENABLED``
* ``TEST_JOIN_OPT_ENABLED``
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-project/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-project/pom.xml b/tajo-project/pom.xml
index 97489da..aa8110f 100644
--- a/tajo-project/pom.xml
+++ b/tajo-project/pom.xml
@@ -1006,6 +1006,11 @@
<version>3.6.6.Final</version>
</dependency>
<dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ <version>4.0.23.Final</version>
+ </dependency>
+ <dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.3</version>
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-storage/pom.xml b/tajo-storage/pom.xml
index c50fa66..680e520 100644
--- a/tajo-storage/pom.xml
+++ b/tajo-storage/pom.xml
@@ -314,6 +314,10 @@
<artifactId>parquet-format</artifactId>
<version>${parquet.format.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ </dependency>
</dependencies>
<profiles>
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
index 42b49a8..00112e7 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/BinarySerializerDeserializer.java
@@ -27,6 +27,7 @@ import org.apache.tajo.util.Bytes;
import java.io.IOException;
import java.io.OutputStream;
+@Deprecated
public class BinarySerializerDeserializer implements SerializerDeserializer {
static final byte[] INVALID_UTF__SINGLE_BYTE = {(byte) Integer.parseInt("10111111", 2)};
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/BufferPool.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/BufferPool.java b/tajo-storage/src/main/java/org/apache/tajo/storage/BufferPool.java
new file mode 100644
index 0000000..85c79fa
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/BufferPool.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.util.internal.PlatformDependent;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/* this class is PooledBuffer holder */
+public class BufferPool {
+
+ private static final PooledByteBufAllocator allocator;
+
+ private BufferPool() {
+ }
+
+ static {
+ //TODO we need determine the default params
+ allocator = new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());
+
+ /* if you are finding memory leak, please enable this line */
+ //ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
+ }
+
+ public static long maxDirectMemory() {
+ return PlatformDependent.maxDirectMemory();
+ }
+
+
+ public synchronized static ByteBuf directBuffer(int size) {
+ return allocator.directBuffer(size);
+ }
+
+ /**
+ *
+ * @param size the initial capacity
+ * @param max the max capacity
+ * @return allocated ByteBuf from pool
+ */
+ public static ByteBuf directBuffer(int size, int max) {
+ return allocator.directBuffer(size, max);
+ }
+
+ @InterfaceStability.Unstable
+ public static void forceRelease(ByteBuf buf) {
+ buf.release(buf.refCnt());
+ }
+
+ /**
+ * the ByteBuf will increase to writable size
+ * @param buf
+ * @param minWritableBytes required minimum writable size
+ */
+ public static void ensureWritable(ByteBuf buf, int minWritableBytes) {
+ buf.ensureWritable(minWritableBytes);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/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
new file mode 100644
index 0000000..b1b6d65
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/ByteBufInputChannel.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.hdfs.DFSInputStream;
+import org.apache.hadoop.io.IOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.ScatteringByteChannel;
+import java.nio.channels.spi.AbstractInterruptibleChannel;
+
+public class ByteBufInputChannel extends AbstractInterruptibleChannel implements ScatteringByteChannel {
+
+ ByteBufferReadable byteBufferReadable;
+ ReadableByteChannel channel;
+ InputStream inputStream;
+
+ public ByteBufInputChannel(InputStream inputStream) {
+ if (inputStream instanceof DFSInputStream && inputStream instanceof ByteBufferReadable) {
+ this.byteBufferReadable = (ByteBufferReadable) inputStream;
+ } else {
+ this.channel = Channels.newChannel(inputStream);
+ }
+
+ this.inputStream = inputStream;
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts, int offset, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long read(ByteBuffer[] dsts) {
+ return read(dsts, 0, dsts.length);
+ }
+
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ if (byteBufferReadable != null) {
+ return byteBufferReadable.read(dst);
+ } else {
+ return channel.read(dst);
+ }
+ }
+
+ @Override
+ 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/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/CSVFile.java b/tajo-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
index 2113794..5ddc3fb 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
@@ -81,12 +81,14 @@ public class CSVFile {
this.fs = path.getFileSystem(conf);
this.meta = meta;
this.schema = schema;
- this.delimiter = StringEscapeUtils.unescapeJava(this.meta.getOption(StorageConstants.CSVFILE_DELIMITER,
- StorageConstants.DEFAULT_FIELD_DELIMITER)).charAt(0);
+ this.delimiter = StringEscapeUtils.unescapeJava(
+ this.meta.getOption(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER)).charAt(0);
+
this.columnNum = schema.size();
- String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(StorageConstants.CSVFILE_NULL,
- NullDatum.DEFAULT_TEXT));
+ String nullCharacters = StringEscapeUtils.unescapeJava(
+ this.meta.getOption(StorageConstants.TEXT_NULL, NullDatum.DEFAULT_TEXT));
+
if (StringUtils.isEmpty(nullCharacters)) {
nullChars = NullDatum.get().asTextBytes();
} else {
@@ -140,6 +142,7 @@ public class CSVFile {
}
try {
+ //It will be remove, because we will add custom serde in textfile
String serdeClass = this.meta.getOption(StorageConstants.CSVFILE_SERDE,
TextSerializerDeserializer.class.getName());
serde = (SerializerDeserializer) Class.forName(serdeClass).newInstance();
@@ -261,11 +264,14 @@ public class CSVFile {
}
//Delimiter
- String delim = meta.getOption(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
- this.delimiter = StringEscapeUtils.unescapeJava(delim).charAt(0);
+ this.delimiter = StringEscapeUtils.unescapeJava(
+ meta.getOption(StorageConstants.TEXT_DELIMITER,
+ meta.getOption(StorageConstants.CSVFILE_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER))).charAt(0);
+
+ String nullCharacters = StringEscapeUtils.unescapeJava(
+ meta.getOption(StorageConstants.TEXT_NULL,
+ meta.getOption(StorageConstants.CSVFILE_NULL, NullDatum.DEFAULT_TEXT)));
- String nullCharacters = StringEscapeUtils.unescapeJava(meta.getOption(StorageConstants.CSVFILE_NULL,
- NullDatum.DEFAULT_TEXT));
if (StringUtils.isEmpty(nullCharacters)) {
nullChars = NullDatum.get().asTextBytes();
} else {
@@ -346,6 +352,7 @@ public class CSVFile {
}
try {
+ //FIXME
String serdeClass = this.meta.getOption(StorageConstants.CSVFILE_SERDE,
TextSerializerDeserializer.class.getName());
serde = (SerializerDeserializer) Class.forName(serdeClass).newInstance();
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/FieldSerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/FieldSerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/FieldSerializerDeserializer.java
new file mode 100644
index 0000000..7df4584
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/FieldSerializerDeserializer.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.datum.Datum;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+
+public interface FieldSerializerDeserializer {
+
+ public int serialize(OutputStream out, Datum datum, Column col, int columnIndex, byte[] nullChars) throws IOException;
+
+ public Datum deserialize(ByteBuf buf, Column col, int columnIndex, ByteBuf nullChars) throws IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/SerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/SerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/SerializerDeserializer.java
index 333f205..564a9f5 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/SerializerDeserializer.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/SerializerDeserializer.java
@@ -24,7 +24,7 @@ import org.apache.tajo.datum.Datum;
import java.io.IOException;
import java.io.OutputStream;
-
+@Deprecated
public interface SerializerDeserializer {
public int serialize(Column col, Datum datum, OutputStream out, byte[] nullCharacters) throws IOException;
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/tajo-storage/src/main/java/org/apache/tajo/storage/TextSerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/TextSerializerDeserializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/TextSerializerDeserializer.java
index d2ccdc7..094d285 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/TextSerializerDeserializer.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/TextSerializerDeserializer.java
@@ -32,6 +32,7 @@ import java.io.IOException;
import java.io.OutputStream;
//Compatibility with Apache Hive
+@Deprecated
public class TextSerializerDeserializer implements SerializerDeserializer {
public static final byte[] trueBytes = "true".getBytes();
public static final byte[] falseBytes = "false".getBytes();
http://git-wip-us.apache.org/repos/asf/tajo/blob/3e305b15/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
new file mode 100644
index 0000000..1448885
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/text/ByteBufLineReader.java
@@ -0,0 +1,154 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.storage.text;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.CharsetUtil;
+import org.apache.tajo.storage.BufferPool;
+import org.apache.tajo.storage.ByteBufInputChannel;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ByteBufLineReader implements Closeable {
+ private static int DEFAULT_BUFFER = 64 * 1024;
+
+ private int bufferSize;
+ private long readBytes;
+ private ByteBuf buffer;
+ private final ByteBufInputChannel channel;
+ private final AtomicInteger tempReadBytes = new AtomicInteger();
+ private final LineSplitProcessor processor = new LineSplitProcessor();
+
+ public ByteBufLineReader(ByteBufInputChannel channel) {
+ this(channel, BufferPool.directBuffer(DEFAULT_BUFFER));
+ }
+
+ public ByteBufLineReader(ByteBufInputChannel channel, ByteBuf buf) {
+ this.readBytes = 0;
+ this.channel = channel;
+ this.buffer = buf;
+ this.bufferSize = buf.capacity();
+ }
+
+ public long readBytes() {
+ 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) {
+ this.buffer.release();
+ }
+ this.channel.close();
+ }
+
+ public String readLine() throws IOException {
+ ByteBuf buf = readLineBuf(tempReadBytes);
+ if (buf != null) {
+ return buf.toString(CharsetUtil.UTF_8);
+ }
+ return null;
+ }
+
+ private void fillBuffer() throws IOException {
+
+ int tailBytes = 0;
+ if (this.readBytes > 0) {
+ this.buffer.markReaderIndex();
+ this.buffer.discardSomeReadBytes(); // compact the buffer
+ tailBytes = this.buffer.writerIndex();
+ if (!this.buffer.isWritable()) {
+ // a line bytes is large than the buffer
+ BufferPool.ensureWritable(buffer, bufferSize);
+ this.bufferSize = buffer.capacity();
+ }
+ }
+
+ boolean release = true;
+ try {
+ int readBytes = tailBytes;
+ for (; ; ) {
+ int localReadBytes = buffer.writeBytes(channel, bufferSize - readBytes);
+ if (localReadBytes < 0) {
+ break;
+ }
+ readBytes += localReadBytes;
+ if (readBytes == bufferSize) {
+ break;
+ }
+ }
+ this.readBytes += (readBytes - tailBytes);
+ release = false;
+ this.buffer.readerIndex(this.buffer.readerIndex() + tailBytes); //skip past buffer (tail)
+ } finally {
+ if (release) {
+ buffer.release();
+ }
+ }
+ }
+
+ /**
+ * Read a line terminated by one of CR, LF, or CRLF.
+ */
+ public ByteBuf readLineBuf(AtomicInteger reads) throws IOException {
+ int startIndex = buffer.readerIndex();
+ int readBytes;
+ int readable;
+ int newlineLength; //length of terminating newline
+
+ loop:
+ while (true) {
+ readable = buffer.readableBytes();
+ if (readable <= 0) {
+ buffer.readerIndex(startIndex);
+ fillBuffer(); //compact and fill buffer
+ if (!buffer.isReadable()) {
+ return null;
+ } else {
+ startIndex = 0; // reset the line start position
+ }
+ readable = buffer.readableBytes();
+ }
+
+ int endIndex = buffer.forEachByte(buffer.readerIndex(), readable, processor);
+ if (endIndex < 0) {
+ buffer.readerIndex(buffer.writerIndex());
+ } else {
+ buffer.readerIndex(endIndex + 1);
+ readBytes = buffer.readerIndex() - startIndex;
+ if (processor.isPrevCharCR() && buffer.isReadable()
+ && buffer.getByte(buffer.readerIndex()) == LineSplitProcessor.LF) {
+ buffer.skipBytes(1);
+ newlineLength = 2;
+ } else {
+ newlineLength = 1;
+ }
+ break loop;
+ }
+ }
+ reads.set(readBytes);
+ return buffer.slice(startIndex, readBytes - newlineLength);
+ }
+}