You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by bl...@apache.org on 2014/04/01 09:38:58 UTC
git commit: TAJO-616: SequenceFile support. (jaehwa)
Repository: tajo
Updated Branches:
refs/heads/master bd418a5c3 -> 303066b23
TAJO-616: SequenceFile support. (jaehwa)
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/303066b2
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/303066b2
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/303066b2
Branch: refs/heads/master
Commit: 303066b232d9f96c5983c65c2308cb79d7a7ab3d
Parents: bd418a5
Author: blrunner <jh...@gruter.com>
Authored: Tue Apr 1 16:37:00 2014 +0900
Committer: blrunner <jh...@gruter.com>
Committed: Tue Apr 1 16:37:00 2014 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../apache/tajo/catalog/CatalogConstants.java | 12 +-
.../org/apache/tajo/catalog/CatalogUtil.java | 9 +-
.../src/main/proto/CatalogProtos.proto | 1 +
.../tajo/catalog/store/HCatalogStore.java | 54 ++-
.../apache/tajo/catalog/store/HCatalogUtil.java | 3 +
.../tajo/catalog/store/TestHCatalogStore.java | 69 +++-
.../main/java/org/apache/tajo/util/Bytes.java | 33 ++
.../java/org/apache/tajo/benchmark/TPCH.java | 2 +-
.../test/java/org/apache/tajo/TpchTestBase.java | 2 +-
.../org/apache/tajo/client/TestDDLBuilder.java | 2 +-
.../apache/tajo/engine/query/TestCTASQuery.java | 15 +-
.../tajo/engine/query/TestNullValues.java | 4 +-
.../src/main/sphinx/table_management/csv.rst | 6 +
.../sphinx/table_management/file_formats.rst | 14 +-
.../main/sphinx/table_management/parquet.rst | 6 +
.../src/main/sphinx/table_management/rcfile.rst | 6 +
.../sphinx/table_management/sequencefile.rst | 111 +++++++
tajo-storage/pom.xml | 11 +
.../java/org/apache/tajo/storage/CSVFile.java | 4 +-
.../storage/parquet/TajoRecordMaterializer.java | 2 +-
.../org/apache/tajo/storage/rcfile/RCFile.java | 2 +-
.../sequencefile/SequenceFileAppender.java | 268 +++++++++++++++
.../sequencefile/SequenceFileScanner.java | 330 +++++++++++++++++++
.../tajo/storage/trevni/TrevniScanner.java | 2 +-
.../apache/tajo/storage/v2/CSVFileScanner.java | 2 +-
.../src/main/resources/storage-default.xml | 23 +-
.../tajo/storage/TestCompressionStorages.java | 10 +-
.../apache/tajo/storage/TestMergeScanner.java | 2 +
.../org/apache/tajo/storage/TestStorages.java | 146 +++++++-
.../src/test/resources/storage-default.xml | 24 +-
31 files changed, 1130 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d4b8452..80d05aa 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,8 @@ Release 0.8.0 - unreleased
NEW FEATURES
+ TAJO-616: SequenceFile support. (jaehwa)
+
TAJO-480: Umbrella Jira for adding ALTER TABLE statement.
(Alvin Henrick via hyunsik)
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
index c7116f6..c5e0dd4 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
@@ -47,15 +47,21 @@ public class CatalogConstants {
// table options
public static final String COMPRESSION_CODEC = "compression.codec";
+ public static final String COMPRESSION_TYPE = "compression.type";
+
public static final String CSVFILE_DELIMITER = "csvfile.delimiter";
public static final String CSVFILE_NULL = "csvfile.null";
public static final String CSVFILE_SERDE = "csvfile.serde";
- public static final String CSVFILE_DELIMITER_DEFAULT = "|";
+
+ public static final String SEQUENCEFILE_DELIMITER = "sequencefile.delimiter";
+ public static final String SEQUENCEFILE_NULL = "sequencefile.null";
+ public static final String SEQUENCEFILE_SERDE = "sequencefile.serde";
public static final String RCFILE_NULL = "rcfile.null";
public static final String RCFILE_SERDE = "rcfile.serde";
- public static final String RCFILE_BINARY_SERDE = "org.apache.tajo.storage.BinarySerializerDeserializer";
- public static final String RCFILE_TEXT_SERDE = "org.apache.tajo.storage.TextSerializerDeserializer";
+ public static final String DEFAULT_FIELD_DELIMITER = "|";
+ public static final String DEFAULT_BINARY_SERDE = "org.apache.tajo.storage.BinarySerializerDeserializer";
+ public static final String DEFAULT_TEXT_SERDE = "org.apache.tajo.storage.TextSerializerDeserializer";
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 22f54ca..6e7d2a5 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
@@ -281,6 +281,8 @@ public class CatalogUtil {
return StoreType.TREVNI;
} else if (typeStr.equalsIgnoreCase(StoreType.PARQUET.name())) {
return StoreType.PARQUET;
+ } else if (typeStr.equalsIgnoreCase(StoreType.SEQUENCEFILE.name())) {
+ return StoreType.SEQUENCEFILE;
} else {
return null;
}
@@ -288,9 +290,12 @@ public class CatalogUtil {
public static Options newOptionsWithDefault(StoreType type) {
Options options = new Options();
if(StoreType.CSV == type){
- options.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ options.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
} else if(StoreType.RCFILE == type){
- options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.RCFILE_BINARY_SERDE);
+ options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.DEFAULT_BINARY_SERDE);
+ } else if(StoreType.SEQUENCEFILE == type){
+ options.put(CatalogConstants.SEQUENCEFILE_SERDE, CatalogConstants.DEFAULT_TEXT_SERDE);
+ options.put(CatalogConstants.SEQUENCEFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
}
return options;
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 d3abc70..b985ba7 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -33,6 +33,7 @@ enum StoreType {
HCFILE = 5;
TREVNI = 6;
PARQUET = 7;
+ SEQUENCEFILE = 8;
}
enum OrderType {
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 a4bdb28..e6b0bcd 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
@@ -30,6 +30,8 @@ import org.apache.hadoop.hive.metastore.api.*;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe;
import org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe;
+import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
import org.apache.hcatalog.common.HCatUtil;
import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchema;
@@ -61,6 +63,9 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
private final String defaultTableSpaceUri;
public HCatalogStore(final Configuration conf) throws InternalException {
+ if (!(conf instanceof TajoConf)) {
+ throw new CatalogException("Invalid Configuration Type:" + conf.getClass().getSimpleName());
+ }
this.conf = conf;
this.defaultTableSpaceUri = TajoConf.getWarehouseDir((TajoConf) conf).toString();
this.clientPool = new HCatalogStoreClientPool(CLIENT_POOL_SIZE, conf);
@@ -188,9 +193,18 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
options.put(RCFILE_NULL, StringEscapeUtils.escapeJava(nullFormat));
String serde = properties.getProperty(serdeConstants.SERIALIZATION_LIB);
if (LazyBinaryColumnarSerDe.class.getName().equals(serde)) {
- options.put(RCFILE_SERDE, RCFILE_BINARY_SERDE);
+ options.put(RCFILE_SERDE, DEFAULT_BINARY_SERDE);
} else if (ColumnarSerDe.class.getName().equals(serde)) {
- options.put(RCFILE_SERDE, RCFILE_TEXT_SERDE);
+ options.put(RCFILE_SERDE, DEFAULT_TEXT_SERDE);
+ }
+ } else if (storeType.equals(CatalogProtos.StoreType.SEQUENCEFILE) ) {
+ options.put(SEQUENCEFILE_DELIMITER, StringEscapeUtils.escapeJava(fieldDelimiter));
+ options.put(SEQUENCEFILE_NULL, StringEscapeUtils.escapeJava(nullFormat));
+ String serde = properties.getProperty(serdeConstants.SERIALIZATION_LIB);
+ if (LazyBinarySerDe.class.getName().equals(serde)) {
+ options.put(SEQUENCEFILE_SERDE, DEFAULT_BINARY_SERDE);
+ } else if (LazySimpleSerDe.class.getName().equals(serde)) {
+ options.put(SEQUENCEFILE_SERDE, DEFAULT_TEXT_SERDE);
}
}
@@ -434,7 +448,7 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
String serde = tableDesc.getMeta().getOption(RCFILE_SERDE);
sd.setInputFormat(org.apache.hadoop.hive.ql.io.RCFileInputFormat.class.getName());
sd.setOutputFormat(org.apache.hadoop.hive.ql.io.RCFileOutputFormat.class.getName());
- if (RCFILE_TEXT_SERDE.equals(serde)) {
+ if (DEFAULT_TEXT_SERDE.equals(serde)) {
sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe.class.getName());
} else {
sd.getSerdeInfo().setSerializationLib(
@@ -450,7 +464,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(CSVFILE_DELIMITER, CSVFILE_DELIMITER_DEFAULT);
+ String fieldDelimiter = tableDesc.getMeta().getOption(CSVFILE_DELIMITER, DEFAULT_FIELD_DELIMITER);
// User can use an unicode for filed delimiter such as \u0001, \001.
// In this case, java console will convert this value into "\\u001".
@@ -466,6 +480,33 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
sd.getSerdeInfo().getParameters().put(serdeConstants.SERIALIZATION_NULL_FORMAT,
StringEscapeUtils.unescapeJava(tableDesc.getMeta().getOption(CSVFILE_NULL)));
}
+ } else if (tableDesc.getMeta().getStoreType().equals(CatalogProtos.StoreType.SEQUENCEFILE)) {
+ String serde = tableDesc.getMeta().getOption(SEQUENCEFILE_SERDE);
+ sd.setInputFormat(org.apache.hadoop.mapred.SequenceFileInputFormat.class.getName());
+ sd.setOutputFormat(org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat.class.getName());
+
+ if (DEFAULT_TEXT_SERDE.equals(serde)) {
+ sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe.class.getName());
+
+ String fieldDelimiter = tableDesc.getMeta().getOption(SEQUENCEFILE_DELIMITER, DEFAULT_FIELD_DELIMITER);
+
+ // User can use an unicode for filed delimiter such as \u0001, \001.
+ // In this case, java console will convert this value into "\\u001".
+ // And hive will un-espace this value again.
+ // As a result, user can use right field delimiter.
+ // So, we have to un-escape this value.
+ sd.getSerdeInfo().getParameters().put(serdeConstants.SERIALIZATION_FORMAT,
+ StringEscapeUtils.unescapeJava(fieldDelimiter));
+ sd.getSerdeInfo().getParameters().put(serdeConstants.FIELD_DELIM,
+ StringEscapeUtils.unescapeJava(fieldDelimiter));
+ } else {
+ sd.getSerdeInfo().setSerializationLib(org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe.class.getName());
+ }
+
+ if (tableDesc.getMeta().getOption(SEQUENCEFILE_NULL) != null) {
+ sd.getSerdeInfo().getParameters().put(serdeConstants.SERIALIZATION_NULL_FORMAT,
+ StringEscapeUtils.unescapeJava(tableDesc.getMeta().getOption(SEQUENCEFILE_NULL)));
+ }
} else {
throw new CatalogException(new NotImplementedException(tableDesc.getMeta().getStoreType().name()));
}
@@ -501,9 +542,6 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
@Override
public void alterTable(final CatalogProtos.AlterTableDescProto alterTableDescProto) throws CatalogException {
-
- HCatalogStoreClientPool.HCatalogStoreClient client = clientPool.getClient();
-
final String[] split = CatalogUtil.splitFQTableName(alterTableDescProto.getTableName());
if (split.length == 1) {
@@ -537,8 +575,6 @@ public class HCatalogStore extends CatalogConstants implements CatalogStore {
default:
//TODO
}
-
-
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 d426369..98aa7c5 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
@@ -21,6 +21,7 @@ import com.google.common.base.Preconditions;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat;
+import org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat;
import org.apache.hadoop.hive.ql.io.RCFileOutputFormat;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hcatalog.common.HCatException;
@@ -124,6 +125,8 @@ public class HCatalogUtil {
String outputFormatClass = fileFormatArrary[fileFormatArrary.length-1];
if(outputFormatClass.equals(HiveIgnoreKeyTextOutputFormat.class.getSimpleName())) {
return CatalogProtos.StoreType.CSV.name();
+ } else if(outputFormatClass.equals(HiveSequenceFileOutputFormat.class.getSimpleName())) {
+ return CatalogProtos.StoreType.SEQUENCEFILE.name();
} else if(outputFormatClass.equals(RCFileOutputFormat.class.getSimpleName())) {
return CatalogProtos.StoreType.RCFILE.name();
} else {
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 b9a734e..f4ca4d1 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
@@ -53,7 +53,6 @@ public class TestHCatalogStore {
private static HCatalogStore store;
private static Path warehousePath;
- private static HCatalogStoreClientPool pool;
@BeforeClass
public static void setUp() throws Exception {
@@ -105,7 +104,7 @@ public class TestHCatalogStore {
assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
}
- assertEquals(StringEscapeUtils.escapeJava(CatalogConstants.CSVFILE_DELIMITER_DEFAULT),
+ assertEquals(StringEscapeUtils.escapeJava(CatalogConstants.DEFAULT_FIELD_DELIMITER),
table1.getMeta().getOption(CatalogConstants.CSVFILE_DELIMITER));
store.dropTable(DB_NAME, CUSTOMER);
}
@@ -113,7 +112,7 @@ public class TestHCatalogStore {
@Test
public void testTableUsingRCFileWithBinarySerde() throws Exception {
Options options = new Options();
- options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.RCFILE_BINARY_SERDE);
+ options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.DEFAULT_BINARY_SERDE);
TableMeta meta = new TableMeta(CatalogProtos.StoreType.RCFILE, options);
org.apache.tajo.catalog.Schema schema = new org.apache.tajo.catalog.Schema();
@@ -134,7 +133,7 @@ public class TestHCatalogStore {
assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
}
- assertEquals(CatalogConstants.RCFILE_BINARY_SERDE,
+ assertEquals(CatalogConstants.DEFAULT_BINARY_SERDE,
table1.getMeta().getOption(CatalogConstants.RCFILE_SERDE));
store.dropTable(DB_NAME, REGION);
}
@@ -142,7 +141,7 @@ public class TestHCatalogStore {
@Test
public void testTableUsingRCFileWithTextSerde() throws Exception {
Options options = new Options();
- options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.RCFILE_TEXT_SERDE);
+ options.put(CatalogConstants.RCFILE_SERDE, CatalogConstants.DEFAULT_TEXT_SERDE);
TableMeta meta = new TableMeta(CatalogProtos.StoreType.RCFILE, options);
org.apache.tajo.catalog.Schema schema = new org.apache.tajo.catalog.Schema();
@@ -163,7 +162,7 @@ public class TestHCatalogStore {
assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
}
- assertEquals(CatalogConstants.RCFILE_TEXT_SERDE, table1.getMeta().getOption(CatalogConstants.RCFILE_SERDE));
+ assertEquals(CatalogConstants.DEFAULT_TEXT_SERDE, table1.getMeta().getOption(CatalogConstants.RCFILE_SERDE));
store.dropTable(DB_NAME, REGION);
}
@@ -301,4 +300,62 @@ public class TestHCatalogStore {
assertFalse(store.existTable(DB_NAME, tableName));
fs.close();
}
+
+ @Test
+ public void testTableUsingSequenceFileWithBinarySerde() throws Exception {
+ Options options = new Options();
+ options.put(CatalogConstants.SEQUENCEFILE_SERDE, CatalogConstants.DEFAULT_BINARY_SERDE);
+ TableMeta meta = new TableMeta(CatalogProtos.StoreType.SEQUENCEFILE, options);
+
+ org.apache.tajo.catalog.Schema schema = new org.apache.tajo.catalog.Schema();
+ schema.addColumn("r_regionkey", TajoDataTypes.Type.INT4);
+ schema.addColumn("r_name", TajoDataTypes.Type.TEXT);
+ schema.addColumn("r_comment", TajoDataTypes.Type.TEXT);
+
+ TableDesc table = new TableDesc(CatalogUtil.buildFQName(DB_NAME, REGION), schema, meta,
+ new Path(warehousePath, new Path(DB_NAME, REGION)));
+ store.createTable(table.getProto());
+ assertTrue(store.existTable(DB_NAME, REGION));
+
+ TableDesc table1 = new TableDesc(store.getTable(DB_NAME, REGION));
+ assertEquals(table.getName(), table1.getName());
+ assertEquals(table.getPath(), table1.getPath());
+ assertEquals(table.getSchema().size(), table1.getSchema().size());
+ for (int i = 0; i < table.getSchema().size(); i++) {
+ assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
+ }
+
+ assertEquals(CatalogConstants.DEFAULT_BINARY_SERDE,
+ table1.getMeta().getOption(CatalogConstants.SEQUENCEFILE_SERDE));
+ store.dropTable(DB_NAME, REGION);
+ }
+
+ @Test
+ public void testTableUsingSequenceFileWithTextSerde() throws Exception {
+ Options options = new Options();
+ options.put(CatalogConstants.SEQUENCEFILE_SERDE, CatalogConstants.DEFAULT_TEXT_SERDE);
+ TableMeta meta = new TableMeta(CatalogProtos.StoreType.SEQUENCEFILE, options);
+
+ org.apache.tajo.catalog.Schema schema = new org.apache.tajo.catalog.Schema();
+ schema.addColumn("r_regionkey", TajoDataTypes.Type.INT4);
+ schema.addColumn("r_name", TajoDataTypes.Type.TEXT);
+ schema.addColumn("r_comment", TajoDataTypes.Type.TEXT);
+
+ TableDesc table = new TableDesc(CatalogUtil.buildFQName(DB_NAME, REGION), schema, meta,
+ new Path(warehousePath, new Path(DB_NAME, REGION)));
+ store.createTable(table.getProto());
+ assertTrue(store.existTable(DB_NAME, REGION));
+
+ TableDesc table1 = new TableDesc(store.getTable(DB_NAME, REGION));
+ assertEquals(table.getName(), table1.getName());
+ assertEquals(table.getPath(), table1.getPath());
+ assertEquals(table.getSchema().size(), table1.getSchema().size());
+ for (int i = 0; i < table.getSchema().size(); i++) {
+ assertEquals(table.getSchema().getColumn(i).getSimpleName(), table1.getSchema().getColumn(i).getSimpleName());
+ }
+
+ assertEquals(CatalogConstants.DEFAULT_TEXT_SERDE, table1.getMeta().getOption(CatalogConstants.SEQUENCEFILE_SERDE));
+ store.dropTable(DB_NAME, REGION);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-common/src/main/java/org/apache/tajo/util/Bytes.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/util/Bytes.java b/tajo-common/src/main/java/org/apache/tajo/util/Bytes.java
index f9ba923..448274a 100644
--- a/tajo-common/src/main/java/org/apache/tajo/util/Bytes.java
+++ b/tajo-common/src/main/java/org/apache/tajo/util/Bytes.java
@@ -207,6 +207,11 @@ public class Bytes {
return offset + srcLength;
}
+ public static void writeVLong(ByteArrayOutputStream byteStream, long l) {
+ byte[] vLongBytes = Bytes.vlongToBytes(l);
+ byteStream.write(vLongBytes, 0, vLongBytes.length);
+ }
+
/**
* Put bytes at the specified byte array position.
* @param tgtBytes the byte array
@@ -993,6 +998,34 @@ public class Bytes {
}
/**
+ * Reads a zero-compressed encoded int from input stream and returns it.
+ * @param buffer Binary array
+ * @param offset Offset into array at which vint begins.
+ * @throws java.io.IOException e
+ * @return deserialized long from stream.
+ */
+ public static int readVInt(final byte [] buffer, final int offset)
+ throws IOException {
+ byte firstByte = buffer[offset];
+ int length = (byte) WritableUtils.decodeVIntSize(firstByte);
+ if (length == 1) {
+ return firstByte;
+ }
+ int i = 0;
+ for (int idx = 0; idx < length - 1; idx++) {
+ byte b = buffer[offset + 1 + idx];
+ i = i << 8;
+ i = i | (b & 0xFF);
+ }
+ return (WritableUtils.isNegativeVInt(firstByte) ? (i ^ -1) : i);
+ }
+
+ public static byte getVIntSize(byte[] bytes, int offset) {
+ byte firstByte = bytes[offset];
+ return (byte) WritableUtils.decodeVIntSize(firstByte);
+ }
+
+ /**
* @param left left operand
* @param right right operand
* @return 0 if equal, < 0 if left is less than right, etc.
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/benchmark/TPCH.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/benchmark/TPCH.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/benchmark/TPCH.java
index 20015b5..5dd0b12 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/benchmark/TPCH.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/benchmark/TPCH.java
@@ -189,7 +189,7 @@ public class TPCH extends BenchmarkSet {
private void loadTable(String tableName) throws ServiceException {
TableMeta meta = CatalogUtil.newTableMeta(StoreType.CSV);
- meta.putOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ meta.putOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
try {
tajo.createExternalTable(tableName, getSchema(tableName), new Path(dataDir, tableName), meta);
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TpchTestBase.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TpchTestBase.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TpchTestBase.java
index 0b80231..8487e84 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TpchTestBase.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TpchTestBase.java
@@ -91,7 +91,7 @@ public class TpchTestBase {
private void setUp() throws Exception {
util = new LocalTajoTestingUtility();
Options opt = new Options();
- opt.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ opt.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
util.setup(names, paths, schemas, opt);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/client/TestDDLBuilder.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/client/TestDDLBuilder.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/client/TestDDLBuilder.java
index 73f7d5e..9d8d631 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/client/TestDDLBuilder.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/client/TestDDLBuilder.java
@@ -43,7 +43,7 @@ public class TestDDLBuilder {
schema1.addColumn("addr", TajoDataTypes.Type.TEXT);
meta1 = CatalogUtil.newTableMeta(CatalogProtos.StoreType.CSV);
- meta1.putOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ meta1.putOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
meta1.putOption(CatalogConstants.COMPRESSION_CODEC, GzipCodec.class.getName());
Schema expressionSchema = new Schema();
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
index 72e92d2..ad93738 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestCTASQuery.java
@@ -212,13 +212,22 @@ public class TestCTASQuery extends QueryTestCaseBase {
@Test
public final void testCtasWithManagedTable() throws Exception {
- if (!testingCluster.isHCatalogStoreRunning()) {
- ResultSet res = executeFile("CtasWithManagedTable.sql");
- res.close();
+ ResultSet res = executeFile("CtasWithManagedTable.sql");
+ res.close();
+
+ if (testingCluster.isHCatalogStoreRunning()) {
+ assertTrue(client.existTable("managed_table1"));
+
+ TableDesc desc = client.getTableDesc("managed_table1");
+ assertNotNull(desc);
+ assertEquals("managed_table1", desc.getPath().getName());
+ } else {
assertFalse(client.existTable("MANAGED_TABLE1"));
assertTrue(client.existTable("\"MANAGED_TABLE1\""));
+
TableDesc desc = client.getTableDesc("\"MANAGED_TABLE1\"");
+
assertNotNull(desc);
assertEquals("MANAGED_TABLE1", desc.getPath().getName());
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestNullValues.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
index 1f1d9e1..c460bcf 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestNullValues.java
@@ -52,7 +52,7 @@ public class TestNullValues {
"3|filled|0.2"
};
Options opts = new Options();
- opts.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ opts.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable1 where col3 is null");
@@ -78,7 +78,7 @@ public class TestNullValues {
"3|filled|"
};
Options opts = new Options();
- opts.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ opts.put(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
ResultSet res = TajoTestingCluster
.run(table, schemas, opts, new String[][]{data},
"select * from nulltable2 where col1 is not null");
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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
new file mode 100644
index 0000000..c11a343
--- /dev/null
+++ b/tajo-docs/src/main/sphinx/table_management/csv.rst
@@ -0,0 +1,6 @@
+*************************************
+CSV
+*************************************
+
+
+(TODO)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-docs/src/main/sphinx/table_management/file_formats.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/table_management/file_formats.rst b/tajo-docs/src/main/sphinx/table_management/file_formats.rst
index 26edac4..c15dd3f 100644
--- a/tajo-docs/src/main/sphinx/table_management/file_formats.rst
+++ b/tajo-docs/src/main/sphinx/table_management/file_formats.rst
@@ -2,12 +2,12 @@
File Formats
*************************************
-.. todo::
+Currently, Tajo provides four file formats as follows:
-===============================
-Text File
-===============================
+.. toctree::
+ :maxdepth: 1
-===============================
-RCFile
-===============================
\ No newline at end of file
+ csv
+ rcfile
+ parquet
+ sequencefile
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-docs/src/main/sphinx/table_management/parquet.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/table_management/parquet.rst b/tajo-docs/src/main/sphinx/table_management/parquet.rst
new file mode 100644
index 0000000..a994b7e
--- /dev/null
+++ b/tajo-docs/src/main/sphinx/table_management/parquet.rst
@@ -0,0 +1,6 @@
+*************************************
+Parquet
+*************************************
+
+
+(TODO)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-docs/src/main/sphinx/table_management/rcfile.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/table_management/rcfile.rst b/tajo-docs/src/main/sphinx/table_management/rcfile.rst
new file mode 100644
index 0000000..21f8253
--- /dev/null
+++ b/tajo-docs/src/main/sphinx/table_management/rcfile.rst
@@ -0,0 +1,6 @@
+*************************************
+RCFIle
+*************************************
+
+
+(TODO)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-docs/src/main/sphinx/table_management/sequencefile.rst
----------------------------------------------------------------------
diff --git a/tajo-docs/src/main/sphinx/table_management/sequencefile.rst b/tajo-docs/src/main/sphinx/table_management/sequencefile.rst
new file mode 100644
index 0000000..436dfa5
--- /dev/null
+++ b/tajo-docs/src/main/sphinx/table_management/sequencefile.rst
@@ -0,0 +1,111 @@
+*************************************
+SequenceFile
+*************************************
+
+-----------------------------------------
+Introduce
+-----------------------------------------
+
+SequenceFiles are flat files consisting of binary key/value pairs.
+SequenceFile is basic file format which provided by Hadoop, and Hive also provides it to create a table.
+
+The ``USING sequencefile`` keywords let you create a SequecneFile. Here is an example statement to create a table using ``SequecneFile``:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name text, score float, type text)
+ USING sequencefile;
+
+Also Tajo provides Hive compatibility for SequenceFile. The above statement can be written in Hive as follows:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name string, score float, type string)
+ STORED AS sequencefile;
+
+-----------------------------------------
+SerializerDeserializer (SerDe)
+-----------------------------------------
+
+There are two SerDe for SequenceFile as follows:
+
+ + TextSerializerDeserializer: This class can read and write data in plain text file format.
+ + BinarySerializerDeserializer: This class can read and write data in binary file format.
+
+The default is the SerDe for plain text file in Tajo. The above example statement created the table using TextSerializerDeserializer.If you want to use BinarySerializerDeserializer, you can specify it by ``sequencefile.serde`` keywords:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name text, score float, type text)
+ USING sequencefile with ('sequencefile.serde'='org.apache.tajo.storage.BinarySerializerDeserializer')
+
+In Hive, the above statement can be written in Hive as follows:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name string, score float, type string)
+ ROW FORMAT SERDE
+ 'org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe'
+ STORED AS sequencefile;
+
+-----------------------------------------
+Writer
+-----------------------------------------
+
+There are three SequenceFile Writers based on the SequenceFile.CompressionType used to compress key/value pairs:
+
+ + Writer : Uncompressed records.
+ + RecordCompressWriter : Record-compressed files, only compress values.
+ + BlockCompressWriter : Block-compressed files, both keys & values are collected in 'blocks' separately and compressed. The size of the 'block' is configurable.
+
+The default is Uncompressed Writer in Tajo. If you want to use RecordCompressWriter, you can specify it by ``compression.type`` keywords and ``compression.codec`` keywords:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name text, score float, type text)
+ USING sequencefile with ('compression.type'='RECORD','compression.codec'='org.apache.hadoop.io.compress.SnappyCodec')
+
+In hive, you need to specify settings as follows:
+
+.. code-block:: sql
+
+ hive> SET hive.exec.compress.output = true;
+ hive> SET mapred.output.compression.type = RECORD;
+ hive> SET mapred.output.compression.codec = org.apache.hadoop.io.compress.SnappyCodec;
+ hive> CREATE TABLE table1 (id int, name string, score float, type string) STORED AS sequencefile;;
+
+And if you want to use BlockCompressWriter, you can specify it by ``compression.type`` keywords and ``compression.codec`` keywords:
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name text, score float, type text)
+ USING sequencefile with ('compression.type'='BLOCK','compression.codec'='org.apache.hadoop.io.compress.SnappyCodec')
+
+In hive, you need to specify settings as follows:
+
+.. code-block:: sql
+
+ hive> SET hive.exec.compress.output = true;
+ hive> SET mapred.output.compression.type = BLOCK;
+ hive> SET mapred.output.compression.codec = org.apache.hadoop.io.compress.SnappyCodec;
+ hive> CREATE TABLE table1 (id int, name string, score float, type string) STORED AS sequencefile;;
+
+For reference, you can use TextSerDe or BinarySerDe with compression keywords.
+Here is an example statement for this case.
+
+.. code-block:: sql
+
+ CREATE TABLE table1 (id int, name text, score float, type text)
+ USING sequencefile with ('sequencefile.serde'='org.apache.tajo.storage.BinarySerializerDeserializer', 'compression.type'='BLOCK','compression.codec'='org.apache.hadoop.io.compress.SnappyCodec')
+
+In hive, you need to specify settings as follows:
+
+.. code-block:: sql
+
+ hive> SET hive.exec.compress.output = true;
+ hive> SET mapred.output.compression.type = BLOCK;
+ hive> SET mapred.output.compression.codec = org.apache.hadoop.io.compress.SnappyCodec;
+ hive> CREATE TABLE table1 (id int, name string, score float, type string)
+ ROW FORMAT SERDE
+ 'org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe'
+ STORED AS sequencefile;;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-storage/pom.xml b/tajo-storage/pom.xml
index 9f144bb..c3f08cb 100644
--- a/tajo-storage/pom.xml
+++ b/tajo-storage/pom.xml
@@ -63,6 +63,17 @@
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.12.4</version>
+ <configuration>
+ <systemProperties>
+ <tajo.test>TRUE</tajo.test>
+ </systemProperties>
+ <argLine>-Xms512m -Xmx1024m -XX:MaxPermSize=128m -Dfile.encoding=UTF-8</argLine>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>2.4</version>
<executions>
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 116e25c..857278a 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
@@ -83,7 +83,7 @@ public class CSVFile {
this.meta = meta;
this.schema = schema;
this.delimiter = StringEscapeUtils.unescapeJava(this.meta.getOption(CatalogConstants.CSVFILE_DELIMITER,
- CatalogConstants.CSVFILE_DELIMITER_DEFAULT)).charAt(0);
+ CatalogConstants.DEFAULT_FIELD_DELIMITER)).charAt(0);
this.columnNum = schema.size();
String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(CatalogConstants.CSVFILE_NULL));
if (StringUtils.isEmpty(nullCharacters)) {
@@ -260,7 +260,7 @@ public class CSVFile {
}
//Delimiter
- String delim = meta.getOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.CSVFILE_DELIMITER_DEFAULT);
+ String delim = meta.getOption(CatalogConstants.CSVFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
this.delimiter = StringEscapeUtils.unescapeJava(delim).charAt(0);
String nullCharacters = StringEscapeUtils.unescapeJava(meta.getOption(CatalogConstants.CSVFILE_NULL));
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoRecordMaterializer.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoRecordMaterializer.java b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoRecordMaterializer.java
index cce7ead..f099d04 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoRecordMaterializer.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/parquet/TajoRecordMaterializer.java
@@ -50,7 +50,7 @@ class TajoRecordMaterializer extends RecordMaterializer<Tuple> {
Column[] targets = projection.toArray();
int[] projectionMap = new int[targets.length];
for (int i = 0; i < targets.length; ++i) {
- int tid = schema.getColumnIdByName(targets[i].getSimpleName());
+ int tid = schema.getColumnId(targets[i].getQualifiedName());
projectionMap[i] = tid;
}
return projectionMap;
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java b/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
index 2fd3455..bbb9df1 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java
@@ -1206,7 +1206,7 @@ public class RCFile {
targetColumnIndexes = new int[targets.length];
for (int i = 0; i < targets.length; i++) {
- targetColumnIndexes[i] = schema.getColumnIdByName(targets[i].getSimpleName());
+ targetColumnIndexes[i] = schema.getColumnId(targets[i].getQualifiedName());
}
Arrays.sort(targetColumnIndexes);
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java b/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
new file mode 100644
index 0000000..8092c0c
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
@@ -0,0 +1,268 @@
+/**
+ * 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.sequencefile;
+
+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.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.tajo.catalog.CatalogConstants;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.datum.ProtobufDatum;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.exception.AlreadyExistsStorageException;
+import org.apache.tajo.storage.rcfile.NonSyncByteArrayOutputStream;
+import org.apache.tajo.util.Bytes;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+public class SequenceFileAppender extends FileAppender {
+ private static final Log LOG = LogFactory.getLog(SequenceFileScanner.class);
+
+ private SequenceFile.Writer writer;
+
+ private TableMeta meta;
+ private Schema schema;
+ private TableStatistics stats = null;
+
+ private int columnNum;
+ private FileSystem fs;
+ private char delimiter;
+ private byte[] nullChars;
+
+ private final static int BUFFER_SIZE = 128 * 1024;
+ private long pos = 0;
+
+ private CompressionCodecFactory codecFactory;
+ private CompressionCodec codec;
+
+ private NonSyncByteArrayOutputStream os;
+ private SerializerDeserializer serde;
+
+ long rowCount;
+ private boolean isShuffle;
+ private static final BytesWritable EMPTY_KEY = new BytesWritable();
+
+ public SequenceFileAppender(Configuration conf, Schema schema, TableMeta meta, Path path) throws IOException {
+ super(conf, schema, meta, path);
+ this.meta = meta;
+ this.schema = schema;
+ }
+
+ @Override
+ public void init() throws IOException {
+ os = new NonSyncByteArrayOutputStream(BUFFER_SIZE);
+
+ this.fs = path.getFileSystem(conf);
+
+ //determine the intermediate file type
+ String store = conf.get(TajoConf.ConfVars.SHUFFLE_FILE_FORMAT.varname,
+ TajoConf.ConfVars.SHUFFLE_FILE_FORMAT.defaultVal);
+ if (enabledStats && CatalogProtos.StoreType.SEQUENCEFILE == CatalogProtos.StoreType.valueOf(store.toUpperCase())) {
+ isShuffle = true;
+ } else {
+ isShuffle = false;
+ }
+
+ this.delimiter = StringEscapeUtils.unescapeJava(this.meta.getOption(CatalogConstants.SEQUENCEFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER)).charAt(0);
+ this.columnNum = schema.size();
+ String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(CatalogConstants.SEQUENCEFILE_NULL));
+ if (StringUtils.isEmpty(nullCharacters)) {
+ nullChars = NullDatum.get().asTextBytes();
+ } else {
+ nullChars = nullCharacters.getBytes();
+ }
+
+ if (!fs.exists(path.getParent())) {
+ throw new FileNotFoundException(path.toString());
+ }
+
+ String codecName = this.meta.getOption(CatalogConstants.COMPRESSION_CODEC);
+ if(!StringUtils.isEmpty(codecName)){
+ codecFactory = new CompressionCodecFactory(conf);
+ codec = codecFactory.getCodecByClassName(codecName);
+ } else {
+ if (fs.exists(path)) {
+ throw new AlreadyExistsStorageException(path);
+ }
+ }
+
+ try {
+ String serdeClass = this.meta.getOption(CatalogConstants.SEQUENCEFILE_SERDE, TextSerializerDeserializer.class.getName());
+ serde = (SerializerDeserializer) Class.forName(serdeClass).newInstance();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e);
+ }
+
+ Class<? extends Writable> valueClass;
+ if (serde instanceof BinarySerializerDeserializer) {
+ valueClass = BytesWritable.class;
+ } else {
+ valueClass = Text.class;
+ }
+
+ String type = this.meta.getOption(CatalogConstants.COMPRESSION_TYPE, CompressionType.NONE.name());
+ if (type.equals(CompressionType.BLOCK.name())) {
+ writer = SequenceFile.createWriter(fs, conf, path, BytesWritable.class, valueClass, CompressionType.BLOCK, codec);
+ } else if (type.equals(CompressionType.RECORD.name())) {
+ writer = SequenceFile.createWriter(fs, conf, path, BytesWritable.class, valueClass, CompressionType.RECORD, codec);
+ } else {
+ writer = SequenceFile.createWriter(fs, conf, path, BytesWritable.class, valueClass, CompressionType.NONE, codec);
+ }
+
+ if (enabledStats) {
+ this.stats = new TableStatistics(this.schema);
+ }
+
+ super.init();
+ }
+
+ @Override
+ public void addTuple(Tuple tuple) throws IOException {
+ Datum datum;
+
+ if (serde instanceof BinarySerializerDeserializer) {
+ byte nullByte = 0;
+ int lasti = 0;
+ for (int i = 0; i < columnNum; i++) {
+ datum = tuple.get(i);
+
+ // set bit to 1 if a field is not null
+ if (null != datum) {
+ nullByte |= 1 << (i % 8);
+ }
+
+ // write the null byte every eight elements or
+ // if this is the last element and serialize the
+ // corresponding 8 struct fields at the same time
+ if (7 == i % 8 || i == columnNum - 1) {
+ os.write(nullByte);
+
+ for (int j = lasti; j <= i; j++) {
+ datum = tuple.get(j);
+
+ switch (schema.getColumn(j).getDataType().getType()) {
+ case TEXT:
+ Bytes.writeVLong(os, datum.asTextBytes().length);
+ break;
+ case PROTOBUF:
+ ProtobufDatum protobufDatum = (ProtobufDatum) datum;
+ Bytes.writeVLong(os, protobufDatum.asByteArray().length);
+ break;
+ case CHAR:
+ case INET4:
+ case BLOB:
+ Bytes.writeVLong(os, datum.asByteArray().length);
+ break;
+ default:
+ }
+
+ serde.serialize(schema.getColumn(j), datum, os, nullChars);
+
+ if (isShuffle) {
+ // it is to calculate min/max values, and it is only used for the intermediate file.
+ stats.analyzeField(j, datum);
+ }
+ }
+ lasti = i + 1;
+ nullByte = 0;
+ }
+ }
+
+ BytesWritable b = new BytesWritable();
+ b.set(os.getData(), 0, os.getLength());
+ writer.append(EMPTY_KEY, b);
+
+ } else {
+ for (int i = 0; i < columnNum; i++) {
+ datum = tuple.get(i);
+ serde.serialize(schema.getColumn(i), datum, os, nullChars);
+
+ if (columnNum -1 > i) {
+ os.write((byte) delimiter);
+ }
+
+ if (isShuffle) {
+ // it is to calculate min/max values, and it is only used for the intermediate file.
+ stats.analyzeField(i, datum);
+ }
+
+ }
+ writer.append(EMPTY_KEY, new Text(os.toByteArray()));
+ }
+
+ os.reset();
+ pos += writer.getLength();
+ rowCount++;
+
+ if (enabledStats) {
+ stats.incrementRow();
+ }
+ }
+
+ @Override
+ public long getOffset() throws IOException {
+ return pos;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ os.flush();
+ writer.close();
+ }
+
+ @Override
+ public void close() throws IOException {
+ // Statistical section
+ if (enabledStats) {
+ stats.setNumBytes(getOffset());
+ }
+
+ os.close();
+ writer.close();
+ }
+
+ @Override
+ public TableStats getStats() {
+ if (enabledStats) {
+ return stats.getTableStat();
+ } else {
+ return null;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java b/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java
new file mode 100644
index 0000000..c5720e3
--- /dev/null
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java
@@ -0,0 +1,330 @@
+/**
+ * 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.sequencefile;
+
+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.FileSystem;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.tajo.catalog.CatalogConstants;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.storage.*;
+import org.apache.tajo.storage.fragment.FileFragment;
+import org.apache.tajo.util.Bytes;
+
+import java.io.IOException;
+
+public class SequenceFileScanner extends FileScanner {
+ private static final Log LOG = LogFactory.getLog(SequenceFileScanner.class);
+
+ private FileSystem fs;
+ private SequenceFile.Reader reader;
+ private SerializerDeserializer serde;
+ private byte[] nullChars;
+ private char delimiter;
+
+ private int currentIdx = 0;
+ private int[] projectionMap;
+
+ private boolean hasBinarySerDe = false;
+ private long totalBytes = 0L;
+
+ private long start, end;
+ private boolean more = true;
+
+ /**
+ * Whether a field is null or not. Because length is 0 does not means the
+ * field is null. In particular, a 0-length string is not null.
+ */
+ private boolean[] fieldIsNull;
+
+ /**
+ * The start positions and lengths of fields. Only valid when the data is parsed.
+ */
+ private int[] fieldStart;
+ private int[] fieldLength;
+
+ private int elementOffset, elementSize;
+
+ private static final BytesWritable EMPTY_KEY = new BytesWritable();
+
+ public SequenceFileScanner(Configuration conf, Schema schema, TableMeta meta, FileFragment fragment) throws IOException {
+ super(conf, schema, meta, fragment);
+ }
+
+ @Override
+ public void init() throws IOException {
+ // FileFragment information
+ if(fs == null) {
+ fs = FileScanner.getFileSystem((TajoConf)conf, fragment.getPath());
+ }
+
+ reader = new SequenceFile.Reader(fs, fragment.getPath(), conf);
+
+ String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(CatalogConstants.SEQUENCEFILE_NULL));
+ if (StringUtils.isEmpty(nullCharacters)) {
+ nullChars = NullDatum.get().asTextBytes();
+ } else {
+ nullChars = nullCharacters.getBytes();
+ }
+
+ String delim = meta.getOption(CatalogConstants.SEQUENCEFILE_DELIMITER, CatalogConstants.DEFAULT_FIELD_DELIMITER);
+ this.delimiter = StringEscapeUtils.unescapeJava(delim).charAt(0);
+
+ this.start = fragment.getStartKey();
+ this.end = start + fragment.getEndKey();
+
+ if (fragment.getStartKey() > reader.getPosition())
+ reader.sync(this.start);
+
+ more = start < end;
+
+ if (targets == null) {
+ targets = schema.toArray();
+ }
+
+
+ fieldIsNull = new boolean[schema.getColumns().size()];
+ fieldStart = new int[schema.getColumns().size()];
+ fieldLength = new int[schema.getColumns().size()];
+
+ prepareProjection(targets);
+
+ try {
+ String serdeClass = this.meta.getOption(CatalogConstants.SEQUENCEFILE_SERDE, TextSerializerDeserializer.class.getName());
+ serde = (SerializerDeserializer) Class.forName(serdeClass).newInstance();
+
+ if (serde instanceof BinarySerializerDeserializer)
+ hasBinarySerDe = true;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new IOException(e);
+ }
+ super.init();
+ }
+
+ private void prepareProjection(Column [] targets) {
+ projectionMap = new int[targets.length];
+
+ int tid;
+ for (int i = 0; i < targets.length; i++) {
+ tid = schema.getColumnId(targets[i].getQualifiedName());
+ projectionMap[i] = tid;
+ }
+ }
+
+ @Override
+ public Tuple next() throws IOException {
+ if (!more) return null;
+
+ long pos = reader.getPosition();
+ boolean remaining = reader.next(EMPTY_KEY);
+
+ if (pos >= end && reader.syncSeen()) {
+ more = false;
+ } else {
+ more = remaining;
+ }
+
+ if (more) {
+ Tuple tuple = null;
+ byte[][] cells;
+
+ if (hasBinarySerDe) {
+ BytesWritable bytesWritable = new BytesWritable();
+ reader.getCurrentValue(bytesWritable);
+ tuple = makeTuple(bytesWritable);
+ totalBytes += (long)bytesWritable.getBytes().length;
+ } else {
+ Text text = new Text();
+ reader.getCurrentValue(text);
+ cells = Bytes.splitPreserveAllTokens(text.getBytes(), delimiter, projectionMap);
+ totalBytes += (long)text.getBytes().length;
+ tuple = new LazyTuple(schema, cells, 0, nullChars, serde);
+ }
+ currentIdx++;
+ return tuple;
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * In hive, LazyBinarySerDe is serialized as follows: start A B A B A B end bytes[] ->
+ * |-----|---------|--- ... ---|-----|---------|
+ *
+ * Section A is one null-byte, corresponding to eight struct fields in Section
+ * B. Each bit indicates whether the corresponding field is null (0) or not null
+ * (1). Each field is a LazyBinaryObject.
+ *
+ * Following B, there is another section A and B. This pattern repeats until the
+ * all struct fields are serialized.
+ *
+ * So, tajo must make a tuple after parsing hive style BinarySerDe.
+ */
+ private Tuple makeTuple(BytesWritable value) throws IOException{
+ Tuple tuple = new VTuple(schema.getColumns().size());
+
+ int start = 0;
+ int length = value.getLength();
+
+ /**
+ * Please note that one null byte is followed by eight fields, then more
+ * null byte and fields.
+ */
+ int structByteEnd = start + length;
+ byte[] bytes = value.getBytes();
+
+ byte nullByte = bytes[start];
+ int lastFieldByteEnd = start + 1;
+
+ // Go through all bytes in the byte[]
+ for (int i = 0; i < schema.getColumns().size(); i++) {
+ fieldIsNull[i] = true;
+ if ((nullByte & (1 << (i % 8))) != 0) {
+ fieldIsNull[i] = false;
+ parse(schema.getColumn(i), bytes, lastFieldByteEnd);
+
+ fieldStart[i] = lastFieldByteEnd + elementOffset;
+ fieldLength[i] = elementSize;
+ lastFieldByteEnd = fieldStart[i] + fieldLength[i];
+
+ for (int j = 0; j < projectionMap.length; j++) {
+ if (projectionMap[j] == i) {
+ Datum datum = serde.deserialize(schema.getColumn(i), bytes, fieldStart[i], fieldLength[i], nullChars);
+ tuple.put(i, datum);
+ }
+ }
+ }
+
+ // next byte is a null byte if there are more bytes to go
+ if (7 == (i % 8)) {
+ if (lastFieldByteEnd < structByteEnd) {
+ nullByte = bytes[lastFieldByteEnd];
+ lastFieldByteEnd++;
+ } else {
+ // otherwise all null afterwards
+ nullByte = 0;
+ lastFieldByteEnd++;
+ }
+ }
+ }
+
+ return tuple;
+ }
+
+ /**
+ * Check a particular field and set its size and offset in bytes based on the
+ * field type and the bytes arrays.
+ *
+ * For void, boolean, byte, short, int, long, float and double, there is no
+ * offset and the size is fixed. For string, the first four bytes are used to store the size.
+ * So the offset is 4 and the size is computed by concating the first four bytes together.
+ * The first four bytes are defined with respect to the offset in the bytes arrays.
+ *
+ * @param col
+ * catalog column information
+ * @param bytes
+ * bytes arrays store the table row
+ * @param offset
+ * offset of this field
+ */
+ private void parse(Column col, byte[] bytes, int offset) throws
+ IOException {
+ switch (col.getDataType().getType()) {
+ case BOOLEAN:
+ case BIT:
+ elementOffset = 0;
+ elementSize = 1;
+ break;
+ case INT2:
+ elementOffset = 0;
+ elementSize = 2;
+ break;
+ case INT4:
+ case INT8:
+ elementOffset = 0;
+ elementSize = WritableUtils.decodeVIntSize(bytes[offset]);
+ break;
+ case FLOAT4:
+ elementOffset = 0;
+ elementSize = 4;
+ break;
+ case FLOAT8:
+ elementOffset = 0;
+ elementSize = 8;
+ break;
+ case BLOB:
+ case PROTOBUF:
+ case INET4:
+ case CHAR:
+ case TEXT:
+ elementOffset = 1;
+ elementSize = bytes[offset];
+ break;
+ default:
+ elementOffset = 0;
+ elementSize = 0;
+ }
+ }
+
+ @Override
+ public void reset() throws IOException {
+ if (reader != null) {
+ reader.sync(0);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null)
+ reader.close();
+
+ if (tableStats != null) {
+ tableStats.setReadBytes(totalBytes);
+ tableStats.setNumRows(currentIdx);
+ }
+ }
+
+ @Override
+ public boolean isProjectable() {
+ return true;
+ }
+
+ @Override
+ public boolean isSelectable() {
+ return true;
+ }
+
+ @Override
+ public boolean isSplittable(){
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java b/tajo-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
index ed71730..36a589a 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
@@ -71,7 +71,7 @@ public class TrevniScanner extends FileScanner {
projectionMap = new int[targets.length];
int tid;
for (int i = 0; i < targets.length; i++) {
- tid = schema.getColumnIdByName(targets[i].getSimpleName());
+ tid = schema.getColumnId(targets[i].getQualifiedName());
projectionMap[i] = tid;
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/main/java/org/apache/tajo/storage/v2/CSVFileScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/main/java/org/apache/tajo/storage/v2/CSVFileScanner.java b/tajo-storage/src/main/java/org/apache/tajo/storage/v2/CSVFileScanner.java
index 79ddfdd..19209bd 100644
--- a/tajo-storage/src/main/java/org/apache/tajo/storage/v2/CSVFileScanner.java
+++ b/tajo-storage/src/main/java/org/apache/tajo/storage/v2/CSVFileScanner.java
@@ -136,7 +136,7 @@ public class CSVFileScanner extends FileScannerV2 {
targetColumnIndexes = new int[targets.length];
for (int i = 0; i < targets.length; i++) {
- targetColumnIndexes[i] = schema.getColumnIdByName(targets[i].getSimpleName());
+ targetColumnIndexes[i] = schema.getColumnId(targets[i].getQualifiedName());
}
if (LOG.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 e3a0eaf..ae1380c 100644
--- a/tajo-storage/src/main/resources/storage-default.xml
+++ b/tajo-storage/src/main/resources/storage-default.xml
@@ -40,7 +40,7 @@
<!--- Registered Scanner Handler -->
<property>
<name>tajo.storage.scanner-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet</value>
+ <value>csv,raw,rcfile,row,trevni,parquet,sequencefile</value>
</property>
<!--- Fragment Class Configurations -->
@@ -68,6 +68,10 @@
<name>tajo.storage.fragment.parquet.class</name>
<value>org.apache.tajo.storage.fragment.FileFragment</value>
</property>
+ <property>
+ <name>tajo.storage.fragment.sequencefile.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
<!--- Scanner Handler -->
<property>
@@ -130,10 +134,20 @@
<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>
+
<!--- Appender Handler -->
<property>
<name>tajo.storage.appender-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet</value>
+ <value>csv,raw,rcfile,row,trevni,parquet,sequencefile</value>
</property>
<property>
@@ -165,4 +179,9 @@
<name>tajo.storage.appender-handler.parquet.class</name>
<value>org.apache.tajo.storage.parquet.ParquetAppender</value>
</property>
+
+ <property>
+ <name>tajo.storage.appender-handler.sequencefile.class</name>
+ <value>org.apache.tajo.storage.sequencefile.SequenceFileAppender</value>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 a776eb6..9fe5721 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
@@ -21,6 +21,7 @@ package org.apache.tajo.storage;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.*;
import org.apache.hadoop.io.compress.zlib.ZlibFactory;
import org.apache.hadoop.util.NativeCodeLoader;
@@ -65,7 +66,8 @@ public class TestCompressionStorages {
public static Collection<Object[]> generateParameters() {
return Arrays.asList(new Object[][]{
{StoreType.CSV},
- {StoreType.RCFILE}
+ {StoreType.RCFILE},
+ {StoreType.SEQUENCEFILE}
});
}
@@ -80,6 +82,10 @@ public class TestCompressionStorages {
if( ZlibFactory.isNativeZlibLoaded(conf)) {
storageCompressionTest(storeType, GzipCodec.class);
}
+ } else if (storeType == StoreType.SEQUENCEFILE) {
+ if( ZlibFactory.isNativeZlibLoaded(conf)) {
+ storageCompressionTest(storeType, GzipCodec.class);
+ }
} else {
storageCompressionTest(storeType, GzipCodec.class);
}
@@ -176,7 +182,9 @@ public class TestCompressionStorages {
TableMeta meta = CatalogUtil.newTableMeta(storeType);
meta.putOption("compression.codec", codec.getCanonicalName());
+ meta.putOption("compression.type", SequenceFile.CompressionType.BLOCK.name());
meta.putOption("rcfile.serde", TextSerializerDeserializer.class.getName());
+ meta.putOption("sequencefile.serde", TextSerializerDeserializer.class.getName());
String fileName = "Compression_" + codec.getSimpleName();
Path tablePath = new Path(testDir, fileName);
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/tajo-storage/src/test/java/org/apache/tajo/storage/TestMergeScanner.java
----------------------------------------------------------------------
diff --git a/tajo-storage/src/test/java/org/apache/tajo/storage/TestMergeScanner.java b/tajo-storage/src/test/java/org/apache/tajo/storage/TestMergeScanner.java
index ad12c06..5427592 100644
--- a/tajo-storage/src/test/java/org/apache/tajo/storage/TestMergeScanner.java
+++ b/tajo-storage/src/test/java/org/apache/tajo/storage/TestMergeScanner.java
@@ -67,6 +67,7 @@ public class TestMergeScanner {
{StoreType.RCFILE},
{StoreType.TREVNI},
{StoreType.PARQUET},
+ {StoreType.SEQUENCEFILE},
// RowFile requires Byte-buffer read support, so we omitted RowFile.
//{StoreType.ROWFILE},
@@ -172,6 +173,7 @@ public class TestMergeScanner {
case RCFILE:
case TREVNI:
case PARQUET:
+ case SEQUENCEFILE:
case CSV:
return true;
default:
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 b5338ff..0a38985 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
@@ -80,6 +80,7 @@ public class TestStorages {
{StoreType.RAW, false, false},
{StoreType.RCFILE, true, true},
{StoreType.PARQUET, false, false},
+ {StoreType.SEQUENCEFILE, true, true},
});
}
@@ -175,7 +176,8 @@ public class TestStorages {
if (storeType == StoreType.RCFILE
|| storeType == StoreType.TREVNI
|| storeType == StoreType.CSV
- || storeType == StoreType.PARQUET) {
+ || storeType == StoreType.PARQUET
+ || storeType == StoreType.SEQUENCEFILE) {
assertTrue(tuple.get(0) == null);
}
assertTrue(tupleCnt + 2 == tuple.get(1).asInt8());
@@ -270,6 +272,7 @@ public class TestStorages {
meta.putOption(CatalogConstants.CSVFILE_NULL, "\\\\N");
meta.putOption(CatalogConstants.RCFILE_NULL, "\\\\N");
meta.putOption(CatalogConstants.RCFILE_SERDE, TextSerializerDeserializer.class.getName());
+ meta.putOption(CatalogConstants.SEQUENCEFILE_NULL, "\\");
Path tablePath = new Path(testDir, "testVariousTypes.data");
Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(meta, schema, tablePath);
@@ -472,6 +475,144 @@ public class TestStorages {
}
@Test
+ public void testSequenceFileTextSerializeDeserialize() throws IOException {
+ if(storeType != StoreType.SEQUENCEFILE) return;
+
+ Schema schema = new Schema();
+ schema.addColumn("col1", Type.BOOLEAN);
+ schema.addColumn("col2", Type.BIT);
+ schema.addColumn("col3", Type.CHAR, 7);
+ schema.addColumn("col4", Type.INT2);
+ schema.addColumn("col5", Type.INT4);
+ schema.addColumn("col6", Type.INT8);
+ schema.addColumn("col7", Type.FLOAT4);
+ schema.addColumn("col8", Type.FLOAT8);
+ schema.addColumn("col9", Type.TEXT);
+ schema.addColumn("col10", Type.BLOB);
+ schema.addColumn("col11", Type.INET4);
+ schema.addColumn("col12", Type.NULL_TYPE);
+ schema.addColumn("col13", CatalogUtil.newDataType(Type.PROTOBUF, TajoIdProtos.QueryIdProto.class.getName()));
+
+ Options options = new Options();
+ TableMeta meta = CatalogUtil.newTableMeta(storeType, options);
+ meta.putOption(CatalogConstants.SEQUENCEFILE_SERDE, TextSerializerDeserializer.class.getName());
+
+ Path tablePath = new Path(testDir, "testVariousTypes.data");
+ Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(meta, schema, tablePath);
+ appender.enableStats();
+ appender.init();
+
+ QueryId queryid = new QueryId("12345", 5);
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(TajoIdProtos.QueryIdProto.class.getName());
+
+ Tuple tuple = new VTuple(13);
+ tuple.put(new Datum[] {
+ DatumFactory.createBool(true),
+ DatumFactory.createBit((byte) 0x99),
+ DatumFactory.createChar("jinho"),
+ DatumFactory.createInt2((short) 17),
+ DatumFactory.createInt4(59),
+ DatumFactory.createInt8(23l),
+ DatumFactory.createFloat4(77.9f),
+ DatumFactory.createFloat8(271.9f),
+ DatumFactory.createText("jinho"),
+ DatumFactory.createBlob("hyunsik babo".getBytes()),
+ DatumFactory.createInet4("192.168.0.1"),
+ NullDatum.get(),
+ factory.createDatum(queryid.getProto())
+ });
+ appender.addTuple(tuple);
+ appender.flush();
+ appender.close();
+
+ FileStatus status = fs.getFileStatus(tablePath);
+ assertEquals(appender.getStats().getNumBytes().longValue(), status.getLen());
+
+ FileFragment fragment = new FileFragment("table", tablePath, 0, status.getLen());
+ Scanner scanner = StorageManagerFactory.getStorageManager(conf).getScanner(meta, schema, fragment);
+ scanner.init();
+
+ Tuple retrieved;
+ while ((retrieved=scanner.next()) != null) {
+ for (int i = 0; i < tuple.size(); i++) {
+ assertEquals(tuple.get(i), retrieved.get(i));
+ }
+ }
+ scanner.close();
+ assertEquals(appender.getStats().getNumBytes().longValue(), scanner.getInputStats().getNumBytes().longValue());
+ assertEquals(appender.getStats().getNumRows().longValue(), scanner.getInputStats().getNumRows().longValue());
+ }
+
+ @Test
+ public void testSequenceFileBinarySerializeDeserialize() throws IOException {
+ if(storeType != StoreType.SEQUENCEFILE) return;
+
+ Schema schema = new Schema();
+ schema.addColumn("col1", Type.BOOLEAN);
+ schema.addColumn("col2", Type.BIT);
+ schema.addColumn("col3", Type.CHAR, 7);
+ schema.addColumn("col4", Type.INT2);
+ schema.addColumn("col5", Type.INT4);
+ schema.addColumn("col6", Type.INT8);
+ schema.addColumn("col7", Type.FLOAT4);
+ schema.addColumn("col8", Type.FLOAT8);
+ schema.addColumn("col9", Type.TEXT);
+ schema.addColumn("col10", Type.BLOB);
+ schema.addColumn("col11", Type.INET4);
+ schema.addColumn("col12", Type.NULL_TYPE);
+ schema.addColumn("col13", CatalogUtil.newDataType(Type.PROTOBUF, TajoIdProtos.QueryIdProto.class.getName()));
+
+ Options options = new Options();
+ TableMeta meta = CatalogUtil.newTableMeta(storeType, options);
+ meta.putOption(CatalogConstants.SEQUENCEFILE_SERDE, BinarySerializerDeserializer.class.getName());
+
+ Path tablePath = new Path(testDir, "testVariousTypes.data");
+ Appender appender = StorageManagerFactory.getStorageManager(conf).getAppender(meta, schema, tablePath);
+ appender.enableStats();
+ appender.init();
+
+ QueryId queryid = new QueryId("12345", 5);
+ ProtobufDatumFactory factory = ProtobufDatumFactory.get(TajoIdProtos.QueryIdProto.class.getName());
+
+ Tuple tuple = new VTuple(13);
+ tuple.put(new Datum[] {
+ DatumFactory.createBool(true),
+ DatumFactory.createBit((byte) 0x99),
+ DatumFactory.createChar("jinho"),
+ DatumFactory.createInt2((short) 17),
+ DatumFactory.createInt4(59),
+ DatumFactory.createInt8(23l),
+ DatumFactory.createFloat4(77.9f),
+ DatumFactory.createFloat8(271.9f),
+ DatumFactory.createText("jinho"),
+ DatumFactory.createBlob("hyunsik babo".getBytes()),
+ DatumFactory.createInet4("192.168.0.1"),
+ NullDatum.get(),
+ factory.createDatum(queryid.getProto())
+ });
+ appender.addTuple(tuple);
+ appender.flush();
+ appender.close();
+
+ FileStatus status = fs.getFileStatus(tablePath);
+ assertEquals(appender.getStats().getNumBytes().longValue(), status.getLen());
+
+ FileFragment fragment = new FileFragment("table", tablePath, 0, status.getLen());
+ Scanner scanner = StorageManagerFactory.getStorageManager(conf).getScanner(meta, schema, fragment);
+ scanner.init();
+
+ Tuple retrieved;
+ while ((retrieved=scanner.next()) != null) {
+ for (int i = 0; i < tuple.size(); i++) {
+ assertEquals(tuple.get(i), retrieved.get(i));
+ }
+ }
+ scanner.close();
+ assertEquals(appender.getStats().getNumBytes().longValue(), scanner.getInputStats().getNumBytes().longValue());
+ assertEquals(appender.getStats().getNumRows().longValue(), scanner.getInputStats().getNumRows().longValue());
+ }
+
+ @Test
public void testTime() throws IOException {
if (storeType == StoreType.CSV || storeType == StoreType.RAW) {
Schema schema = new Schema();
@@ -510,4 +651,7 @@ public class TestStorages {
scanner.close();
}
}
+
+
+
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/303066b2/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 75274d0..cbbdafa 100644
--- a/tajo-storage/src/test/resources/storage-default.xml
+++ b/tajo-storage/src/test/resources/storage-default.xml
@@ -45,7 +45,7 @@
<!--- Registered Scanner Handler -->
<property>
<name>tajo.storage.scanner-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet</value>
+ <value>csv,raw,rcfile,row,trevni,parquet,sequencefile</value>
</property>
<!--- Fragment Class Configurations -->
@@ -73,6 +73,10 @@
<name>tajo.storage.fragment.parquet.class</name>
<value>org.apache.tajo.storage.fragment.FileFragment</value>
</property>
+ <property>
+ <name>tajo.storage.fragment.sequencefile.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
<!--- Scanner Handler -->
<property>
@@ -135,10 +139,20 @@
<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>
+
<!--- Appender Handler -->
<property>
<name>tajo.storage.appender-handler</name>
- <value>csv,raw,rcfile,row,trevni,parquet</value>
+ <value>csv,raw,rcfile,row,trevni,parquet,sequencefile</value>
</property>
<property>
@@ -170,4 +184,10 @@
<name>tajo.storage.appender-handler.parquet.class</name>
<value>org.apache.tajo.storage.parquet.ParquetAppender</value>
</property>
+
+ <property>
+ <name>tajo.storage.appender-handler.sequencefile.class</name>
+ <value>org.apache.tajo.storage.sequencefile.SequenceFileAppender</value>
+ </property>
+
</configuration>