/**
 * 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.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import net.minidev.json.JSONObject;
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.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.tajo.*;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.statistics.TableStats;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.exception.TajoInternalError;
import org.apache.tajo.exception.UnsupportedException;
import org.apache.tajo.plan.LogicalPlan;
import org.apache.tajo.plan.expr.EvalNode;
import org.apache.tajo.plan.logical.LogicalNode;
import org.apache.tajo.plan.logical.NodeType;
import org.apache.tajo.storage.fragment.FileFragment;
import org.apache.tajo.storage.fragment.Fragment;
import org.apache.tajo.util.Bytes;
import org.apache.tajo.util.TUtil;

import javax.annotation.Nullable;
import java.io.IOException;
import java.net.URI;
import java.text.NumberFormat;
import java.util.*;

import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED_DEFAULT;

public class FileTablespace extends Tablespace {

  public static final PathFilter hiddenFileFilter = new PathFilter() {
    public boolean accept(Path p) {
      String name = p.getName();
      return !name.startsWith("_") && !name.startsWith(".");
    }
  };
  private final Log LOG = LogFactory.getLog(FileTablespace.class);

  static final String OUTPUT_FILE_PREFIX="part-";
  static final ThreadLocal<NumberFormat> OUTPUT_FILE_FORMAT_STAGE =
      new ThreadLocal<NumberFormat>() {
        @Override
        public NumberFormat initialValue() {
          NumberFormat fmt = NumberFormat.getInstance();
          fmt.setGroupingUsed(false);
          fmt.setMinimumIntegerDigits(2);
          return fmt;
        }
      };
  static final ThreadLocal<NumberFormat> OUTPUT_FILE_FORMAT_TASK =
      new ThreadLocal<NumberFormat>() {
        @Override
        public NumberFormat initialValue() {
          NumberFormat fmt = NumberFormat.getInstance();
          fmt.setGroupingUsed(false);
          fmt.setMinimumIntegerDigits(6);
          return fmt;
        }
      };

  static final ThreadLocal<NumberFormat> OUTPUT_FILE_FORMAT_SEQ =
      new ThreadLocal<NumberFormat>() {
        @Override
        public NumberFormat initialValue() {
          NumberFormat fmt = NumberFormat.getInstance();
          fmt.setGroupingUsed(false);
          fmt.setMinimumIntegerDigits(3);
          return fmt;
        }
      };

  private static final StorageProperty FileStorageProperties = new StorageProperty("TEXT", true, true, true, false);
  private static final FormatProperty GeneralFileProperties = new FormatProperty(true, false, true);

  protected FileSystem fs;
  protected Path spacePath;
  protected Path stagingRootPath;
  protected boolean blocksMetadataEnabled;
  private static final HdfsVolumeId zeroVolumeId = new HdfsVolumeId(Bytes.toBytes(0));

  public FileTablespace(String spaceName, URI uri, JSONObject config) {
    super(spaceName, uri, config);
  }

  @Override
  protected void storageInit() throws IOException {
    this.spacePath = new Path(uri);
    this.fs = spacePath.getFileSystem(conf);
    this.stagingRootPath = fs.makeQualified(new Path(conf.getVar(TajoConf.ConfVars.STAGING_ROOT_DIR)));
    this.conf.set(DFSConfigKeys.FS_DEFAULT_NAME_KEY, fs.getUri().toString());

    this.blocksMetadataEnabled =
        conf.getBoolean(DFS_HDFS_BLOCKS_METADATA_ENABLED, DFS_HDFS_BLOCKS_METADATA_ENABLED_DEFAULT);

    if (!this.blocksMetadataEnabled) {
      LOG.warn("does not support block metadata. ('dfs.datanode.hdfs-blocks-metadata.enabled')");
    }
  }

  @Override
  public long getTableVolume(TableDesc table, Optional<EvalNode> filter) throws UnsupportedException {
    Path path = new Path(table.getUri());
    ContentSummary summary;
    try {
      summary = fs.getContentSummary(path);
    } catch (IOException e) {
      throw new TajoInternalError(e);
    }
    return summary.getLength();
  }

  @Override
  public URI getRootUri() {
    return fs.getUri();
  }

  public Scanner getFileScanner(TableMeta meta, Schema schema, Path path)
      throws IOException {
    FileStatus status = fs.getFileStatus(path);
    return getFileScanner(meta, schema, path, status);
  }

  public Scanner getFileScanner(TableMeta meta, Schema schema, Path path, FileStatus status)
      throws IOException {
    Fragment fragment = new FileFragment(path.getName(), path, 0, status.getLen());
    return getScanner(meta, schema, fragment, null);
  }

  public FileSystem getFileSystem() {
    return this.fs;
  }

  public void delete(Path tablePath) throws IOException {
    FileSystem fs = tablePath.getFileSystem(conf);
    fs.delete(tablePath, true);
  }

  public boolean exists(Path path) throws IOException {
    FileSystem fileSystem = path.getFileSystem(conf);
    return fileSystem.exists(path);
  }

  @Override
  public URI getTableUri(String databaseName, String tableName) {
    return StorageUtil.concatPath(spacePath, databaseName, tableName).toUri();
  }

  @VisibleForTesting
  public Appender getAppender(TableMeta meta, Schema schema, Path filePath)
      throws IOException {
    return getAppender(null, null, meta, schema, filePath);
  }

  public FileFragment[] split(String tableName) throws IOException {
    Path tablePath = new Path(spacePath, tableName);
    return split(tableName, tablePath, fs.getDefaultBlockSize());
  }

  public FileFragment[] split(String tableName, long fragmentSize) throws IOException {
    Path tablePath = new Path(spacePath, tableName);
    return split(tableName, tablePath, fragmentSize);
  }

  public FileFragment[] split(Path tablePath) throws IOException {
    FileSystem fs = tablePath.getFileSystem(conf);
    return split(tablePath.getName(), tablePath, fs.getDefaultBlockSize());
  }

  public FileFragment[] split(String tableName, Path tablePath) throws IOException {
    return split(tableName, tablePath, fs.getDefaultBlockSize());
  }

  private FileFragment[] split(String tableName, Path tablePath, long size)
      throws IOException {
    FileSystem fs = tablePath.getFileSystem(conf);

    long defaultBlockSize = size;
    List<FileFragment> listTablets = new ArrayList<FileFragment>();
    FileFragment tablet;

    FileStatus[] fileLists = fs.listStatus(tablePath);
    for (FileStatus file : fileLists) {
      long remainFileSize = file.getLen();
      long start = 0;
      if (remainFileSize > defaultBlockSize) {
        while (remainFileSize > defaultBlockSize) {
          tablet = new FileFragment(tableName, file.getPath(), start, defaultBlockSize);
          listTablets.add(tablet);
          start += defaultBlockSize;
          remainFileSize -= defaultBlockSize;
        }
        listTablets.add(new FileFragment(tableName, file.getPath(), start, remainFileSize));
      } else {
        listTablets.add(new FileFragment(tableName, file.getPath(), 0, remainFileSize));
      }
    }

    FileFragment[] tablets = new FileFragment[listTablets.size()];
    listTablets.toArray(tablets);

    return tablets;
  }

  public static FileFragment[] splitNG(Configuration conf, String tableName, TableMeta meta,
                                       Path tablePath, long size)
      throws IOException {
    FileSystem fs = tablePath.getFileSystem(conf);

    long defaultBlockSize = size;
    List<FileFragment> listTablets = new ArrayList<FileFragment>();
    FileFragment tablet;

    FileStatus[] fileLists = fs.listStatus(tablePath);
    for (FileStatus file : fileLists) {
      long remainFileSize = file.getLen();
      long start = 0;
      if (remainFileSize > defaultBlockSize) {
        while (remainFileSize > defaultBlockSize) {
          tablet = new FileFragment(tableName, file.getPath(), start, defaultBlockSize);
          listTablets.add(tablet);
          start += defaultBlockSize;
          remainFileSize -= defaultBlockSize;
        }
        listTablets.add(new FileFragment(tableName, file.getPath(), start, remainFileSize));
      } else {
        listTablets.add(new FileFragment(tableName, file.getPath(), 0, remainFileSize));
      }
    }

    FileFragment[] tablets = new FileFragment[listTablets.size()];
    listTablets.toArray(tablets);

    return tablets;
  }

  public long calculateSize(Path tablePath) throws IOException {
    FileSystem fs = tablePath.getFileSystem(conf);
    long totalSize = 0;

    if (fs.exists(tablePath)) {
      totalSize = fs.getContentSummary(tablePath).getLength();
    }

    return totalSize;
  }

  /////////////////////////////////////////////////////////////////////////////
  // FileInputFormat Area
  /////////////////////////////////////////////////////////////////////////////
  public Path getAppenderFilePath(TaskAttemptId taskAttemptId, Path workDir) {
    if (taskAttemptId == null) {
      // For testcase
      return workDir;
    }
    // The final result of a task will be written in a file named part-ss-nnnnnnn,
    // where ss is the stage id associated with this task, and nnnnnn is the task id.
    Path outFilePath = StorageUtil.concatPath(workDir, TajoConstants.RESULT_DIR_NAME,
        OUTPUT_FILE_PREFIX +
            OUTPUT_FILE_FORMAT_STAGE.get().format(taskAttemptId.getTaskId().getExecutionBlockId().getId()) + "-" +
            OUTPUT_FILE_FORMAT_TASK.get().format(taskAttemptId.getTaskId().getId()) + "-" +
            OUTPUT_FILE_FORMAT_SEQ.get().format(0));
    LOG.info("Output File Path: " + outFilePath);

    return outFilePath;
  }

  /**
   * Proxy PathFilter that accepts a path only if all filters given in the
   * constructor do. Used by the listPaths() to apply the built-in
   * hiddenFileFilter together with a user provided one (if any).
   */
  private static class MultiPathFilter implements PathFilter {
    private List<PathFilter> filters;

    public MultiPathFilter(List<PathFilter> filters) {
      this.filters = filters;
    }

    public boolean accept(Path path) {
      for (PathFilter filter : filters) {
        if (!filter.accept(path)) {
          return false;
        }
      }
      return true;
    }
  }

  /**
   * List input directories.
   * Subclasses may override to, e.g., select only files matching a regular
   * expression.
   *
   * @return array of FileStatus objects
   * @throws IOException if zero items.
   */
  protected List<FileStatus> listStatus(Path... dirs) throws IOException {
    List<FileStatus> result = new ArrayList<FileStatus>();
    if (dirs.length == 0) {
      throw new IOException("No input paths specified in job");
    }

    List<IOException> errors = new ArrayList<IOException>();

    // creates a MultiPathFilter with the hiddenFileFilter and the
    // user provided one (if any).
    List<PathFilter> filters = new ArrayList<PathFilter>();
    filters.add(hiddenFileFilter);

    PathFilter inputFilter = new MultiPathFilter(filters);

    for (int i = 0; i < dirs.length; ++i) {
      Path p = dirs[i];

      FileStatus[] matches = fs.globStatus(p, inputFilter);
      if (matches == null) {
        LOG.warn("Input path does not exist: " + p);
      } else if (matches.length == 0) {
        LOG.warn("Input Pattern " + p + " matches 0 files");
      } else {
        for (FileStatus globStat : matches) {
          if (globStat.isDirectory()) {
            for (FileStatus stat : fs.listStatus(globStat.getPath(), inputFilter)) {
              result.add(stat);
            }
          } else {
            result.add(globStat);
          }
        }
      }
    }

    if (!errors.isEmpty()) {
      throw new InvalidInputException(errors);
    }
    LOG.info("Total input paths to process : " + result.size());
    return result;
  }

  /**
   * Is the given filename splitable? Usually, true, but if the file is
   * stream compressed, it will not be.
   * <p/>
   * <code>FileInputFormat</code> implementations can override this and return
   * <code>false</code> to ensure that individual input files are never split-up
   * so that Mappers process entire files.
   *
   *
   * @param path the file name to check
   * @param status get the file length
   * @return is this file isSplittable?
   */
  protected boolean isSplittable(TableMeta meta, Schema schema, Path path, FileStatus status) throws IOException {
    Scanner scanner = getFileScanner(meta, schema, path, status);
    boolean split = scanner.isSplittable();
    scanner.close();
    return split;
  }

  private static final double SPLIT_SLOP = 1.1;   // 10% slop

  protected int getBlockIndex(BlockLocation[] blkLocations,
                              long offset) {
    for (int i = 0; i < blkLocations.length; i++) {
      // is the offset inside this block?
      if ((blkLocations[i].getOffset() <= offset) &&
          (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())) {
        return i;
      }
    }
    BlockLocation last = blkLocations[blkLocations.length - 1];
    long fileLength = last.getOffset() + last.getLength() - 1;
    throw new IllegalArgumentException("Offset " + offset +
        " is outside of file (0.." +
        fileLength + ")");
  }

  /**
   * A factory that makes the split for this class. It can be overridden
   * by sub-classes to make sub-types
   */
  protected FileFragment makeSplit(String fragmentId, Path file, long start, long length) {
    return new FileFragment(fragmentId, file, start, length);
  }

  protected FileFragment makeSplit(String fragmentId, Path file, long start, long length,
                                   String[] hosts) {
    return new FileFragment(fragmentId, file, start, length, hosts);
  }

  protected FileFragment makeSplit(String fragmentId, Path file, BlockLocation blockLocation)
      throws IOException {
    return new FileFragment(fragmentId, file, blockLocation);
  }

  // for Non Splittable. eg, compressed gzip TextFile
  protected FileFragment makeNonSplit(String fragmentId, Path file, long start, long length,
                                      BlockLocation[] blkLocations) throws IOException {

    Map<String, Integer> hostsBlockMap = new HashMap<String, Integer>();
    for (BlockLocation blockLocation : blkLocations) {
      for (String host : blockLocation.getHosts()) {
        if (hostsBlockMap.containsKey(host)) {
          hostsBlockMap.put(host, hostsBlockMap.get(host) + 1);
        } else {
          hostsBlockMap.put(host, 1);
        }
      }
    }

    List<Map.Entry<String, Integer>> entries = new ArrayList<Map.Entry<String, Integer>>(hostsBlockMap.entrySet());
    Collections.sort(entries, new Comparator<Map.Entry<String, Integer>>() {

      @Override
      public int compare(Map.Entry<String, Integer> v1, Map.Entry<String, Integer> v2) {
        return v1.getValue().compareTo(v2.getValue());
      }
    });

    String[] hosts = new String[blkLocations[0].getHosts().length];

    for (int i = 0; i < hosts.length; i++) {
      Map.Entry<String, Integer> entry = entries.get((entries.size() - 1) - i);
      hosts[i] = entry.getKey();
    }
    return new FileFragment(fragmentId, file, start, length, hosts);
  }

  /**
   * Get the minimum split size
   *
   * @return the minimum number of bytes that can be in a split
   */
  public long getMinSplitSize() {
    return conf.getLongVar(TajoConf.ConfVars.MINIMUM_SPLIT_SIZE);
  }

  /**
   * Get Disk Ids by Volume Bytes
   */
  private int[] getDiskIds(VolumeId[] volumeIds) {
    int[] diskIds = new int[volumeIds.length];
    for (int i = 0; i < volumeIds.length; i++) {
      int diskId = -1;
      if (volumeIds[i] != null && volumeIds[i].hashCode() > 0) {
        diskId = volumeIds[i].hashCode() - zeroVolumeId.hashCode();
      }
      diskIds[i] = diskId;
    }
    return diskIds;
  }

  /**
   * Generate the list of files and make them into FileSplits.
   *
   * @throws IOException
   */
  public List<Fragment> getSplits(String tableName, TableMeta meta, Schema schema, Path... inputs)
      throws IOException {
    // generate splits'

    List<Fragment> splits = Lists.newArrayList();
    List<Fragment> volumeSplits = Lists.newArrayList();
    List<BlockLocation> blockLocations = Lists.newArrayList();

    for (Path p : inputs) {
      ArrayList<FileStatus> files = Lists.newArrayList();
      if (fs.isFile(p)) {
        files.addAll(Lists.newArrayList(fs.getFileStatus(p)));
      } else {
        files.addAll(listStatus(p));
      }

      int previousSplitSize = splits.size();
      for (FileStatus file : files) {
        Path path = file.getPath();
        long length = file.getLen();
        if (length > 0) {
          // Get locations of blocks of file
          BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
          boolean splittable = isSplittable(meta, schema, path, file);
          if (blocksMetadataEnabled && fs instanceof DistributedFileSystem) {

            if (splittable) {
              for (BlockLocation blockLocation : blkLocations) {
                volumeSplits.add(makeSplit(tableName, path, blockLocation));
              }
              blockLocations.addAll(Arrays.asList(blkLocations));

            } else { // Non splittable
              long blockSize = blkLocations[0].getLength();
              if (blockSize >= length) {
                blockLocations.addAll(Arrays.asList(blkLocations));
                for (BlockLocation blockLocation : blkLocations) {
                  volumeSplits.add(makeSplit(tableName, path, blockLocation));
                }
              } else {
                splits.add(makeNonSplit(tableName, path, 0, length, blkLocations));
              }
            }

          } else {
            if (splittable) {

              long minSize = Math.max(getMinSplitSize(), 1);

              long blockSize = file.getBlockSize(); // s3n rest api contained block size but blockLocations is one
              long splitSize = Math.max(minSize, blockSize);
              long bytesRemaining = length;

              // for s3
              while (((double) bytesRemaining) / splitSize > SPLIT_SLOP) {
                int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining);
                splits.add(makeSplit(tableName, path, length - bytesRemaining, splitSize,
                    blkLocations[blkIndex].getHosts()));
                bytesRemaining -= splitSize;
              }
              if (bytesRemaining > 0) {
                int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining);
                splits.add(makeSplit(tableName, path, length - bytesRemaining, bytesRemaining,
                    blkLocations[blkIndex].getHosts()));
              }
            } else { // Non splittable
              splits.add(makeNonSplit(tableName, path, 0, length, blkLocations));
            }
          }
        }
      }
      if(LOG.isDebugEnabled()){
        LOG.debug("# of splits per partition: " + (splits.size() - previousSplitSize));
      }
    }

    // Combine original fileFragments with new VolumeId information
    setVolumeMeta(volumeSplits, blockLocations);
    splits.addAll(volumeSplits);
    LOG.info("Total # of splits: " + splits.size());
    return splits;
  }

  private void setVolumeMeta(List<Fragment> splits, final List<BlockLocation> blockLocations)
      throws IOException {

    int locationSize = blockLocations.size();
    int splitSize = splits.size();
    if (locationSize == 0 || splitSize == 0) return;

    if (locationSize != splitSize) {
      // splits and locations don't match up
      LOG.warn("Number of block locations not equal to number of splits: "
          + "#locations=" + locationSize
          + " #splits=" + splitSize);
      return;
    }

    DistributedFileSystem fs = (DistributedFileSystem) this.fs;
    int lsLimit = conf.getInt(DFSConfigKeys.DFS_LIST_LIMIT, DFSConfigKeys.DFS_LIST_LIMIT_DEFAULT);
    int blockLocationIdx = 0;

    Iterator<Fragment> iter = splits.iterator();
    while (locationSize > blockLocationIdx) {

      int subSize = Math.min(locationSize - blockLocationIdx, lsLimit);
      List<BlockLocation> locations = blockLocations.subList(blockLocationIdx, blockLocationIdx + subSize);
      //BlockStorageLocation containing additional volume location information for each replica of each block.
      BlockStorageLocation[] blockStorageLocations = fs.getFileBlockStorageLocations(locations);

      for (BlockStorageLocation blockStorageLocation : blockStorageLocations) {
        ((FileFragment)iter.next()).setDiskIds(getDiskIds(blockStorageLocation.getVolumeIds()));
        blockLocationIdx++;
      }
    }
    LOG.info("# of splits with volumeId " + splitSize);
  }

  private static class InvalidInputException extends IOException {
    List<IOException> errors;
    public InvalidInputException(List<IOException> errors) {
      this.errors = errors;
    }

    @Override
    public String getMessage(){
      StringBuffer sb = new StringBuffer();
      int messageLimit = Math.min(errors.size(), 10);
      for (int i = 0; i < messageLimit ; i ++) {
        sb.append(errors.get(i).getMessage()).append("\n");
      }

      if(messageLimit < errors.size())
        sb.append("skipped .....").append("\n");

      return sb.toString();
    }
  }

  @Override
  public List<Fragment> getSplits(String inputSourceId,
                                  TableDesc table,
                                  @Nullable EvalNode filterCondition) throws IOException {
    return getSplits(inputSourceId, table.getMeta(), table.getSchema(), new Path(table.getUri()));
  }

  @Override
  public void createTable(TableDesc tableDesc, boolean ifNotExists) throws IOException {
    if (!tableDesc.isExternal()) {
      String [] splitted = CatalogUtil.splitFQTableName(tableDesc.getName());
      String databaseName = splitted[0];
      String simpleTableName = splitted[1];

      // create a table directory (i.e., ${WAREHOUSE_DIR}/${DATABASE_NAME}/${TABLE_NAME} )
      Path tablePath = StorageUtil.concatPath(spacePath, databaseName, simpleTableName);
      tableDesc.setUri(tablePath.toUri());
    } else {
      Preconditions.checkState(tableDesc.getUri() != null, "ERROR: LOCATION must be given.");
    }

    Path path = new Path(tableDesc.getUri());

    FileSystem fs = path.getFileSystem(conf);
    TableStats stats = new TableStats();
    if (tableDesc.isExternal()) {
      if (!fs.exists(path)) {
        LOG.error(path.toUri() + " does not exist");
        throw new IOException("ERROR: " + path.toUri() + " does not exist");
      }
    } else {
      fs.mkdirs(path);
    }

    long totalSize = 0;

    try {
      totalSize = calculateSize(path);
    } catch (IOException e) {
      LOG.warn("Cannot calculate the size of the relation", e);
    }

    stats.setNumBytes(totalSize);

    if (tableDesc.isExternal()) { // if it is an external table, there is no way to know the exact row number without processing.
      stats.setNumRows(TajoConstants.UNKNOWN_ROW_NUMBER);
    }

    tableDesc.setStats(stats);
  }

  @Override
  public void purgeTable(TableDesc tableDesc) throws IOException {
    try {
      Path path = new Path(tableDesc.getUri());
      FileSystem fs = path.getFileSystem(conf);
      LOG.info("Delete table data dir: " + path);
      fs.delete(path, true);
    } catch (IOException e) {
      throw new InternalError(e.getMessage());
    }
  }

  @Override
  public StorageProperty getProperty() {
    return FileStorageProperties;
  }

  @Override
  public FormatProperty getFormatProperty(TableMeta meta) {
    return GeneralFileProperties;
  }

  @Override
  public void close() {
  }

  @Override
  public void prepareTable(LogicalNode node) throws IOException {
  }

  @Override
  public void rollbackTable(LogicalNode node) throws IOException {
  }

  @Override
  public URI getStagingUri(OverridableConf context, String queryId, TableMeta meta) throws IOException {
    String outputPath = context.get(QueryVars.OUTPUT_TABLE_URI, "");

    Path stagingDir;
    // The fact that there is no output means that this query is neither CTAS or INSERT (OVERWRITE) INTO
    // So, this query results won't be materialized as a part of a table.
    // The result will be temporarily written in the staging directory.
    if (outputPath.isEmpty()) {
      // for temporarily written in the storage directory
      stagingDir = fs.makeQualified(new Path(stagingRootPath, queryId));
    } else {
      Tablespace space = TablespaceManager.get(outputPath);
      if (space.getProperty().isMovable()) { // checking if this tablespace allows MOVE operation
        // If this space allows move operation, the staging directory will be underneath the final output table uri.
        stagingDir = fs.makeQualified(StorageUtil.concatPath(outputPath, TMP_STAGING_DIR_PREFIX, queryId));
      } else {
        stagingDir = fs.makeQualified(new Path(stagingRootPath, queryId));
      }
    }

    return stagingDir.toUri();
  }

  // query submission directory is private!
  final public static FsPermission STAGING_DIR_PERMISSION = FsPermission.createImmutable((short) 0700); // rwx--------
  public static final String TMP_STAGING_DIR_PREFIX = ".staging";

  public URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf context, TableMeta meta)
      throws IOException {

    String realUser;
    String currentUser;
    UserGroupInformation ugi;
    ugi = UserGroupInformation.getLoginUser();
    realUser = ugi.getShortUserName();
    currentUser = UserGroupInformation.getCurrentUser().getShortUserName();


    Path stagingDir = new Path(getStagingUri(context, queryId, meta));

    ////////////////////////////////////////////
    // Create Output Directory
    ////////////////////////////////////////////

    if (fs.exists(stagingDir)) {
      throw new IOException("The staging directory '" + stagingDir + "' already exists");
    }
    fs.mkdirs(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
    FileStatus fsStatus = fs.getFileStatus(stagingDir);
    String owner = fsStatus.getOwner();

    if (!owner.isEmpty() && !(owner.equals(currentUser) || owner.equals(realUser))) {
      throw new IOException("The ownership on the user's query " +
          "directory " + stagingDir + " is not as expected. " +
          "It is owned by " + owner + ". The directory must " +
          "be owned by the submitter " + currentUser + " or " +
          "by " + realUser);
    }

    if (!fsStatus.getPermission().equals(STAGING_DIR_PERMISSION)) {
      LOG.info("Permissions on staging directory " + stagingDir + " are " +
          "incorrect: " + fsStatus.getPermission() + ". Fixing permissions " +
          "to correct value " + STAGING_DIR_PERMISSION);
      fs.setPermission(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
    }

    Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);
    fs.mkdirs(stagingResultDir);

    return stagingDir.toUri();
  }

  @Override
  public void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema) {
  }

  @Override
  public Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId, LogicalPlan plan,
                          Schema schema, TableDesc tableDesc) throws IOException {
    return commitOutputData(queryContext, true);
  }

  @Override
  public TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
                                          Schema inputSchema, SortSpec[] sortSpecs, TupleRange dataRange)
      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 changeFileSeq If true change result file name with max sequence.
   * @return Saved path
   * @throws java.io.IOException
   */
  protected Path commitOutputData(OverridableConf queryContext, 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_URI, "").isEmpty()) {
      finalOutputDir = new Path(queryContext.get(QueryVars.OUTPUT_TABLE_URI));
      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, 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, 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;
  }


}