/**
 * 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.annotations.VisibleForTesting;
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.*;
import org.apache.tajo.*;
import org.apache.tajo.catalog.*;
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.LogicalPlanRewriteRule;
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.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,
      TaskAttemptId.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();


  /**
   * Clear all class cache
   */
  @VisibleForTesting
  protected synchronized static void clearCache() {
    CONSTRUCTOR_CACHE.clear();
    SCANNER_HANDLER_CACHE.clear();
    APPENDER_HANDLER_CACHE.clear();
    storageManagers.clear();
  }

  /**
   * 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 static void close() throws IOException {
    synchronized(storageManagers) {
      for (StorageManager eachStorageManager: storageManagers.values()) {
        eachStorageManager.closeStorageManager();
      }
    }
    clearCache();
  }

  /**
   * 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 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, String storeType) throws IOException {
    if ("HBASE".equalsIgnoreCase(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 {
    FileSystem fileSystem = TajoConf.getWarehouseDir(tajoConf).getFileSystem(tajoConf);
    if (fileSystem != null) {
      return getStorageManager(tajoConf, storeType, fileSystem.getUri().toString());
    } else {
      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
   */
  private static synchronized StorageManager getStorageManager (
      TajoConf tajoConf, StoreType storeType, String managerKey) throws IOException {

    String typeName;
    switch (storeType) {
      case HBASE:
        typeName = "hbase";
        break;
      default:
        typeName = "hdfs";
    }

    synchronized (storageManagers) {
      String storeKey = typeName + "_" + managerKey;
      StorageManager manager = storageManagers.get(storeKey);

      if (manager == null) {
        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,
                              TaskAttemptId taskAttemptId, TableMeta meta, Schema schema, Path workDir)
      throws IOException {
    Appender appender;

    Class<? extends Appender> appenderClass;

    String handlerName = CatalogUtil.getStoreTypeString(meta.getStoreType()).toLowerCase();
    appenderClass = APPENDER_HANDLER_CACHE.get(handlerName);
    if (appenderClass == null) {
      appenderClass = conf.getClass(
          String.format("tajo.storage.appender-handler.%s.class", handlerName), 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, TaskAttemptId 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 = CatalogUtil.getStoreTypeString(storeType).toLowerCase();
    Class<? extends Scanner> scannerClass = SCANNER_HANDLER_CACHE.get(handlerName);
    if (scannerClass == null) {
      scannerClass = conf.getClass(
          String.format("tajo.storage.scanner-handler.%s.class", handlerName), 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<LogicalPlanRewriteRule> 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));
      try {
        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);
          ContentSummary summary = fs.getContentSummary(stagingResultDir);

          // When inserting empty data into a partitioned table, check if keep existing data need to be remove or not.
          boolean overwriteEnabled = queryContext.getBool(SessionVars.PARTITION_NO_RESULT_OVERWRITE_ENABLED);

          // If existing data doesn't need to keep, check if there are some files.
          if ( (!queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty())
            && (!overwriteEnabled || (overwriteEnabled && summary.getFileCount() > 0L))) {
            // 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);
              }

              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());
              }

            } 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);

                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);

          if (queryType != null && queryType.equals(NodeType.INSERT.name())) { // INSERT INTO an existing table

            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;
                }
                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);
              }
            }
            // 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);
          }
        }

        // 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;
  }
}