You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hj...@apache.org on 2014/12/05 09:21:27 UTC
[23/29] tajo git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/tajo into hbase_storage
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/Target.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/Target.java
index f49a93d,f49a93d..0524240
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/Target.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/Target.java
@@@ -21,7 -21,7 +21,7 @@@ package org.apache.tajo.plan
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.common.TajoDataTypes.DataType;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.expr.FieldEval;
import org.apache.tajo.plan.serder.PlanGsonHelper;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CaseWhenEval.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/expr/CaseWhenEval.java
index 4321d02,4321d02..24101e5
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CaseWhenEval.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/CaseWhenEval.java
@@@ -27,7 -27,7 +27,7 @@@ import org.apache.tajo.common.TajoDataT
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.NullDatum;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
import org.apache.tajo.plan.serder.PlanGsonHelper;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.util.TUtil;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
index 638383a,638383a..e028481
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/expr/EvalNode.java
@@@ -22,7 -22,7 +22,7 @@@ import com.google.gson.annotations.Expo
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.datum.Datum;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
import org.apache.tajo.plan.serder.PlanGsonHelper;
import org.apache.tajo.storage.Tuple;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/function/GeneralFunction.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/function/GeneralFunction.java
index 006449f,006449f..5a83f48
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/function/GeneralFunction.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/function/GeneralFunction.java
@@@ -23,7 -23,7 +23,7 @@@ import org.apache.tajo.catalog.json.Cat
import org.apache.tajo.catalog.proto.CatalogProtos;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.function.Function;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
import org.apache.tajo.plan.expr.FunctionEval;
import org.apache.tajo.storage.Tuple;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/logical/BinaryNode.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/logical/BinaryNode.java
index 709ef34,709ef34..55b6871
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/BinaryNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/BinaryNode.java
@@@ -19,7 -19,7 +19,7 @@@
package org.apache.tajo.plan.logical;
import com.google.gson.annotations.Expose;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
public abstract class BinaryNode extends LogicalNode implements Cloneable, GsonObject {
@Expose LogicalNode leftChild = null;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/logical/LogicalNode.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/logical/LogicalNode.java
index c42a05e,c42a05e..beb2a8f
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/logical/LogicalNode.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/logical/LogicalNode.java
@@@ -23,7 -23,7 +23,7 @@@ package org.apache.tajo.plan.logical
import com.google.gson.annotations.Expose;
import org.apache.tajo.catalog.Schema;
--import org.apache.tajo.json.GsonObject;
++import org.apache.tajo.storage.json.GsonObject;
import org.apache.tajo.plan.PlanString;
import org.apache.tajo.plan.util.PlannerUtil;
import org.apache.tajo.plan.serder.PlanGsonHelper;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeAdapter.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeAdapter.java
index 5a75e58,5a75e58..8003af5
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeAdapter.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/EvalNodeAdapter.java
@@@ -22,8 -22,8 +22,8 @@@
package org.apache.tajo.plan.serder;
import com.google.gson.*;
--import org.apache.tajo.json.CommonGsonHelper;
--import org.apache.tajo.json.GsonSerDerAdapter;
++import org.apache.tajo.storage.json.CommonGsonHelper;
++import org.apache.tajo.storage.json.GsonSerDerAdapter;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.expr.EvalType;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeAdapter.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeAdapter.java
index f86ba5f,f86ba5f..20c16bb
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeAdapter.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/LogicalNodeAdapter.java
@@@ -22,10 -22,10 +22,10 @@@
package org.apache.tajo.plan.serder;
import com.google.gson.*;
--import org.apache.tajo.json.CommonGsonHelper;
++import org.apache.tajo.storage.json.CommonGsonHelper;
import org.apache.tajo.plan.logical.LogicalNode;
import org.apache.tajo.plan.logical.NodeType;
--import org.apache.tajo.json.GsonSerDerAdapter;
++import org.apache.tajo.storage.json.GsonSerDerAdapter;
import java.lang.reflect.Type;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-plan/src/main/java/org/apache/tajo/plan/serder/PlanGsonHelper.java
----------------------------------------------------------------------
diff --cc tajo-plan/src/main/java/org/apache/tajo/plan/serder/PlanGsonHelper.java
index 8cafbd0,8cafbd0..e0a7ad5
--- a/tajo-plan/src/main/java/org/apache/tajo/plan/serder/PlanGsonHelper.java
+++ b/tajo-plan/src/main/java/org/apache/tajo/plan/serder/PlanGsonHelper.java
@@@ -27,7 -27,7 +27,7 @@@ import org.apache.tajo.catalog.json.Tab
import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.function.Function;
--import org.apache.tajo.json.*;
++import org.apache.tajo.storage.json.*;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.function.AggFunction;
import org.apache.tajo.plan.function.GeneralFunction;
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/StorageManager.java
----------------------------------------------------------------------
diff --cc tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/StorageManager.java
index d2a692d,0000000..58d26d8
mode 100644,000000..100644
--- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/StorageManager.java
+++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/StorageManager.java
@@@ -1,933 -1,0 +1,980 @@@
+/**
+ * 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 com.google.common.collect.Maps;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
++import org.apache.hadoop.fs.PathFilter;
+import org.apache.tajo.*;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.SortSpec;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.catalog.proto.CatalogProtos.FragmentProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.conf.TajoConf.ConfVars;
+import org.apache.tajo.plan.LogicalPlan;
+import org.apache.tajo.plan.logical.LogicalNode;
+import org.apache.tajo.plan.logical.NodeType;
+import org.apache.tajo.plan.logical.ScanNode;
+import org.apache.tajo.plan.rewrite.RewriteRule;
+import org.apache.tajo.storage.fragment.Fragment;
+import org.apache.tajo.storage.fragment.FragmentConvertor;
+import org.apache.tajo.util.TUtil;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.net.URI;
+import java.text.NumberFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * StorageManager manages the functions of storing and reading data.
+ * StorageManager is a abstract class.
+ * For supporting such as HDFS, HBASE, a specific StorageManager should be implemented by inheriting this class.
+ *
+ */
+public abstract class StorageManager {
+ private final Log LOG = LogFactory.getLog(StorageManager.class);
+
+ private static final Class<?>[] DEFAULT_SCANNER_PARAMS = {
+ Configuration.class,
+ Schema.class,
+ TableMeta.class,
+ Fragment.class
+ };
+
+ private static final Class<?>[] DEFAULT_APPENDER_PARAMS = {
+ Configuration.class,
+ QueryUnitAttemptId.class,
+ Schema.class,
+ TableMeta.class,
+ Path.class
+ };
+
++ public static final PathFilter hiddenFileFilter = new PathFilter() {
++ public boolean accept(Path p) {
++ String name = p.getName();
++ return !name.startsWith("_") && !name.startsWith(".");
++ }
++ };
++
+ protected TajoConf conf;
+ protected StoreType storeType;
+
+ /**
+ * Cache of StorageManager.
+ * Key is manager key(warehouse path) + store type
+ */
+ private static final Map<String, StorageManager> storageManagers = Maps.newHashMap();
+
+ /**
+ * Cache of scanner handlers for each storage type.
+ */
+ protected static final Map<String, Class<? extends Scanner>> SCANNER_HANDLER_CACHE
+ = new ConcurrentHashMap<String, Class<? extends Scanner>>();
+
+ /**
+ * Cache of appender handlers for each storage type.
+ */
+ protected static final Map<String, Class<? extends Appender>> APPENDER_HANDLER_CACHE
+ = new ConcurrentHashMap<String, Class<? extends Appender>>();
+
+ /**
+ * Cache of constructors for each class. Pins the classes so they
+ * can't be garbage collected until ReflectionUtils can be collected.
+ */
+ private static final Map<Class<?>, Constructor<?>> CONSTRUCTOR_CACHE =
+ new ConcurrentHashMap<Class<?>, Constructor<?>>();
+
+ public StorageManager(StoreType storeType) {
+ this.storeType = storeType;
+ }
+
+ /**
+ * Initialize storage manager.
+ * @throws java.io.IOException
+ */
+ protected abstract void storageInit() throws IOException;
+
+ /**
+ * This method is called after executing "CREATE TABLE" statement.
+ * If a storage is a file based storage, a storage manager may create directory.
+ *
+ * @param tableDesc Table description which is created.
+ * @param ifNotExists Creates the table only when the table does not exist.
+ * @throws java.io.IOException
+ */
+ public abstract void createTable(TableDesc tableDesc, boolean ifNotExists) throws IOException;
+
+ /**
+ * This method is called after executing "DROP TABLE" statement with the 'PURGE' option
+ * which is the option to delete all the data.
+ *
+ * @param tableDesc
+ * @throws java.io.IOException
+ */
+ public abstract void purgeTable(TableDesc tableDesc) throws IOException;
+
+ /**
+ * Returns the splits that will serve as input for the scan tasks. The
+ * number of splits matches the number of regions in a table.
+ * @param fragmentId The table name or previous ExecutionBlockId
+ * @param tableDesc The table description for the target data.
+ * @param scanNode The logical node for scanning.
+ * @return The list of input fragments.
+ * @throws java.io.IOException
+ */
+ public abstract List<Fragment> getSplits(String fragmentId, TableDesc tableDesc,
+ ScanNode scanNode) throws IOException;
+
+ /**
+ * It returns the splits that will serve as input for the non-forward query scanner such as 'select * from table1'.
+ * The result list should be small. If there is many fragments for scanning, TajoMaster uses the paging navigation.
+ * @param tableDesc The table description for the target data.
+ * @param currentPage The current page number within the entire list.
+ * @param numFragments The number of fragments in the result.
+ * @return The list of input fragments.
+ * @throws java.io.IOException
+ */
+ public abstract List<Fragment> getNonForwardSplit(TableDesc tableDesc, int currentPage, int numFragments)
+ throws IOException;
+
+ /**
+ * It returns the storage property.
+ * @return The storage property
+ */
+ public abstract StorageProperty getStorageProperty();
+
+ /**
+ * Release storage manager resource
+ */
+ public abstract void closeStorageManager();
+
+ /**
+ * It is called by a Repartitioner for range shuffling when the SortRangeType of SortNode is USING_STORAGE_MANAGER.
+ * In general Repartitioner determines the partition range using previous output statistics data.
+ * In the special cases, such as HBase Repartitioner uses the result of this method.
+ *
+ * @param queryContext The current query context which contains query properties.
+ * @param tableDesc The table description for the target data.
+ * @param inputSchema The input schema
+ * @param sortSpecs The sort specification that contains the sort column and sort order.
+ * @return The list of sort ranges.
+ * @throws java.io.IOException
+ */
+ public abstract TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
+ Schema inputSchema, SortSpec[] sortSpecs,
+ TupleRange dataRange) throws IOException;
+
+ /**
+ * This method is called before executing 'INSERT' or 'CREATE TABLE as SELECT'.
+ * In general Tajo creates the target table after finishing the final sub-query of CATS.
+ * But In the special cases, such as HBase INSERT or CAST query uses the target table information.
+ * That kind of the storage should implements the logic related to creating table in this method.
+ *
+ * @param node The child node of the root node.
+ * @throws java.io.IOException
+ */
+ public abstract void beforeInsertOrCATS(LogicalNode node) throws IOException;
+
+ /**
+ * It is called when the query failed.
+ * Each storage manager should implement to be processed when the query fails in this method.
+ *
+ * @param node The child node of the root node.
+ * @throws java.io.IOException
+ */
+ public abstract void rollbackOutputCommit(LogicalNode node) throws IOException;
+
+ /**
+ * Returns the current storage type.
+ * @return
+ */
+ public StoreType getStoreType() {
+ return storeType;
+ }
+
+ /**
+ * Initialize StorageManager instance. It should be called before using.
+ *
+ * @param tajoConf
+ * @throws java.io.IOException
+ */
+ public void init(TajoConf tajoConf) throws IOException {
+ this.conf = tajoConf;
+ storageInit();
+ }
+
+ /**
+ * Close StorageManager
+ * @throws java.io.IOException
+ */
+ public void close() throws IOException {
+ synchronized(storageManagers) {
+ for (StorageManager eachStorageManager: storageManagers.values()) {
+ eachStorageManager.closeStorageManager();
+ }
+ }
+ }
+
+ /**
+ * Returns the splits that will serve as input for the scan tasks. The
+ * number of splits matches the number of regions in a table.
+ *
+ * @param fragmentId The table name or previous ExecutionBlockId
+ * @param tableDesc The table description for the target data.
+ * @return The list of input fragments.
+ * @throws java.io.IOException
+ */
+ public List<Fragment> getSplits(String fragmentId, TableDesc tableDesc) throws IOException {
+ return getSplits(fragmentId, tableDesc, null);
+ }
+
+ /**
+ * Returns FileStorageManager instance.
+ *
+ * @param tajoConf Tajo system property.
+ * @return
+ * @throws java.io.IOException
+ */
+ public static StorageManager getFileStorageManager(TajoConf tajoConf) throws IOException {
+ return getFileStorageManager(tajoConf, null);
+ }
+
+ /**
+ * Returns FileStorageManager instance and sets WAREHOUSE_DIR property in tajoConf with warehousePath parameter.
+ *
+ * @param tajoConf Tajo system property.
+ * @param warehousePath The warehouse directory to be set in the tajoConf.
+ * @return
+ * @throws java.io.IOException
+ */
+ public static StorageManager getFileStorageManager(TajoConf tajoConf, Path warehousePath) throws IOException {
+ URI uri;
+ TajoConf copiedConf = new TajoConf(tajoConf);
+ if (warehousePath != null) {
+ copiedConf.setVar(ConfVars.WAREHOUSE_DIR, warehousePath.toUri().toString());
+ }
+ uri = TajoConf.getWarehouseDir(copiedConf).toUri();
+ String key = "file".equals(uri.getScheme()) ? "file" : uri.toString();
+ return getStorageManager(copiedConf, StoreType.CSV, key);
+ }
+
+ /**
+ * Returns the proper StorageManager instance according to the storeType.
+ *
+ * @param tajoConf Tajo system property.
+ * @param storeType Storage type
+ * @return
+ * @throws java.io.IOException
+ */
+ public static StorageManager getStorageManager(TajoConf tajoConf, String storeType) throws IOException {
+ if ("HBASE".equals(storeType)) {
+ return getStorageManager(tajoConf, StoreType.HBASE);
+ } else {
+ return getStorageManager(tajoConf, StoreType.CSV);
+ }
+ }
+
+ /**
+ * Returns the proper StorageManager instance according to the storeType.
+ *
+ * @param tajoConf Tajo system property.
+ * @param storeType Storage type
+ * @return
+ * @throws java.io.IOException
+ */
+ public static StorageManager getStorageManager(TajoConf tajoConf, StoreType storeType) throws IOException {
+ return getStorageManager(tajoConf, storeType, null);
+ }
+
+ /**
+ * Returns the proper StorageManager instance according to the storeType
+ *
+ * @param tajoConf Tajo system property.
+ * @param storeType Storage type
+ * @param managerKey Key that can identify each storage manager(may be a path)
+ * @return
+ * @throws java.io.IOException
+ */
+ public static synchronized StorageManager getStorageManager (
+ TajoConf tajoConf, StoreType storeType, String managerKey) throws IOException {
+ synchronized (storageManagers) {
+ String storeKey = storeType + managerKey;
+ StorageManager manager = storageManagers.get(storeKey);
+ if (manager == null) {
+ String typeName = "hdfs";
+
+ switch (storeType) {
+ case HBASE:
+ typeName = "hbase";
+ break;
+ default:
+ typeName = "hdfs";
+ }
+
+ Class<? extends StorageManager> storageManagerClass =
+ tajoConf.getClass(String.format("tajo.storage.manager.%s.class", typeName), null, StorageManager.class);
+
+ if (storageManagerClass == null) {
+ throw new IOException("Unknown Storage Type: " + typeName);
+ }
+
+ try {
+ Constructor<? extends StorageManager> constructor =
+ (Constructor<? extends StorageManager>) CONSTRUCTOR_CACHE.get(storageManagerClass);
+ if (constructor == null) {
+ constructor = storageManagerClass.getDeclaredConstructor(new Class<?>[]{StoreType.class});
+ constructor.setAccessible(true);
+ CONSTRUCTOR_CACHE.put(storageManagerClass, constructor);
+ }
+ manager = constructor.newInstance(new Object[]{storeType});
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ manager.init(tajoConf);
+ storageManagers.put(storeKey, manager);
+ }
+
+ return manager;
+ }
+ }
+
+ /**
+ * Returns Scanner instance.
+ *
+ * @param meta The table meta
+ * @param schema The input schema
+ * @param fragment The fragment for scanning
+ * @param target Columns which are selected.
+ * @return Scanner instance
+ * @throws java.io.IOException
+ */
+ public Scanner getScanner(TableMeta meta, Schema schema, FragmentProto fragment, Schema target) throws IOException {
+ return getScanner(meta, schema, FragmentConvertor.convert(conf, fragment), target);
+ }
+
+ /**
+ * Returns Scanner instance.
+ *
+ * @param meta The table meta
+ * @param schema The input schema
+ * @param fragment The fragment for scanning
+ * @return Scanner instance
+ * @throws java.io.IOException
+ */
+ public Scanner getScanner(TableMeta meta, Schema schema, Fragment fragment) throws IOException {
+ return getScanner(meta, schema, fragment, schema);
+ }
+
+ /**
+ * Returns Scanner instance.
+ *
+ * @param meta The table meta
+ * @param schema The input schema
+ * @param fragment The fragment for scanning
+ * @param target The output schema
+ * @return Scanner instance
+ * @throws java.io.IOException
+ */
+ public Scanner getScanner(TableMeta meta, Schema schema, Fragment fragment, Schema target) throws IOException {
+ if (fragment.isEmpty()) {
+ Scanner scanner = new NullScanner(conf, schema, meta, fragment);
+ scanner.setTarget(target.toArray());
+
+ return scanner;
+ }
+
+ Scanner scanner;
+
+ Class<? extends Scanner> scannerClass = getScannerClass(meta.getStoreType());
+ scanner = newScannerInstance(scannerClass, conf, schema, meta, fragment);
+ if (scanner.isProjectable()) {
+ scanner.setTarget(target.toArray());
+ }
+
+ return scanner;
+ }
+
+ /**
+ * Returns Scanner instance.
+ *
+ * @param conf The system property
+ * @param meta The table meta
+ * @param schema The input schema
+ * @param fragment The fragment for scanning
+ * @param target The output schema
+ * @return Scanner instance
+ * @throws java.io.IOException
+ */
+ public static synchronized SeekableScanner getSeekableScanner(
+ TajoConf conf, TableMeta meta, Schema schema, Fragment fragment, Schema target) throws IOException {
+ return (SeekableScanner)getStorageManager(conf, meta.getStoreType()).getScanner(meta, schema, fragment, target);
+ }
+
+ /**
+ * Returns Appender instance.
+ * @param queryContext Query property.
+ * @param taskAttemptId Task id.
+ * @param meta Table meta data.
+ * @param schema Output schema.
+ * @param workDir Working directory
+ * @return Appender instance
+ * @throws java.io.IOException
+ */
+ public Appender getAppender(OverridableConf queryContext,
+ QueryUnitAttemptId taskAttemptId, TableMeta meta, Schema schema, Path workDir)
+ throws IOException {
+ Appender appender;
+
+ Class<? extends Appender> appenderClass;
+
+ String handlerName = meta.getStoreType().name().toLowerCase();
+ appenderClass = APPENDER_HANDLER_CACHE.get(handlerName);
+ if (appenderClass == null) {
+ appenderClass = conf.getClass(
+ String.format("tajo.storage.appender-handler.%s.class",
+ meta.getStoreType().name().toLowerCase()), null, Appender.class);
+ APPENDER_HANDLER_CACHE.put(handlerName, appenderClass);
+ }
+
+ if (appenderClass == null) {
+ throw new IOException("Unknown Storage Type: " + meta.getStoreType());
+ }
+
+ appender = newAppenderInstance(appenderClass, conf, taskAttemptId, meta, schema, workDir);
+
+ return appender;
+ }
+
+ /**
+ * Creates a scanner instance.
+ *
+ * @param theClass Concrete class of scanner
+ * @param conf System property
+ * @param schema Input schema
+ * @param meta Table meta data
+ * @param fragment The fragment for scanning
+ * @param <T>
+ * @return The scanner instance
+ */
+ public static <T> T newScannerInstance(Class<T> theClass, Configuration conf, Schema schema, TableMeta meta,
+ Fragment fragment) {
+ T result;
+ try {
+ Constructor<T> meth = (Constructor<T>) CONSTRUCTOR_CACHE.get(theClass);
+ if (meth == null) {
+ meth = theClass.getDeclaredConstructor(DEFAULT_SCANNER_PARAMS);
+ meth.setAccessible(true);
+ CONSTRUCTOR_CACHE.put(theClass, meth);
+ }
+ result = meth.newInstance(new Object[]{conf, schema, meta, fragment});
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return result;
+ }
+
+ /**
+ * Creates a scanner instance.
+ *
+ * @param theClass Concrete class of scanner
+ * @param conf System property
+ * @param taskAttemptId Task id
+ * @param meta Table meta data
+ * @param schema Input schema
+ * @param workDir Working directory
+ * @param <T>
+ * @return The scanner instance
+ */
+ public static <T> T newAppenderInstance(Class<T> theClass, Configuration conf, QueryUnitAttemptId taskAttemptId,
+ TableMeta meta, Schema schema, Path workDir) {
+ T result;
+ try {
+ Constructor<T> meth = (Constructor<T>) CONSTRUCTOR_CACHE.get(theClass);
+ if (meth == null) {
+ meth = theClass.getDeclaredConstructor(DEFAULT_APPENDER_PARAMS);
+ meth.setAccessible(true);
+ CONSTRUCTOR_CACHE.put(theClass, meth);
+ }
+ result = meth.newInstance(new Object[]{conf, taskAttemptId, schema, meta, workDir});
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return result;
+ }
+
+ /**
+ * Return the Scanner class for the StoreType that is defined in storage-default.xml.
+ *
+ * @param storeType store type
+ * @return The Scanner class
+ * @throws java.io.IOException
+ */
+ public Class<? extends Scanner> getScannerClass(CatalogProtos.StoreType storeType) throws IOException {
+ String handlerName = storeType.name().toLowerCase();
+ Class<? extends Scanner> scannerClass = SCANNER_HANDLER_CACHE.get(handlerName);
+ if (scannerClass == null) {
+ scannerClass = conf.getClass(
+ String.format("tajo.storage.scanner-handler.%s.class",storeType.name().toLowerCase()), null, Scanner.class);
+ SCANNER_HANDLER_CACHE.put(handlerName, scannerClass);
+ }
+
+ if (scannerClass == null) {
+ throw new IOException("Unknown Storage Type: " + storeType.name());
+ }
+
+ return scannerClass;
+ }
+
+ /**
+ * Return length of the fragment.
+ * In the UNKNOWN_LENGTH case get FRAGMENT_ALTERNATIVE_UNKNOWN_LENGTH from the configuration.
+ *
+ * @param conf Tajo system property
+ * @param fragment Fragment
+ * @return
+ */
+ public static long getFragmentLength(TajoConf conf, Fragment fragment) {
+ if (fragment.getLength() == TajoConstants.UNKNOWN_LENGTH) {
+ return conf.getLongVar(ConfVars.FRAGMENT_ALTERNATIVE_UNKNOWN_LENGTH);
+ } else {
+ return fragment.getLength();
+ }
+ }
+
+ /**
+ * It is called after making logical plan. Storage manager should verify the schema for inserting.
+ *
+ * @param tableDesc The table description of insert target.
+ * @param outSchema The output schema of select query for inserting.
+ * @throws java.io.IOException
+ */
+ public void verifyInsertTableSchema(TableDesc tableDesc, Schema outSchema) throws IOException {
+ // nothing to do
+ }
+
+ /**
+ * Returns the list of storage specified rewrite rules.
+ * This values are used by LogicalOptimizer.
+ *
+ * @param queryContext The query property
+ * @param tableDesc The description of the target table.
+ * @return The list of storage specified rewrite rules
+ * @throws java.io.IOException
+ */
+ public List<RewriteRule> getRewriteRules(OverridableConf queryContext, TableDesc tableDesc) throws IOException {
+ return null;
+ }
+
+ /**
+ * Finalizes result data. Tajo stores result data in the staging directory.
+ * If the query fails, clean up the staging directory.
+ * Otherwise the query is successful, move to the final directory from the staging directory.
+ *
+ * @param queryContext The query property
+ * @param finalEbId The final execution block id
+ * @param plan The query plan
+ * @param schema The final output schema
+ * @param tableDesc The description of the target table
+ * @return Saved path
+ * @throws java.io.IOException
+ */
+ public Path commitOutputData(OverridableConf queryContext, ExecutionBlockId finalEbId,
+ LogicalPlan plan, Schema schema,
+ TableDesc tableDesc) throws IOException {
+ return commitOutputData(queryContext, finalEbId, plan, schema, tableDesc, true);
+ }
+
+ /**
+ * Finalizes result data. Tajo stores result data in the staging directory.
+ * If the query fails, clean up the staging directory.
+ * Otherwise the query is successful, move to the final directory from the staging directory.
+ *
+ * @param queryContext The query property
+ * @param finalEbId The final execution block id
+ * @param plan The query plan
+ * @param schema The final output schema
+ * @param tableDesc The description of the target table
+ * @param changeFileSeq If true change result file name with max sequence.
+ * @return Saved path
+ * @throws java.io.IOException
+ */
+ protected Path commitOutputData(OverridableConf queryContext, ExecutionBlockId finalEbId,
+ LogicalPlan plan, Schema schema,
+ TableDesc tableDesc, boolean changeFileSeq) throws IOException {
+ Path stagingDir = new Path(queryContext.get(QueryVars.STAGING_DIR));
+ Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);
+ Path finalOutputDir;
+ if (!queryContext.get(QueryVars.OUTPUT_TABLE_PATH, "").isEmpty()) {
+ finalOutputDir = new Path(queryContext.get(QueryVars.OUTPUT_TABLE_PATH));
- FileSystem fs = stagingResultDir.getFileSystem(conf);
-
- if (queryContext.getBool(QueryVars.OUTPUT_OVERWRITE, false)) { // INSERT OVERWRITE INTO
-
- // It moves the original table into the temporary location.
- // Then it moves the new result table into the original table location.
- // Upon failed, it recovers the original table if possible.
- boolean movedToOldTable = false;
- boolean committed = false;
- Path oldTableDir = new Path(stagingDir, TajoConstants.INSERT_OVERWIRTE_OLD_TABLE_NAME);
-
- if (!queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty()) {
- // This is a map for existing non-leaf directory to rename. A key is current directory and a value is
- // renaming directory.
- Map<Path, Path> renameDirs = TUtil.newHashMap();
- // This is a map for recovering existing partition directory. A key is current directory and a value is
- // temporary directory to back up.
- Map<Path, Path> recoveryDirs = TUtil.newHashMap();
-
- try {
- if (!fs.exists(finalOutputDir)) {
- fs.mkdirs(finalOutputDir);
- }
++ try {
++ FileSystem fs = stagingResultDir.getFileSystem(conf);
++
++ if (queryContext.getBool(QueryVars.OUTPUT_OVERWRITE, false)) { // INSERT OVERWRITE INTO
+
- visitPartitionedDirectory(fs, stagingResultDir, finalOutputDir, stagingResultDir.toString(),
- renameDirs, oldTableDir);
-
- // Rename target partition directories
- for(Map.Entry<Path, Path> entry : renameDirs.entrySet()) {
- // Backup existing data files for recovering
- if (fs.exists(entry.getValue())) {
- String recoveryPathString = entry.getValue().toString().replaceAll(finalOutputDir.toString(),
- oldTableDir.toString());
- Path recoveryPath = new Path(recoveryPathString);
- fs.rename(entry.getValue(), recoveryPath);
- fs.exists(recoveryPath);
- recoveryDirs.put(entry.getValue(), recoveryPath);
++ // It moves the original table into the temporary location.
++ // Then it moves the new result table into the original table location.
++ // Upon failed, it recovers the original table if possible.
++ boolean movedToOldTable = false;
++ boolean committed = false;
++ Path oldTableDir = new Path(stagingDir, TajoConstants.INSERT_OVERWIRTE_OLD_TABLE_NAME);
++
++ if (!queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty()) {
++ // This is a map for existing non-leaf directory to rename. A key is current directory and a value is
++ // renaming directory.
++ Map<Path, Path> renameDirs = TUtil.newHashMap();
++ // This is a map for recovering existing partition directory. A key is current directory and a value is
++ // temporary directory to back up.
++ Map<Path, Path> recoveryDirs = TUtil.newHashMap();
++
++ try {
++ if (!fs.exists(finalOutputDir)) {
++ fs.mkdirs(finalOutputDir);
+ }
- // Delete existing directory
- fs.delete(entry.getValue(), true);
- // Rename staging directory to final output directory
- fs.rename(entry.getKey(), entry.getValue());
- }
+
- } catch (IOException ioe) {
- // Remove created dirs
- for(Map.Entry<Path, Path> entry : renameDirs.entrySet()) {
- fs.delete(entry.getValue(), true);
- }
++ visitPartitionedDirectory(fs, stagingResultDir, finalOutputDir, stagingResultDir.toString(),
++ renameDirs, oldTableDir);
++
++ // Rename target partition directories
++ for(Map.Entry<Path, Path> entry : renameDirs.entrySet()) {
++ // Backup existing data files for recovering
++ if (fs.exists(entry.getValue())) {
++ String recoveryPathString = entry.getValue().toString().replaceAll(finalOutputDir.toString(),
++ oldTableDir.toString());
++ Path recoveryPath = new Path(recoveryPathString);
++ fs.rename(entry.getValue(), recoveryPath);
++ fs.exists(recoveryPath);
++ recoveryDirs.put(entry.getValue(), recoveryPath);
++ }
++ // Delete existing directory
++ fs.delete(entry.getValue(), true);
++ // Rename staging directory to final output directory
++ fs.rename(entry.getKey(), entry.getValue());
++ }
+
- // Recovery renamed dirs
- for(Map.Entry<Path, Path> entry : recoveryDirs.entrySet()) {
- fs.delete(entry.getValue(), true);
- fs.rename(entry.getValue(), entry.getKey());
- }
- throw new IOException(ioe.getMessage());
- }
- } else {
- try {
- if (fs.exists(finalOutputDir)) {
- fs.rename(finalOutputDir, oldTableDir);
- movedToOldTable = fs.exists(oldTableDir);
- } else { // if the parent does not exist, make its parent directory.
- fs.mkdirs(finalOutputDir.getParent());
++ } catch (IOException ioe) {
++ // Remove created dirs
++ for(Map.Entry<Path, Path> entry : renameDirs.entrySet()) {
++ fs.delete(entry.getValue(), true);
++ }
++
++ // Recovery renamed dirs
++ for(Map.Entry<Path, Path> entry : recoveryDirs.entrySet()) {
++ fs.delete(entry.getValue(), true);
++ fs.rename(entry.getValue(), entry.getKey());
++ }
++ throw new IOException(ioe.getMessage());
+ }
++ } else { // no partition
++ try {
++
++ // if the final output dir exists, move all contents to the temporary table dir.
++ // Otherwise, just make the final output dir. As a result, the final output dir will be empty.
++ if (fs.exists(finalOutputDir)) {
++ fs.mkdirs(oldTableDir);
+
- fs.rename(stagingResultDir, finalOutputDir);
- committed = fs.exists(finalOutputDir);
- } catch (IOException ioe) {
- // recover the old table
- if (movedToOldTable && !committed) {
- fs.rename(oldTableDir, finalOutputDir);
++ for (FileStatus status : fs.listStatus(finalOutputDir, StorageManager.hiddenFileFilter)) {
++ fs.rename(status.getPath(), oldTableDir);
++ }
++
++ movedToOldTable = fs.exists(oldTableDir);
++ } else { // if the parent does not exist, make its parent directory.
++ fs.mkdirs(finalOutputDir);
++ }
++
++ // Move the results to the final output dir.
++ for (FileStatus status : fs.listStatus(stagingResultDir)) {
++ fs.rename(status.getPath(), finalOutputDir);
++ }
++
++ // Check the final output dir
++ committed = fs.exists(finalOutputDir);
++
++ } catch (IOException ioe) {
++ // recover the old table
++ if (movedToOldTable && !committed) {
++
++ // if commit is failed, recover the old data
++ for (FileStatus status : fs.listStatus(finalOutputDir, StorageManager.hiddenFileFilter)) {
++ fs.delete(status.getPath(), true);
++ }
++
++ for (FileStatus status : fs.listStatus(oldTableDir)) {
++ fs.rename(status.getPath(), finalOutputDir);
++ }
++ }
++
++ throw new IOException(ioe.getMessage());
+ }
+ }
- }
- } else {
- String queryType = queryContext.get(QueryVars.COMMAND_TYPE);
++ } else {
++ String queryType = queryContext.get(QueryVars.COMMAND_TYPE);
+
- if (queryType != null && queryType.equals(NodeType.INSERT.name())) { // INSERT INTO an existing table
++ if (queryType != null && queryType.equals(NodeType.INSERT.name())) { // INSERT INTO an existing table
+
- NumberFormat fmt = NumberFormat.getInstance();
- fmt.setGroupingUsed(false);
- fmt.setMinimumIntegerDigits(3);
++ NumberFormat fmt = NumberFormat.getInstance();
++ fmt.setGroupingUsed(false);
++ fmt.setMinimumIntegerDigits(3);
+
- if (!queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty()) {
- for(FileStatus eachFile: fs.listStatus(stagingResultDir)) {
- if (eachFile.isFile()) {
- LOG.warn("Partition table can't have file in a staging dir: " + eachFile.getPath());
- continue;
++ if (!queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty()) {
++ for(FileStatus eachFile: fs.listStatus(stagingResultDir)) {
++ if (eachFile.isFile()) {
++ LOG.warn("Partition table can't have file in a staging dir: " + eachFile.getPath());
++ continue;
++ }
++ moveResultFromStageToFinal(fs, stagingResultDir, eachFile, finalOutputDir, fmt, -1, changeFileSeq);
++ }
++ } else {
++ int maxSeq = StorageUtil.getMaxFileSequence(fs, finalOutputDir, false) + 1;
++ for(FileStatus eachFile: fs.listStatus(stagingResultDir)) {
++ if (eachFile.getPath().getName().startsWith("_")) {
++ continue;
++ }
++ moveResultFromStageToFinal(fs, stagingResultDir, eachFile, finalOutputDir, fmt, maxSeq++, changeFileSeq);
+ }
- moveResultFromStageToFinal(fs, stagingResultDir, eachFile, finalOutputDir, fmt, -1, changeFileSeq);
+ }
- } else {
- int maxSeq = StorageUtil.getMaxFileSequence(fs, finalOutputDir, false) + 1;
- for(FileStatus eachFile: fs.listStatus(stagingResultDir)) {
- if (eachFile.getPath().getName().startsWith("_")) {
- continue;
++ // checking all file moved and remove empty dir
++ verifyAllFileMoved(fs, stagingResultDir);
++ FileStatus[] files = fs.listStatus(stagingResultDir);
++ if (files != null && files.length != 0) {
++ for (FileStatus eachFile: files) {
++ LOG.error("There are some unmoved files in staging dir:" + eachFile.getPath());
+ }
- moveResultFromStageToFinal(fs, stagingResultDir, eachFile, finalOutputDir, fmt, maxSeq++, changeFileSeq);
+ }
- }
- // checking all file moved and remove empty dir
- verifyAllFileMoved(fs, stagingResultDir);
- FileStatus[] files = fs.listStatus(stagingResultDir);
- if (files != null && files.length != 0) {
- for (FileStatus eachFile: files) {
- LOG.error("There are some unmoved files in staging dir:" + eachFile.getPath());
++ } else { // CREATE TABLE AS SELECT (CTAS)
++ if (fs.exists(finalOutputDir)) {
++ for (FileStatus status : fs.listStatus(stagingResultDir)) {
++ fs.rename(status.getPath(), finalOutputDir);
++ }
++ } else {
++ fs.rename(stagingResultDir, finalOutputDir);
+ }
++ LOG.info("Moved from the staging dir to the output directory '" + finalOutputDir);
+ }
- } else { // CREATE TABLE AS SELECT (CTAS)
- fs.rename(stagingResultDir, finalOutputDir);
- LOG.info("Moved from the staging dir to the output directory '" + finalOutputDir);
+ }
++
++ // remove the staging directory if the final output dir is given.
++ Path stagingDirRoot = stagingDir.getParent();
++ fs.delete(stagingDirRoot, true);
++ } catch (Throwable t) {
++ LOG.error(t);
++ throw new IOException(t);
+ }
+ } else {
+ finalOutputDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);
+ }
+
+ return finalOutputDir;
+ }
+
+ /**
+ * Attach the sequence number to the output file name and than move the file into the final result path.
+ *
+ * @param fs FileSystem
+ * @param stagingResultDir The staging result dir
+ * @param fileStatus The file status
+ * @param finalOutputPath Final output path
+ * @param nf Number format
+ * @param fileSeq The sequence number
+ * @throws java.io.IOException
+ */
+ private void moveResultFromStageToFinal(FileSystem fs, Path stagingResultDir,
+ FileStatus fileStatus, Path finalOutputPath,
+ NumberFormat nf,
+ int fileSeq, boolean changeFileSeq) throws IOException {
+ if (fileStatus.isDirectory()) {
+ String subPath = extractSubPath(stagingResultDir, fileStatus.getPath());
+ if (subPath != null) {
+ Path finalSubPath = new Path(finalOutputPath, subPath);
+ if (!fs.exists(finalSubPath)) {
+ fs.mkdirs(finalSubPath);
+ }
+ int maxSeq = StorageUtil.getMaxFileSequence(fs, finalSubPath, false);
+ for (FileStatus eachFile : fs.listStatus(fileStatus.getPath())) {
+ if (eachFile.getPath().getName().startsWith("_")) {
+ continue;
+ }
+ moveResultFromStageToFinal(fs, stagingResultDir, eachFile, finalOutputPath, nf, ++maxSeq, changeFileSeq);
+ }
+ } else {
+ throw new IOException("Wrong staging dir:" + stagingResultDir + "," + fileStatus.getPath());
+ }
+ } else {
+ String subPath = extractSubPath(stagingResultDir, fileStatus.getPath());
+ if (subPath != null) {
+ Path finalSubPath = new Path(finalOutputPath, subPath);
+ if (changeFileSeq) {
+ finalSubPath = new Path(finalSubPath.getParent(), replaceFileNameSeq(finalSubPath, fileSeq, nf));
+ }
+ if (!fs.exists(finalSubPath.getParent())) {
+ fs.mkdirs(finalSubPath.getParent());
+ }
+ if (fs.exists(finalSubPath)) {
+ throw new IOException("Already exists data file:" + finalSubPath);
+ }
+ boolean success = fs.rename(fileStatus.getPath(), finalSubPath);
+ if (success) {
+ LOG.info("Moving staging file[" + fileStatus.getPath() + "] + " +
+ "to final output[" + finalSubPath + "]");
+ } else {
+ LOG.error("Can't move staging file[" + fileStatus.getPath() + "] + " +
+ "to final output[" + finalSubPath + "]");
+ }
+ }
+ }
+ }
+
+ /**
+ * Removes the path of the parent.
+ * @param parentPath
+ * @param childPath
+ * @return
+ */
+ private String extractSubPath(Path parentPath, Path childPath) {
+ String parentPathStr = parentPath.toUri().getPath();
+ String childPathStr = childPath.toUri().getPath();
+
+ if (parentPathStr.length() > childPathStr.length()) {
+ return null;
+ }
+
+ int index = childPathStr.indexOf(parentPathStr);
+ if (index != 0) {
+ return null;
+ }
+
+ return childPathStr.substring(parentPathStr.length() + 1);
+ }
+
+ /**
+ * Attach the sequence number to a path.
+ *
+ * @param path Path
+ * @param seq sequence number
+ * @param nf Number format
+ * @return New path attached with sequence number
+ * @throws java.io.IOException
+ */
+ private String replaceFileNameSeq(Path path, int seq, NumberFormat nf) throws IOException {
+ String[] tokens = path.getName().split("-");
+ if (tokens.length != 4) {
+ throw new IOException("Wrong result file name:" + path);
+ }
+ return tokens[0] + "-" + tokens[1] + "-" + tokens[2] + "-" + nf.format(seq);
+ }
+
+ /**
+ * Make sure all files are moved.
+ * @param fs FileSystem
+ * @param stagingPath The stagind directory
+ * @return
+ * @throws java.io.IOException
+ */
+ private boolean verifyAllFileMoved(FileSystem fs, Path stagingPath) throws IOException {
+ FileStatus[] files = fs.listStatus(stagingPath);
+ if (files != null && files.length != 0) {
+ for (FileStatus eachFile: files) {
+ if (eachFile.isFile()) {
+ LOG.error("There are some unmoved files in staging dir:" + eachFile.getPath());
+ return false;
+ } else {
+ if (verifyAllFileMoved(fs, eachFile.getPath())) {
+ fs.delete(eachFile.getPath(), false);
+ } else {
+ return false;
+ }
+ }
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * This method sets a rename map which includes renamed staging directory to final output directory recursively.
+ * If there exists some data files, this delete it for duplicate data.
+ *
+ *
+ * @param fs
+ * @param stagingPath
+ * @param outputPath
+ * @param stagingParentPathString
+ * @throws java.io.IOException
+ */
+ private void visitPartitionedDirectory(FileSystem fs, Path stagingPath, Path outputPath,
+ String stagingParentPathString,
+ Map<Path, Path> renameDirs, Path oldTableDir) throws IOException {
+ FileStatus[] files = fs.listStatus(stagingPath);
+
+ for(FileStatus eachFile : files) {
+ if (eachFile.isDirectory()) {
+ Path oldPath = eachFile.getPath();
+
+ // Make recover directory.
+ String recoverPathString = oldPath.toString().replaceAll(stagingParentPathString,
+ oldTableDir.toString());
+ Path recoveryPath = new Path(recoverPathString);
+ if (!fs.exists(recoveryPath)) {
+ fs.mkdirs(recoveryPath);
+ }
+
+ visitPartitionedDirectory(fs, eachFile.getPath(), outputPath, stagingParentPathString,
+ renameDirs, oldTableDir);
+ // Find last order partition for renaming
+ String newPathString = oldPath.toString().replaceAll(stagingParentPathString,
+ outputPath.toString());
+ Path newPath = new Path(newPathString);
+ if (!isLeafDirectory(fs, eachFile.getPath())) {
+ renameDirs.put(eachFile.getPath(), newPath);
+ } else {
+ if (!fs.exists(newPath)) {
+ fs.mkdirs(newPath);
+ }
+ }
+ }
+ }
+ }
+
+ private boolean isLeafDirectory(FileSystem fs, Path path) throws IOException {
+ boolean retValue = false;
+
+ FileStatus[] files = fs.listStatus(path);
+ for (FileStatus file : files) {
+ if (fs.isDirectory(file.getPath())) {
+ retValue = true;
+ break;
+ }
+ }
+
+ return retValue;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-storage/tajo-storage-common/src/main/resources/storage-default.xml
----------------------------------------------------------------------
diff --cc tajo-storage/tajo-storage-common/src/main/resources/storage-default.xml
index 47d11c7,0000000..67033ed
mode 100644,000000..100644
--- a/tajo-storage/tajo-storage-common/src/main/resources/storage-default.xml
+++ b/tajo-storage/tajo-storage-common/src/main/resources/storage-default.xml
@@@ -1,184 -1,0 +1,198 @@@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+
+<!--
+ 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.
+ -->
+
+<configuration>
+ <!-- Storage Manager Configuration -->
+ <property>
+ <name>tajo.storage.manager.hdfs.class</name>
+ <value>org.apache.tajo.storage.FileStorageManager</value>
+ </property>
+ <property>
+ <name>tajo.storage.manager.hbase.class</name>
+ <value>org.apache.tajo.storage.hbase.HBaseStorageManager</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.manager.concurrency.perDisk</name>
+ <value>1</value>
+ <description></description>
+ </property>
+
+ <!--- Registered Scanner Handler -->
+ <property>
+ <name>tajo.storage.scanner-handler</name>
- <value>textfile,csv,raw,rcfile,row,parquet,sequencefile,avro,hbase</value>
++ <value>textfile,csv,json,raw,rcfile,row,parquet,sequencefile,avro,hbase</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>
+ <property>
++ <name>tajo.storage.fragment.json.class</name>
++ <value>org.apache.tajo.storage.fragment.FileFragment</value>
++ </property>
++ <property>
+ <name>tajo.storage.fragment.raw.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
+ <name>tajo.storage.fragment.rcfile.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
+ <name>tajo.storage.fragment.row.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
+ <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>
+ <property>
+ <name>tajo.storage.fragment.avro.class</name>
+ <value>org.apache.tajo.storage.fragment.FileFragment</value>
+ </property>
+ <property>
+ <name>tajo.storage.fragment.hbase.class</name>
+ <value>org.apache.tajo.storage.hbase.HBaseFragment</value>
+ </property>
+
+ <!--- 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>
+
+ <property>
++ <name>tajo.storage.scanner-handler.json.class</name>
++ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileScanner</value>
++ </property>
++
++ <property>
+ <name>tajo.storage.scanner-handler.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.rowfile.class</name>
+ <value>org.apache.tajo.storage.RowFile$RowFileScanner</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.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>
+
+ <property>
+ <name>tajo.storage.scanner-handler.hbase.class</name>
+ <value>org.apache.tajo.storage.hbase.HBaseScanner</value>
+ </property>
+
+ <!--- Appender Handler -->
+ <property>
+ <name>tajo.storage.appender-handler</name>
+ <value>textfile,csv,raw,rcfile,row,parquet,sequencefile,avro,hbase</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.textfile.class</name>
+ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileAppender</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.csv.class</name>
+ <value>org.apache.tajo.storage.CSVFile$CSVAppender</value>
+ </property>
+
+ <property>
++ <name>tajo.storage.appender-handler.json.class</name>
++ <value>org.apache.tajo.storage.text.DelimitedTextFile$DelimitedTextFileAppender</value>
++ </property>
++
++ <property>
+ <name>tajo.storage.appender-handler.raw.class</name>
+ <value>org.apache.tajo.storage.RawFile$RawFileAppender</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.rcfile.class</name>
+ <value>org.apache.tajo.storage.rcfile.RCFile$RCFileAppender</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.rowfile.class</name>
+ <value>org.apache.tajo.storage.RowFile$RowFileAppender</value>
+ </property>
+
+ <property>
+ <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>
+
+ <property>
+ <name>tajo.storage.appender-handler.avro.class</name>
+ <value>org.apache.tajo.storage.avro.AvroAppender</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.hbase.class</name>
+ <value>org.apache.tajo.storage.hbase.HFileAppender</value>
+ </property>
+
+ <property>
+ <name>tajo.storage.appender-handler.hfile.class</name>
+ <value>org.apache.tajo.storage.hbase.HFileAppender</value>
+ </property>
+</configuration>
http://git-wip-us.apache.org/repos/asf/tajo/blob/940546a0/tajo-storage/tajo-storage-hdfs/pom.xml
----------------------------------------------------------------------
diff --cc tajo-storage/tajo-storage-hdfs/pom.xml
index 5105ac5,0000000..ef8e9c2
mode 100644,000000..100644
--- a/tajo-storage/tajo-storage-hdfs/pom.xml
+++ b/tajo-storage/tajo-storage-hdfs/pom.xml
@@@ -1,380 -1,0 +1,385 @@@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Copyright 2012 Database Lab., Korea Univ.
+
+Licensed 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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>tajo-project</artifactId>
+ <groupId>org.apache.tajo</groupId>
+ <version>0.9.1-SNAPSHOT</version>
+ <relativePath>../../tajo-project</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>tajo-storage-hdfs</artifactId>
+ <packaging>jar</packaging>
+ <name>Tajo HDFS Storage</name>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ <parquet.version>1.5.0</parquet.version>
+ <parquet.format.version>2.1.0</parquet.format.version>
+ </properties>
+
+ <repositories>
+ <repository>
+ <id>repository.jboss.org</id>
+ <url>https://repository.jboss.org/nexus/content/repositories/releases/
+ </url>
+ <snapshots>
+ <enabled>false</enabled>
+ </snapshots>
+ </repository>
+ </repositories>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>1.6</source>
+ <target>1.6</target>
+ <encoding>${project.build.sourceEncoding}</encoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>verify</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <excludes>
- <exclude>src/test/resources/testVariousTypes.avsc</exclude>
++ <exclude>src/test/resources/dataset/**</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <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>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>create-protobuf-generated-sources-directory</id>
+ <phase>initialize</phase>
+ <configuration>
+ <target>
+ <mkdir dir="target/generated-sources/proto" />
+ </target>
+ </configuration>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.2</version>
+ <executions>
+ <execution>
+ <id>generate-sources</id>
+ <phase>generate-sources</phase>
+ <configuration>
+ <executable>protoc</executable>
+ <arguments>
+ <argument>-Isrc/main/proto/</argument>
+ <argument>--proto_path=../../tajo-common/src/main/proto</argument>
+ <argument>--proto_path=../../tajo-catalog/tajo-catalog-common/src/main/proto</argument>
+ <argument>--java_out=target/generated-sources/proto</argument>
+ <argument>src/main/proto/StorageFragmentProtos.proto</argument>
+ </arguments>
+ </configuration>
+ <goals>
+ <goal>exec</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.5</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>target/generated-sources/proto</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-report-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.tajo</groupId>
+ <artifactId>tajo-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tajo</groupId>
+ <artifactId>tajo-catalog-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tajo</groupId>
+ <artifactId>tajo-plan</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tajo</groupId>
+ <artifactId>tajo-storage-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>trevni-core</artifactId>
+ <version>1.7.3</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>trevni-avro</artifactId>
+ <version>1.7.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>zookeeper</artifactId>
+ <groupId>org.apache.zookeeper</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>slf4j-api</artifactId>
+ <groupId>org.slf4j</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>jersey-json</artifactId>
+ <groupId>com.sun.jersey</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-el</groupId>
+ <artifactId>commons-el</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-compiler</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.mortbay.jetty</groupId>
+ <artifactId>jsp-2.1-jetty</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jersey.jersey-test-framework</groupId>
+ <artifactId>jersey-test-framework-grizzly2</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minicluster</artifactId>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-el</groupId>
+ <artifactId>commons-el</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-compiler</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.mortbay.jetty</groupId>
+ <artifactId>jsp-2.1-jetty</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jersey.jersey-test-framework</groupId>
+ <artifactId>jersey-test-framework-grizzly2</artifactId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-yarn-server-tests</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-mapreduce-client-app</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-yarn-api</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-mapreduce-client-hs</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <groupId>org.apache.hadoop</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>parquet-column</artifactId>
+ <version>${parquet.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>parquet-hadoop</artifactId>
+ <version>${parquet.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>parquet-format</artifactId>
+ <version>${parquet.format.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ </dependency>
++ <dependency>
++ <groupId>net.minidev</groupId>
++ <artifactId>json-smart</artifactId>
++ <version>2.0</version>
++ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>docs</id>
+ <activation>
+ <activeByDefault>false</activeByDefault>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <executions>
+ <execution>
+ <!-- build javadoc jars per jar for publishing to maven -->
+ <id>module-javadocs</id>
+ <phase>package</phase>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ <configuration>
+ <destDir>${project.build.directory}</destDir>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-report-plugin</artifactId>
+ <version>2.15</version>
+ </plugin>
+ </plugins>
+ </reporting>
+</project>