diff --git a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/AbstractCatalogClient.java b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/AbstractCatalogClient.java
index b0acbb12a4..ec139c2ceb 100644
--- a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/AbstractCatalogClient.java
+++ b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/AbstractCatalogClient.java
@@ -25,6 +25,7 @@
 import org.apache.tajo.catalog.CatalogProtocol.CatalogProtocolService.BlockingInterface;
 import org.apache.tajo.catalog.CatalogProtocol.*;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.*;
 import org.apache.tajo.common.TajoDataTypes.DataType;
 import org.apache.tajo.conf.TajoConf;
@@ -957,4 +958,65 @@ public void updateTableStats(final UpdateTableStatsProto updateTableStatsProto)
       throw new RuntimeException(e);
     }
   }
+
+  @Override
+  public void addDirectOutputCommitHistory(DirectOutputCommitHistoryProto history) throws DuplicateQueryIdException
+    , InsufficientPrivilegeException{
+    try {
+      final BlockingInterface stub = getStub();
+      final ReturnState state = stub.addDirectOutputCommitHistory(null, history);
+
+      throwsIfThisError(state, DuplicateQueryIdException.class);
+      throwsIfThisError(state, InsufficientPrivilegeException.class);
+      ensureOk(state);
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void updateDirectOutputCommitHistoryProto(UpdateDirectOutputCommitHistoryProto history)
+    throws UndefinedQueryIdException, InsufficientPrivilegeException {
+    try {
+      final BlockingInterface stub = getStub();
+      final ReturnState state = stub.updateDirectOutputCommitHistory(null, history);
+
+      throwsIfThisError(state, UndefinedQueryIdException.class);
+      throwsIfThisError(state, InsufficientPrivilegeException.class);
+      ensureOk(state);
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getAllDirectOutputCommitHistories() {
+    try {
+      final BlockingInterface stub = getStub();
+      final DirectOutputCommitHistoryListResponse response = stub.getAllDirectOutputCommitHistories(null,
+        ProtoUtil.NULL_PROTO);
+
+      ensureOk(response.getState());
+      return response.getCommitHistoryList();
+
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getIncompleteDirectOutputCommitHistories() {
+    try {
+      final BlockingInterface stub = getStub();
+      final DirectOutputCommitHistoryListResponse response = stub.getIncompleteDirectOutputCommitHistories(null,
+        ProtoUtil.NULL_PROTO);
+
+      ensureOk(response.getState());
+      return response.getCommitHistoryList();
+
+    } catch (ServiceException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
 }
diff --git a/tajo-catalog/tajo-catalog-client/src/main/proto/CatalogProtocol.proto b/tajo-catalog/tajo-catalog-client/src/main/proto/CatalogProtocol.proto
index 33fe91e1a3..f5f1dacca7 100644
--- a/tajo-catalog/tajo-catalog-client/src/main/proto/CatalogProtocol.proto
+++ b/tajo-catalog/tajo-catalog-client/src/main/proto/CatalogProtocol.proto
@@ -148,4 +148,10 @@ service CatalogProtocolService {
   rpc getFunctions(NullProto) returns (GetFunctionsResponse);
   rpc getFunctionMeta(GetFunctionMetaRequest) returns (FunctionResponse);
   rpc containFunction(ContainFunctionRequest) returns (ReturnState);
+
+  rpc addDirectOutputCommitHistory(DirectOutputCommitHistoryProto) returns (ReturnState);
+  rpc updateDirectOutputCommitHistory(UpdateDirectOutputCommitHistoryProto) returns (ReturnState);
+  rpc getAllDirectOutputCommitHistories(NullProto) returns (DirectOutputCommitHistoryListResponse);
+  rpc getIncompleteDirectOutputCommitHistories(NullProto) returns (DirectOutputCommitHistoryListResponse);
+
 }
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
index f2acf983db..a26643ac7c 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogConstants.java
@@ -49,6 +49,7 @@ public class CatalogConstants {
   public static final String TB_PARTITION_METHODS = "PARTITION_METHODS";
   public static final String TB_PARTTIONS = "PARTITIONS";
   public static final String TB_PARTTION_KEYS = "PARTITION_KEYS";
+  public static final String TB_DIRECT_OUTPUT_COMMIT_HISTORIES = "DIRECT_OUTPUT_COMMIT_HISTORIES";
 
   public static final String COL_TABLESPACE_PK = "SPACE_ID";
   public static final String COL_DATABASES_PK = "DB_ID";
diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogService.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogService.java
index 54d22e7878..b700312939 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogService.java
+++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogService.java
@@ -257,4 +257,17 @@ void alterTable(AlterTableDesc desc)
        NotImplementedException;
 
   void updateTableStats(UpdateTableStatsProto stats) throws UndefinedTableException, InsufficientPrivilegeException;
+
+
+  /************************** DirectOutputCommitter *****************************/
+
+  void addDirectOutputCommitHistory(DirectOutputCommitHistoryProto history) throws DuplicateQueryIdException
+    , InsufficientPrivilegeException;
+
+  void updateDirectOutputCommitHistoryProto(UpdateDirectOutputCommitHistoryProto history)
+    throws UndefinedQueryIdException, InsufficientPrivilegeException;
+
+  List<DirectOutputCommitHistoryProto> getAllDirectOutputCommitHistories();
+
+  List<DirectOutputCommitHistoryProto> getIncompleteDirectOutputCommitHistories();
 }
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index d064c62586..984db05c65 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -429,3 +429,26 @@ message PartitionListResponse {
   required ReturnState state = 1;
   repeated PartitionDescProto partition = 2;
 }
+
+
+////////////////////////////////////////////////
+// DirectOutputCommitter
+////////////////////////////////////////////////
+
+message DirectOutputCommitHistoryProto {
+  required string query_id = 1;
+  required string path = 2;
+  required int64 start_time = 3;
+  optional int64 end_time = 4;
+  required string query_state = 5;
+}
+
+message DirectOutputCommitHistoryListResponse {
+  required ReturnState state = 1;
+  repeated DirectOutputCommitHistoryProto commitHistory = 2;
+}
+
+message UpdateDirectOutputCommitHistoryProto {
+  required string query_id = 1;
+  required string query_state = 2;
+}
diff --git a/tajo-catalog/tajo-catalog-drivers/tajo-hive/src/main/java/org/apache/tajo/catalog/store/HiveCatalogStore.java b/tajo-catalog/tajo-catalog-drivers/tajo-hive/src/main/java/org/apache/tajo/catalog/store/HiveCatalogStore.java
index 6aa058561e..772ef4973f 100644
--- a/tajo-catalog/tajo-catalog-drivers/tajo-hive/src/main/java/org/apache/tajo/catalog/store/HiveCatalogStore.java
+++ b/tajo-catalog/tajo-catalog-drivers/tajo-hive/src/main/java/org/apache/tajo/catalog/store/HiveCatalogStore.java
@@ -1294,4 +1294,24 @@ public List<TableDescriptorProto> getAllTables() {
   public List<TablespaceProto> getTablespaces() {
     return Lists.newArrayList(getTablespace(TajoConstants.DEFAULT_TABLESPACE_NAME));
   }
+
+  @Override
+  public void addDirectOutputCommitHistory(DirectOutputCommitHistoryProto history) {
+    throw new TajoRuntimeException(new UnsupportedException());
+  }
+
+  @Override
+  public void updateDirectOutputCommitHistoryProto(UpdateDirectOutputCommitHistoryProto history) {
+    throw new TajoRuntimeException(new UnsupportedException());
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getAllDirectOutputCommitHistories() {
+    throw new TajoRuntimeException(new UnsupportedException());
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getIncompleteDirectOutputCommitHistories() {
+    throw new TajoRuntimeException(new UnsupportedException());
+  }
 }
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
index df40b9b751..59943a5457 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
@@ -45,6 +45,7 @@
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.conf.TajoConf.ConfVars;
 import org.apache.tajo.rpc.BlockingRpcServer;
+import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.NullProto;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.ReturnState;
 import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.StringListResponse;
@@ -1613,6 +1614,84 @@ public ReturnState containFunction(RpcController controller, ContainFunctionRequ
         return returnError(t);
       }
     }
+
+    @Override
+    public ReturnState addDirectOutputCommitHistory(RpcController controller, DirectOutputCommitHistoryProto request)
+      throws ServiceException {
+      wlock.lock();
+      try {
+        store.addDirectOutputCommitHistory(request);
+        return OK;
+
+      } catch (Throwable t) {
+        printStackTraceIfError(LOG, t);
+        return returnError(t);
+
+      } finally {
+        wlock.unlock();
+      }
+    }
+
+    @Override
+    public ReturnState updateDirectOutputCommitHistory(RpcController controller,
+      UpdateDirectOutputCommitHistoryProto request) throws ServiceException {
+      wlock.lock();
+      try {
+        store.updateDirectOutputCommitHistoryProto(request);
+        return OK;
+
+      } catch (Throwable t) {
+        printStackTraceIfError(LOG, t);
+        return returnError(t);
+
+      } finally {
+        wlock.unlock();
+      }
+    }
+
+    @Override
+    public DirectOutputCommitHistoryListResponse getAllDirectOutputCommitHistories(RpcController controller,
+      NullProto request) throws ServiceException {
+      rlock.lock();
+      try {
+        return DirectOutputCommitHistoryListResponse.newBuilder()
+          .setState(OK)
+          .addAllCommitHistory(store.getAllDirectOutputCommitHistories())
+          .build();
+
+      } catch (Throwable t) {
+        printStackTraceIfError(LOG, t);
+
+        return DirectOutputCommitHistoryListResponse.newBuilder()
+          .setState(returnError(t))
+          .build();
+
+      } finally {
+        rlock.unlock();
+      }
+    }
+
+    @Override
+    public DirectOutputCommitHistoryListResponse getIncompleteDirectOutputCommitHistories(RpcController controller,
+      NullProto request) throws ServiceException {
+      rlock.lock();
+      try {
+        return DirectOutputCommitHistoryListResponse.newBuilder()
+          .setState(OK)
+          .addAllCommitHistory(store.getIncompleteDirectOutputCommitHistories())
+          .build();
+
+      } catch (Throwable t) {
+        printStackTraceIfError(LOG, t);
+
+        return DirectOutputCommitHistoryListResponse.newBuilder()
+          .setState(returnError(t))
+          .build();
+
+      } finally {
+        rlock.unlock();
+      }
+    }
   }
 
   private static class FunctionSignature {
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
index 88fabe2b8a..4eda0b42b2 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/AbstractDBStore.java
@@ -3024,6 +3024,166 @@ private boolean existColumn(final int tableId, final String columnName) {
     return exist;
   }
 
+  private boolean existQueryIdFromDirectOutputCommitHistories(String queryId) {
+    ResultSet res = null;
+    boolean exist = false;
+
+    String sql = "SELECT PATH FROM " + TB_DIRECT_OUTPUT_COMMIT_HISTORIES + " WHERE QUERY_ID = ? ";
+
+    if (LOG.isDebugEnabled()) {
+      LOG.debug(sql);
+    }
+
+    try (PreparedStatement pstmt = getConnection().prepareStatement(sql)) {
+      pstmt.setString(1, queryId);
+      res = pstmt.executeQuery();
+      exist = res.next();
+    } catch (SQLException se) {
+      throw new TajoInternalError(se);
+    } finally {
+      CatalogUtil.closeQuietly(res);
+    }
+
+    return exist;
+  }
+  @Override
+  public void addDirectOutputCommitHistory(DirectOutputCommitHistoryProto history) throws DuplicateQueryIdException {
+
+    if (existQueryIdFromDirectOutputCommitHistories(history.getQueryId())) {
+      throw new DuplicateQueryIdException(history.getQueryId());
+    }
+
+    Connection conn = null;
+    PreparedStatement pstmt = null;
+
+    try {
+      conn = getConnection();
+      conn.setAutoCommit(false);
+
+      final String insertHistorySql = "INSERT INTO " + TB_DIRECT_OUTPUT_COMMIT_HISTORIES
+        + " (QUERY_ID, PATH, START_TIME, QUERY_STATE) VALUES (?, ?, ?, ?) ";
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug(insertHistorySql);
+      }
+
+      pstmt = conn.prepareStatement(insertHistorySql);
+      pstmt.setString(1, history.getQueryId());
+      pstmt.setString(2, history.getPath());
+      pstmt.setLong(3, history.getStartTime());
+      pstmt.setString(4, history.getQueryState());
+      pstmt.executeUpdate();
+      conn.commit();
+    } catch (SQLException se) {
+      if (conn != null) {
+        try {
+          conn.rollback();
+        } catch (SQLException e) {
+          LOG.error(e, e);
+        }
+      }
+      throw new TajoInternalError(se);
+    } finally {
+      CatalogUtil.closeQuietly(pstmt);
+    }
+  }
+
+  @Override
+  public void updateDirectOutputCommitHistoryProto(UpdateDirectOutputCommitHistoryProto history)
+    throws UndefinedQueryIdException {
+
+    if (!existQueryIdFromDirectOutputCommitHistories(history.getQueryId())) {
+      throw new UndefinedQueryIdException(history.getQueryId());
+    }
+
+    Connection conn = null;
+    PreparedStatement pstmt = null;
+
+    try {
+      conn = getConnection();
+      conn.setAutoCommit(false);
+
+      final String updateHistorySql = "UPDATE " + TB_DIRECT_OUTPUT_COMMIT_HISTORIES
+        + " SET END_TIME = ?, QUERY_STATE = ? WHERE QUERY_ID = ? ";
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug(updateHistorySql);
+      }
+
+      pstmt = conn.prepareStatement(updateHistorySql);
+      pstmt.setLong(1, System.currentTimeMillis());
+      pstmt.setString(2, history.getQueryState());
+      pstmt.setString(3, history.getQueryId());
+      pstmt.executeUpdate();
+      conn.commit();
+    } catch (SQLException se) {
+      if (conn != null) {
+        try {
+          conn.rollback();
+        } catch (SQLException e) {
+          LOG.error(e, e);
+        }
+      }
+      throw new TajoInternalError(se);
+    } finally {
+      CatalogUtil.closeQuietly(pstmt);
+    }
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getAllDirectOutputCommitHistories() {
+    List<DirectOutputCommitHistoryProto> histories = new ArrayList<>();
+
+    String sql = "SELECT * FROM " + TB_DIRECT_OUTPUT_COMMIT_HISTORIES;
+
+    try (Statement stmt = getConnection().createStatement();
+      ResultSet rs = stmt.executeQuery(sql)) {
+      while (rs.next()) {
+        DirectOutputCommitHistoryProto.Builder builder = DirectOutputCommitHistoryProto.newBuilder();
+
+        builder.setQueryId(rs.getString("QUERY_ID"));
+        builder.setPath(rs.getString("PATH"));
+        builder.setStartTime(rs.getLong("START_TIME"));
+        builder.setEndTime(rs.getLong("END_TIME"));
+        builder.setQueryState(rs.getString("QUERY_STATE"));
+        histories.add(builder.build());
+      }
+
+    } catch (SQLException e) {
+      throw new TajoInternalError(e);
+    }
+
+    return histories;
+  }
+
+  @Override
+  public List<DirectOutputCommitHistoryProto> getIncompleteDirectOutputCommitHistories() {
+    List<DirectOutputCommitHistoryProto> histories = new ArrayList<>();
+
+    String sql = "SELECT * FROM " + TB_DIRECT_OUTPUT_COMMIT_HISTORIES
+      + " WHERE END_TIME IS NULL "
+      + " OR QUERY_STATE NOT IN ('QUERY_SUCCEEDED', 'QUERY_FAILED', 'QUERY_KILLED', 'QUERY_ERROR')";
+
+    try (Statement stmt = getConnection().createStatement();
+         ResultSet rs = stmt.executeQuery(sql)) {
+      while (rs.next()) {
+        DirectOutputCommitHistoryProto.Builder builder = DirectOutputCommitHistoryProto.newBuilder();
+
+        builder.setQueryId(rs.getString("QUERY_ID"));
+        builder.setPath(rs.getString("PATH"));
+        builder.setStartTime(rs.getLong("START_TIME"));
+        builder.setEndTime(rs.getLong("END_TIME"));
+        builder.setQueryState(rs.getString("QUERY_STATE"));
+        histories.add(builder.build());
+      }
+
+    } catch (SQLException e) {
+      throw new TajoInternalError(e);
+    }
+
+    return histories;
+  }
+
   class PartitionFilterSet {
     private String columnName;
     private List<Pair<Type, Object>> parameters;
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
index 52889791c3..f8e203dbd3 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/CatalogStore.java
@@ -226,4 +226,15 @@ boolean existIndexesByTable(String databaseName, String tableName)
   void existFunction(FunctionDesc func);
 
   List<String> getAllFunctionNames();
+
+  /************************** DirectOutputCommitter *****************************/
+
+  void addDirectOutputCommitHistory(DirectOutputCommitHistoryProto history) throws DuplicateQueryIdException;
+
+  void updateDirectOutputCommitHistoryProto(UpdateDirectOutputCommitHistoryProto history)
+    throws UndefinedQueryIdException;
+
+  List<DirectOutputCommitHistoryProto> getAllDirectOutputCommitHistories();
+
+  List<DirectOutputCommitHistoryProto> getIncompleteDirectOutputCommitHistories();
 }
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
index 96100e8430..316abd53a3 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/derby/derby.xml
@@ -19,6 +19,7 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+			* 13 - 2016-04-11: Support DirectOutputCommitter for AWS S3 file system (TAJO-2087)
       * 12 - 2015-09-28: Change the variable name storeType to dataFormat (TAJO-1663)
       * 11 - 2015-09-23: Add contents length and file count for partition directory (TAJO-1493)
       * 10 - 2015-09-22: Well support for self-describing data formats (TAJO-1832)
@@ -199,7 +200,22 @@
       <tns:Object name="PARTITION_KEYS_IDX" type="index" dependsOn="PARTITION_KEYS" order="21">
         <tns:sql><![CDATA[CREATE INDEX PARTITION_KEYS_IDX ON PARTITION_KEYS(TID , COLUMN_NAME, PARTITION_VALUE)]]></tns:sql>
       </tns:Object>
-    </tns:objects>
+			<tns:Object name="DIRECT_OUTPUT_COMMIT_HISTORIES" type="table" order="22">
+				<tns:sql><![CDATA[
+				CREATE TABLE DIRECT_OUTPUT_COMMIT_HISTORIES (
+					QUERY_ID VARCHAR(128) NOT NULL,
+					PATH VARCHAR(4096) NOT NULL,
+					START_TIME BIGINT NOT NULL,
+					END_TIME BIGINT,
+					QUERY_STATE VARCHAR(50) NOT NULL,
+					CONSTRAINT DIRECT_OUTPUT_COMMIT_HISTORIES_PK PRIMARY KEY (QUERY_ID)
+				)]]>
+				</tns:sql>
+			</tns:Object>
+			<tns:Object name="DIRECT_OUTPUT_COMMIT_HISTORIES_IDX" type="index" dependsOn="DIRECT_OUTPUT_COMMIT_HISTORIES" order="23">
+				<tns:sql><![CDATA[CREATE INDEX DIRECT_OUTPUT_COMMIT_HISTORIES_IDX ON DIRECT_OUTPUT_COMMIT_HISTORIES(END_TIME , QUERY_STATE)]]></tns:sql>
+			</tns:Object>
+		</tns:objects>
 	</tns:base>
 	<tns:existQueries>
 	  <tns:existQuery type="trigger">
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
index 4611e70792..e1f148ebbc 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mariadb/mariadb.xml
@@ -19,6 +19,7 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+      * 14 - 2016-05-12: Support DirectOutputCommitter for AWS S3 file system (TAJO-2087)
       * 13 - 2016-05-07: Add some missing index to MariaDBStore (TAJO-2117)
       * 12 - 2015-09-28: Change the variable name storeType to dataFormat (TAJO-1663)
       * 11 - 2015-09-23: Add contents length and file count for partition directory (TAJO-1493)
@@ -33,7 +34,7 @@
       * 2 - 2014-06-09: First versioning
       * 1-  Before 2013-03-20
     -->
-  <tns:base version="13">
+  <tns:base version="14">
     <tns:objects>
       <tns:Object order="0" type="table" name="META">
         <tns:sql><![CDATA[CREATE TABLE META (VERSION INT NOT NULL)]]></tns:sql>
@@ -176,6 +177,19 @@
         )]]>
         </tns:sql>
       </tns:Object>
+      <tns:Object order="12" type="table" name="DIRECT_OUTPUT_COMMIT_HISTORIES">
+        <tns:sql><![CDATA[
+        CREATE TABLE DIRECT_OUTPUT_COMMIT_HISTORIES (
+          QUERY_ID VARCHAR(128) NOT NULL,
+          PATH VARCHAR(4096) NOT NULL,
+          START_TIME BIGINT NOT NULL,
+          END_TIME BIGINT,
+          QUERY_STATE VARCHAR(50) NOT NULL,
+          PRIMARY KEY (QUERY_ID),
+          INDEX DIRECT_OUTPUT_COMMIT_HISTORIES_IDX (END_TIME, QUERY_STATE)
+        )]]>
+        </tns:sql>
+      </tns:Object>
     </tns:objects>
   </tns:base>
 
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
index c0dadaa3da..166b3c9f6e 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/mysql/mysql.xml
@@ -19,6 +19,7 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
     Catalog base version history
+    * 13 - 2016-04-11: Support DirectOutputCommitter for AWS S3 file system (TAJO-2087)
     * 12 - 2015-09-28: Change the variable name storeType to dataFormat (TAJO-1663)
     * 11 - 2015-09-23: Add contents length and file count for partition directory (TAJO-1493)
     * 10 - 2015-09-22: Well support for self-describing data formats (TAJO-1832)
@@ -176,6 +177,19 @@
         )]]>
         </tns:sql>
       </tns:Object>
+      <tns:Object order="12" type="table" name="DIRECT_OUTPUT_COMMIT_HISTORIES">
+        <tns:sql><![CDATA[
+        CREATE TABLE DIRECT_OUTPUT_COMMIT_HISTORIES (
+          QUERY_ID VARCHAR(128) NOT NULL,
+          PATH VARCHAR(4096) NOT NULL,
+          START_TIME BIGINT NOT NULL,
+          END_TIME BIGINT,
+          QUERY_STATE VARCHAR(50) NOT NULL,
+          PRIMARY KEY (QUERY_ID),
+          INDEX DIRECT_OUTPUT_COMMIT_HISTORIES_IDX (END_TIME, QUERY_STATE)
+        )]]>
+        </tns:sql>
+      </tns:Object>
     </tns:objects>
   </tns:base>
 
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
index 190270c87d..4e71af2bca 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/oracle/oracle.xml
@@ -19,6 +19,7 @@
 <tns:store xmlns:tns="http://tajo.apache.org/catalogstore" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+			* 13 - 2016-04-11: Support DirectOutputCommitter for AWS S3 file system (TAJO-2087)
       * 12 - 2015-09-28: Change the variable name storeType to dataFormat (TAJO-1663)
       * 11 - 2015-09-23: Add contents length and file count for partition directory (TAJO-1493)
       * 10 - 2015-09-22: Well support for self-describing data formats (TAJO-1832)
@@ -251,6 +252,20 @@
       <tns:Object order="23" type="index" name="PARTITION_KEYS_IDX" dependsOn="PARTITION_KEYS">
         <tns:sql><![CDATA[CREATE INDEX PARTITION_KEYS_IDX on PARTITION_KEYS (TID, COLUMN_NAME, PARTITION_VALUE)]]></tns:sql>
       </tns:Object>
+			<tns:Object order="24" type="table" name="DIRECT_OUTPUT_COMMIT_HISTORIES">
+				<tns:sql><![CDATA[
+				CREATE TABLE DIRECT_OUTPUT_COMMIT_HISTORIES (
+					QUERY_ID VARCHAR(128) NOT NULL PRIMARY KEY,
+					PATH VARCHAR2(4096) NOT NULL,
+					START_TIME NUMBER NOT NULL,
+					END_TIME NUMBER,
+					QUERY_STATE VARCHAR(50) NOT NULL
+				)]]>
+				</tns:sql>
+			</tns:Object>
+			<tns:Object order="25" type="index" name="DIRECT_OUTPUT_COMMIT_HISTORIES_IDX" dependsOn="DIRECT_OUTPUT_COMMIT_HISTORIES">
+				<tns:sql><![CDATA[CREATE INDEX DIRECT_OUTPUT_COMMIT_HISTORIES_IDX on DIRECT_OUTPUT_COMMIT_HISTORIES (END_TIME, QUERY_STATE)]]></tns:sql>
+			</tns:Object>
     </tns:objects>
   </tns:base>
   <tns:existQueries>
diff --git a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
index 33a1fd24c4..b507e5d167 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
+++ b/tajo-catalog/tajo-catalog-server/src/main/resources/schemas/postgresql/postgresql.xml
@@ -21,6 +21,7 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.xsd ">
   <!--
       Catalog base version history
+			* 13 - 2016-04-11: Support DirectOutputCommitter for AWS S3 file system (TAJO-2087)
       * 12 - 2015-09-28: Change the variable name storeType to dataFormat (TAJO-1663)
       * 11 - 2015-09-23: Add contents length and file count for partition directory (TAJO-1493)
       * 10 - 2015-09-22: Well support for self-describing data formats (TAJO-1832)
@@ -197,6 +198,20 @@ xsi:schemaLocation="http://tajo.apache.org/catalogstore ../DBMSSchemaDefinition.
       <tns:Object name="PARTITION_KEYS_IDX" type="index" order="18" dependsOn="PARTITION_KEYS">
         <tns:sql><![CDATA[CREATE INDEX PARTITION_KEYS_IDX on PARTITION_KEYS (TID, COLUMN_NAME, PARTITION_VALUE)]]></tns:sql>
       </tns:Object>
+			<tns:Object name="DIRECT_OUTPUT_COMMIT_HISTORIES" type="table" order="19">
+				<tns:sql><![CDATA[
+				CREATE TABLE DIRECT_OUTPUT_COMMIT_HISTORIES (
+					QUERY_ID VARCHAR(128) NOT NULL PRIMARY KEY,
+					PATH VARCHAR(4096) NOT NULL,
+					START_TIME BIGINT NOT NULL,
+					END_TIME BIGINT,
+					QUERY_STATE VARCHAR(50) NOT NULL
+				)]]>
+				</tns:sql>
+			</tns:Object>
+			<tns:Object name="DIRECT_OUTPUT_COMMIT_HISTORIES_IDX" type="index" order="20" dependsOn="DIRECT_OUTPUT_COMMIT_HISTORIES">
+				<tns:sql><![CDATA[CREATE INDEX DIRECT_OUTPUT_COMMIT_HISTORIES_IDX on DIRECT_OUTPUT_COMMIT_HISTORIES (END_TIME, QUERY_STATE)]]></tns:sql>
+			</tns:Object>
 		</tns:objects>
 	</tns:base>
 	<tns:existQueries>
diff --git a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
index 14d6e303f9..18bdccecc5 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
+++ b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalog.java
@@ -22,6 +22,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.BuiltinStorages;
 import org.apache.tajo.TajoConstants;
+import org.apache.tajo.TajoProtos;
 import org.apache.tajo.catalog.dictionary.InfoSchemaMetadataDictionary;
 import org.apache.tajo.catalog.partition.PartitionDesc;
 import org.apache.tajo.catalog.partition.PartitionMethodDesc;
@@ -1089,7 +1090,7 @@ public final void testFindIntFunc() throws Exception {
 
     assertEquals(retrieved.getFunctionName(), "testint");
     assertEquals(retrieved.getParamTypes()[0], CatalogUtil.newSimpleDataType(Type.INT4));
-    assertEquals(retrieved.getParamTypes()[1] , CatalogUtil.newSimpleDataType(Type.INT4));
+    assertEquals(retrieved.getParamTypes()[1], CatalogUtil.newSimpleDataType(Type.INT4));
   }
 
   @Test(expected=UndefinedFunctionException.class)
@@ -1118,7 +1119,7 @@ public final void testFindFloatFunc() throws Exception {
 
     assertEquals(retrieved.getFunctionName(), "testfloat");
     assertEquals(retrieved.getParamTypes()[0], CatalogUtil.newSimpleDataType(Type.FLOAT8));
-    assertEquals(retrieved.getParamTypes()[1] , CatalogUtil.newSimpleDataType(Type.INT4));
+    assertEquals(retrieved.getParamTypes()[1], CatalogUtil.newSimpleDataType(Type.INT4));
   }
 
   @Test(expected=UndefinedFunctionException.class)
@@ -1157,4 +1158,56 @@ public final void testFindAnyTypeParamFunc() throws Exception {
     assertEquals(retrieved.getFunctionName(), "testany");
     assertEquals(retrieved.getParamTypes()[0], CatalogUtil.newSimpleDataType(Type.ANY));
   }
+
+  @Test
+  public final void testDirectOutputCommitHistories() throws Exception {
+    catalog.addDirectOutputCommitHistory(getDirectOutputCommitHistoryProto(1));
+    catalog.addDirectOutputCommitHistory(getDirectOutputCommitHistoryProto(2));
+
+    List<CatalogProtos.DirectOutputCommitHistoryProto> allHistories = catalog.getAllDirectOutputCommitHistories();
+    assertNotNull(allHistories);
+    assertEquals(2, allHistories.size());
+
+    catalog.updateDirectOutputCommitHistoryProto(getUpdateDirectOutputCommitHistoryProto(1));
+
+    List<CatalogProtos.DirectOutputCommitHistoryProto> incompleteHistories = catalog
+      .getIncompleteDirectOutputCommitHistories();
+
+    assertNotNull(incompleteHistories);
+    assertEquals(1, incompleteHistories.size());
+
+    catalog.updateDirectOutputCommitHistoryProto(getUpdateDirectOutputCommitHistoryProto(2));
+
+    incompleteHistories = catalog.getIncompleteDirectOutputCommitHistories();
+
+    assertNotNull(incompleteHistories);
+    assertEquals(0, incompleteHistories.size());
+
+    allHistories = catalog.getAllDirectOutputCommitHistories();
+    assertNotNull(allHistories);
+    assertEquals(2, allHistories.size());
+
+    for (CatalogProtos.DirectOutputCommitHistoryProto history : allHistories) {
+      assertEquals(TajoProtos.QueryState.QUERY_SUCCEEDED.name(), history.getQueryState());
+      assertNotNull(history.getEndTime());
+    }
+  }
+
+  private CatalogProtos.DirectOutputCommitHistoryProto getDirectOutputCommitHistoryProto(int index) {
+    CatalogProtos.DirectOutputCommitHistoryProto.Builder builder = CatalogProtos.DirectOutputCommitHistoryProto
+      .newBuilder();
+    builder.setQueryId("query_" + index);
+    builder.setPath("hdfs://localhost:9010/table" + index);
+    builder.setStartTime(System.currentTimeMillis());
+    builder.setQueryState(TajoProtos.QueryState.QUERY_RUNNING.name());
+    return builder.build();
+  }
+
+  private CatalogProtos.UpdateDirectOutputCommitHistoryProto getUpdateDirectOutputCommitHistoryProto(int index) {
+    CatalogProtos.UpdateDirectOutputCommitHistoryProto.Builder builder = CatalogProtos
+      .UpdateDirectOutputCommitHistoryProto.newBuilder();
+    builder.setQueryId("query_" + index);
+    builder.setQueryState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+    return builder.build();
+  }
 }
diff --git a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalogExceptions.java b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalogExceptions.java
index 8ba3d59ec9..a330df1295 100644
--- a/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalogExceptions.java
+++ b/tajo-catalog/tajo-catalog-server/src/test/java/org/apache/tajo/catalog/TestCatalogExceptions.java
@@ -20,7 +20,9 @@
 
 import org.apache.hadoop.fs.Path;
 import org.apache.tajo.TajoConstants;
+import org.apache.tajo.TajoProtos;
 import org.apache.tajo.catalog.partition.PartitionDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto.AlterTablespaceCommand;
 import org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto.AlterTablespaceType;
@@ -36,6 +38,10 @@
 import org.junit.Test;
 
 import java.net.URI;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 public class TestCatalogExceptions {
 
@@ -89,11 +95,11 @@ public void testAlterTablespaceWithWrongUri() throws Exception {
   @Test(expected = UndefinedTablespaceException.class)
   public void testAlterUndefinedTablespace() throws Exception {
     catalog.alterTablespace(AlterTablespaceProto.newBuilder().
-        setSpaceName("undefined").
-        addCommand(
-            AlterTablespaceCommand.newBuilder().
-                setType(AlterTablespaceType.LOCATION).
-                setLocation("hdfs://zzz.com/warehouse")).build());
+      setSpaceName("undefined").
+      addCommand(
+        AlterTablespaceCommand.newBuilder().
+          setType(AlterTablespaceType.LOCATION).
+          setLocation("hdfs://zzz.com/warehouse")).build());
   }
 
   @Test(expected = DuplicateDatabaseException.class)
@@ -132,7 +138,7 @@ public void testCreateTableWithWrongUri() throws Exception {
   @Test(expected = DuplicateTableException.class)
   public void testCreateDuplicateTable() throws Exception {
     catalog.createTable(CatalogTestingUtil.buildTableDesc("TestDatabase1", "TestTable1",
-        CommonTestingUtil.getTestDir().toString()));
+      CommonTestingUtil.getTestDir().toString()));
   }
 
   @Test(expected = UndefinedDatabaseException.class)
@@ -267,7 +273,7 @@ public void testRenameUndefinedColumn() throws Exception {
   public void testAddDuplicateColumn() throws Exception {
     AlterTableDesc alterTableDesc = new AlterTableDesc();
     alterTableDesc.setAddColumn(new Column(CatalogUtil.buildFQName("TestDatabase1", "TestTable1", "cOlumn"),
-        CatalogUtil.newSimpleDataType(Type.BLOB)));
+      CatalogUtil.newSimpleDataType(Type.BLOB)));
     alterTableDesc.setTableName(CatalogUtil.buildFQName("TestDatabase1", "TestTable1"));
     alterTableDesc.setAlterTableType(AlterTableType.ADD_COLUMN);
     catalog.alterTable(alterTableDesc);
@@ -279,17 +285,17 @@ public void testCreateDuplicateIndex() throws Exception {
     SortSpec[] colSpecs = new SortSpec[1];
     colSpecs[0] = new SortSpec(tableDesc.getSchema().getColumn(0), true, false);
     catalog.createIndex(
-        new IndexDesc("TestDatabase1", "testTable1",
-            "new_index", new URI("idx_test"), colSpecs,
-            IndexMethod.TWO_LEVEL_BIN_TREE, true, true, tableDesc.getSchema()));
+      new IndexDesc("TestDatabase1", "testTable1",
+        "new_index", new URI("idx_test"), colSpecs,
+        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, tableDesc.getSchema()));
 
     tableDesc = catalog.getTableDesc("TestDatabase1", "testTable1");
     colSpecs = new SortSpec[1];
     colSpecs[0] = new SortSpec(tableDesc.getSchema().getColumn(0), true, false);
     catalog.createIndex(
-        new IndexDesc("TestDatabase1", "testTable1",
-            "new_index", new URI("idx_test"), colSpecs,
-            IndexMethod.TWO_LEVEL_BIN_TREE, true, true, tableDesc.getSchema()));
+      new IndexDesc("TestDatabase1", "testTable1",
+        "new_index", new URI("idx_test"), colSpecs,
+        IndexMethod.TWO_LEVEL_BIN_TREE, true, true, tableDesc.getSchema()));
   }
 
   @Test(expected = UndefinedDatabaseException.class)
@@ -301,4 +307,34 @@ public void testDropIndexOfUndefinedDatabase() throws Exception {
   public void testDropUndefinedIndex() throws Exception {
     catalog.dropIndex("TestDatabase1", "undefined");
   }
+
+  @Test(expected = DuplicateQueryIdException.class)
+  public final void testAddDirectOutputCommitHistories() throws Exception {
+    catalog.addDirectOutputCommitHistory(getDirectOutputCommitHistoryProto(1));
+    catalog.addDirectOutputCommitHistory(getDirectOutputCommitHistoryProto(2));
+    catalog.addDirectOutputCommitHistory(getDirectOutputCommitHistoryProto(2));
+  }
+
+  @Test(expected = UndefinedQueryIdException.class)
+  public final void testUpdateDirectOutputCommitHistory() throws Exception {
+    catalog.updateDirectOutputCommitHistoryProto(getUpdateDirectOutputCommitHistoryProto(10));
+  }
+
+  private CatalogProtos.DirectOutputCommitHistoryProto getDirectOutputCommitHistoryProto(int index) {
+    CatalogProtos.DirectOutputCommitHistoryProto.Builder builder = CatalogProtos.DirectOutputCommitHistoryProto
+      .newBuilder();
+    builder.setQueryId("query_" + index);
+    builder.setPath("hdfs://localhost:9010/table" + index);
+    builder.setStartTime(System.currentTimeMillis());
+    builder.setQueryState(TajoProtos.QueryState.QUERY_RUNNING.name());
+    return builder.build();
+  }
+
+  private CatalogProtos.UpdateDirectOutputCommitHistoryProto getUpdateDirectOutputCommitHistoryProto(int index) {
+    CatalogProtos.UpdateDirectOutputCommitHistoryProto.Builder builder = CatalogProtos
+      .UpdateDirectOutputCommitHistoryProto.newBuilder();
+    builder.setQueryId("query_" + index);
+    builder.setQueryState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+    return builder.build();
+  }
 }
diff --git a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
index ab00a41749..3e3306c322 100644
--- a/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
+++ b/tajo-common/src/main/java/org/apache/tajo/SessionVars.java
@@ -148,6 +148,10 @@ public enum SessionVars implements ConfigKey {
   INDEX_SELECTIVITY_THRESHOLD(ConfVars.$INDEX_SELECTIVITY_THRESHOLD, "the selectivity threshold for index scan",
       DEFAULT),
 
+  // for DirectOutputCommitter
+  DIRECT_OUTPUT_COMMITTER_ENABLED(ConfVars.$DIRECT_OUTPUT_COMMITTER_ENABLED,
+    "If true, a task will write the output data directly to the final location.", DEFAULT),
+
   // for partition overwrite
   PARTITION_NO_RESULT_OVERWRITE_ENABLED(ConfVars.$PARTITION_NO_RESULT_OVERWRITE_ENABLED,
     "If true, a partitioned table is overwritten even if a sub query leads to no result. "
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 6e3eaeae8a..0b11833adf 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -370,6 +370,9 @@ public static enum ConfVars implements ConfigKey {
     $INDEX_ENABLED("tajo.query.index.enabled", false),
     $INDEX_SELECTIVITY_THRESHOLD("tajo.query.index.selectivity.threshold", 0.05f),
 
+    // for DirectOutputCommitter
+    $DIRECT_OUTPUT_COMMITTER_ENABLED("tajo.query.direct-output-committer.enabled", false, Validators.bool()),
+
     // Client -----------------------------------------------------------------
     $CLIENT_SESSION_EXPIRY_TIME("tajo.client.session.expiry-time-sec", 3600), // default time is one hour.
 
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/DuplicateQueryIdException.java b/tajo-common/src/main/java/org/apache/tajo/exception/DuplicateQueryIdException.java
new file mode 100644
index 0000000000..386bb4689c
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/DuplicateQueryIdException.java
@@ -0,0 +1,34 @@
+/**
+ * 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.exception;
+
+import org.apache.tajo.error.Errors;
+import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.ReturnState;
+
+public class DuplicateQueryIdException extends TajoException {
+  private static final long serialVersionUID = 3224521585413794700L;
+
+  public DuplicateQueryIdException(ReturnState state) {
+    super(state);
+  }
+
+  public DuplicateQueryIdException(String signature) {
+    super(Errors.ResultCode.DUPLICATE_QUERY_ID, signature);
+  }
+}
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java b/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
index a6222cf9fb..088ca21149 100644
--- a/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/ErrorMessages.java
@@ -74,6 +74,7 @@ public class ErrorMessages {
     ADD_MESSAGE(UNDEFINED_INDEX_FOR_COLUMNS, "index does not exist for '%s' columns of '%s' table", 2);
     ADD_MESSAGE(UNDEFINED_INDEX_NAME, "index name '%s' does not exist", 1);
     ADD_MESSAGE(UNDEFINED_TABLESPACE_HANDLER, "No tablespace handler for URI scheme '%s'", 1);
+    ADD_MESSAGE(UNDEFINED_QUERY_ID, "queryId '%s' does not exist", 1);
 
     ADD_MESSAGE(DUPLICATE_TABLESPACE, "tablespace '%s' already exists", 1);
     ADD_MESSAGE(DUPLICATE_DATABASE, "database '%s' already exists", 1);
@@ -83,6 +84,7 @@ public class ErrorMessages {
     ADD_MESSAGE(DUPLICATE_ALIAS, "table name '%s' specified more than once", 1);
     ADD_MESSAGE(DUPLICATE_INDEX, "index '%s' already exists", 1);
     ADD_MESSAGE(DUPLICATE_PARTITION, "partition for '%s' already exists", 1);
+    ADD_MESSAGE(DUPLICATE_QUERY_ID, "queryId '%s' already exists", 1);
 
     ADD_MESSAGE(AMBIGUOUS_TABLE, "table name '%s' is ambiguous", 1);
     ADD_MESSAGE(AMBIGUOUS_COLUMN, "column name '%s' is ambiguous", 1);
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/ExceptionUtil.java b/tajo-common/src/main/java/org/apache/tajo/exception/ExceptionUtil.java
index adfb0c4a45..1bd04558a8 100644
--- a/tajo-common/src/main/java/org/apache/tajo/exception/ExceptionUtil.java
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/ExceptionUtil.java
@@ -73,6 +73,7 @@ public class ExceptionUtil {
     ADD_EXCEPTION(UNDEFINED_OPERATOR, UndefinedOperatorException.class);
     ADD_EXCEPTION(UNDEFINED_INDEX_NAME, UndefinedIndexException.class);
     ADD_EXCEPTION(UNDEFINED_TABLESPACE_HANDLER, UndefinedTablespaceHandlerException.class);
+    ADD_EXCEPTION(UNDEFINED_QUERY_ID, UndefinedQueryIdException.class);
 
     ADD_EXCEPTION(DUPLICATE_TABLESPACE, DuplicateTablespaceException.class);
     ADD_EXCEPTION(DUPLICATE_DATABASE, DuplicateDatabaseException.class);
@@ -82,6 +83,7 @@ public class ExceptionUtil {
     // ADD_EXCEPTION(DUPLICATE_ALIAS, );
     ADD_EXCEPTION(DUPLICATE_INDEX, DuplicateIndexException.class);
     ADD_EXCEPTION(DUPLICATE_PARTITION, DuplicatePartitionException.class);
+    ADD_EXCEPTION(DUPLICATE_QUERY_ID, DuplicateQueryIdException.class);
 
     ADD_EXCEPTION(AMBIGUOUS_TABLE, AmbiguousTableException.class);
     ADD_EXCEPTION(AMBIGUOUS_COLUMN, AmbiguousColumnException.class);
diff --git a/tajo-common/src/main/java/org/apache/tajo/exception/UndefinedQueryIdException.java b/tajo-common/src/main/java/org/apache/tajo/exception/UndefinedQueryIdException.java
new file mode 100644
index 0000000000..4171d40adf
--- /dev/null
+++ b/tajo-common/src/main/java/org/apache/tajo/exception/UndefinedQueryIdException.java
@@ -0,0 +1,34 @@
+/**
+ * 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.exception;
+
+import org.apache.tajo.error.Errors.ResultCode;
+import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos.ReturnState;
+
+public class UndefinedQueryIdException extends TajoException {
+	private static final long serialVersionUID = 5062193018697228020L;
+
+  public UndefinedQueryIdException(ReturnState state) {
+    super(state);
+  }
+
+  public UndefinedQueryIdException(String signature) {
+    super(ResultCode.UNDEFINED_QUERY_ID, signature);
+  }
+}
diff --git a/tajo-common/src/main/proto/errors.proto b/tajo-common/src/main/proto/errors.proto
index 6e13eb1490..5df751d9d8 100644
--- a/tajo-common/src/main/proto/errors.proto
+++ b/tajo-common/src/main/proto/errors.proto
@@ -117,6 +117,7 @@ enum ResultCode {
   UNDEFINED_PARTITION_KEY               = 523; // ?
   UNDEFINED_TABLESPACE_HANDLER          = 524; // SQLState: 42T11 - No Tablespace Handler for the URI scheme
   UNDEFINED_PARTITIONS                  = 525; // ?
+  UNDEFINED_QUERY_ID                    = 526; // ?
 
   DUPLICATE_TABLESPACE                  = 531;
   DUPLICATE_DATABASE                    = 532; // SQLState: 42P04
@@ -127,6 +128,7 @@ enum ResultCode {
   DUPLICATE_FUNCTION                    = 537; // SQLState: 42723
   DUPLICATE_INDEX                       = 538; // SQLState: 42T07
   DUPLICATE_PARTITION                   = 539; // SQLState: 42T08
+  DUPLICATE_QUERY_ID                    = 540; // ?
 
   AMBIGUOUS_TABLE                       = 541; // SQLState: 42P09
   AMBIGUOUS_COLUMN                      = 542; // SQLState: 42702;
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
index b87639f56c..f527442d9f 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
@@ -26,22 +26,52 @@
 import org.apache.hadoop.io.compress.DeflateCodec;
 import org.apache.tajo.IntegrationTest;
 import org.apache.tajo.QueryTestCaseBase;
+import org.apache.tajo.SessionVars;
+import org.apache.tajo.TajoConstants;
 import org.apache.tajo.catalog.CatalogService;
 import org.apache.tajo.catalog.CatalogUtil;
 import org.apache.tajo.catalog.TableDesc;
 import org.apache.tajo.util.CommonTestingUtil;
+import org.junit.After;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.io.BufferedReader;
 import java.io.InputStreamReader;
 import java.sql.ResultSet;
-import java.util.List;
+import java.util.*;
 
 import static org.junit.Assert.*;
 
 @Category(IntegrationTest.class)
+@RunWith(Parameterized.class)
 public class TestInsertQuery extends QueryTestCaseBase {
+  private boolean isDirectOutputCommit = false;
+
+  public TestInsertQuery(boolean isDirectOutputCommit) {
+    super(TajoConstants.DEFAULT_DATABASE_NAME);
+
+    this.isDirectOutputCommit = isDirectOutputCommit;
+
+    Map<String, String> variables = new HashMap<>();
+    variables.put(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname(), Boolean.toString(isDirectOutputCommit));
+    client.updateSessionVariables(variables);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    client.unsetSessionVariables(Arrays.asList(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+  }
+
+  @Parameterized.Parameters
+  public static Collection<Object[]> generateParameters() {
+    return Arrays.asList(new Object[][]{
+      {false},
+      {true},
+    });
+  }
 
   @Test
   public final void testInsertOverwrite() throws Exception {
@@ -87,10 +117,16 @@ public final void testInsertInto() throws Exception {
 
     for (int i = 0; i < dataFiles.size(); i++) {
       String name = dataFiles.get(i).getName();
-      assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
-      String[] tokens = name.split("-");
-      assertEquals(4, tokens.length);
-      assertEquals(i, Integer.parseInt(tokens[3]));
+      if (isDirectOutputCommit) {
+        assertTrue(name.matches("UUID-[0-9]*-[0-9]*-[0-9]*-[0-9]*-[0-9]*"));
+        String[] tokens = name.split("-");
+        assertEquals(6, tokens.length);
+      } else {
+        assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+        String[] tokens = name.split("-");
+        assertEquals(4, tokens.length);
+        assertEquals(i, Integer.parseInt(tokens[3]));
+      }
     }
 
     String tableDatas = getTableFileContents("table1");
@@ -156,7 +192,11 @@ public final void assertTestInsertIntoLocation(Path path) throws Exception {
 
       for (FileStatus eachFileStatus : files) {
         String name = eachFileStatus.getPath().getName();
-        assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+        if (isDirectOutputCommit) {
+          assertTrue(name.matches("UUID-[0-9]*-[0-9]*-[0-9]*-[0-9]*-[0-9]*"));
+        } else {
+          assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+        }
       }
 
       executeString("insert into location '" + path + "' select l_orderkey, l_partkey, l_linenumber from default.lineitem").close();
@@ -178,7 +218,11 @@ public final void assertTestInsertIntoLocation(Path path) throws Exception {
 
       for (FileStatus eachFileStatus : files) {
         String name = eachFileStatus.getPath().getName();
-        assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+        if (isDirectOutputCommit) {
+          assertTrue(name.matches("UUID-[0-9]*-[0-9]*-[0-9]*-[0-9]*-[0-9]*"));
+        } else {
+          assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+        }
       }
     } finally {
       if (fs != null) {
@@ -309,9 +353,14 @@ public final void testInsertIntoPartitionedTable() throws Exception {
         assertTrue(eachFileStatus.getPath().getName().indexOf("n_nationkey=") == 0);
         FileStatus[] dataFiles = fs.listStatus(eachFileStatus.getPath());
         assertEquals(2, dataFiles.length);
+
         for (FileStatus eachDataFileStatus: dataFiles) {
           String name = eachDataFileStatus.getPath().getName();
-          assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+          if (isDirectOutputCommit) {
+            assertTrue(name.matches("UUID-[0-9]*-[0-9]*-[0-9]*-[0-9]*-[0-9]*"));
+          } else {
+            assertTrue(name.matches("part-[0-9]*-[0-9]*-[0-9]*"));
+          }
         }
       }
     } finally {
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java b/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
index e74d72223f..b0aa2f8e3a 100644
--- a/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/engine/query/TestTablePartitions.java
@@ -23,10 +23,7 @@
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.apache.hadoop.io.compress.DeflateCodec;
-import org.apache.tajo.QueryId;
-import org.apache.tajo.QueryTestCaseBase;
-import org.apache.tajo.TajoConstants;
-import org.apache.tajo.TajoTestingCluster;
+import org.apache.tajo.*;
 import org.apache.tajo.catalog.*;
 import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.client.TajoClientUtil;
@@ -41,6 +38,8 @@
 import org.apache.tajo.querymaster.QueryMasterTask;
 import org.apache.tajo.storage.StorageConstants;
 import org.apache.tajo.util.CommonTestingUtil;
+import org.apache.tajo.util.KeyValueSet;
+import org.junit.After;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -59,17 +58,28 @@ public class TestTablePartitions extends QueryTestCaseBase {
 
   private NodeType nodeType;
 
-  public TestTablePartitions(NodeType nodeType) throws IOException {
+  public TestTablePartitions(NodeType nodeType, boolean isDirectOutputCommit) throws IOException {
     super(TajoConstants.DEFAULT_DATABASE_NAME);
     this.nodeType = nodeType;
+
+    Map<String, String> variables = new HashMap<>();
+    variables.put(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname(), Boolean.toString(isDirectOutputCommit));
+    client.updateSessionVariables(variables);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    client.unsetSessionVariables(Arrays.asList(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
   }
 
   @Parameters(name = "{index}: {0}")
   public static Collection<Object[]> generateParameters() {
     return Arrays.asList(new Object[][] {
       //type
-      {NodeType.INSERT},
-      {NodeType.CREATE_TABLE},
+      {NodeType.INSERT, false},
+      {NodeType.INSERT, true},
+      {NodeType.CREATE_TABLE, false},
+      {NodeType.CREATE_TABLE, true},
     });
   }
 
@@ -1236,6 +1246,7 @@ public final void testIgnoreFilesInIntermediateDir() throws Exception {
       ResultSet res = executeString("select * from testIgnoreFilesInIntermediateDir;");
       assertFalse(res.next());
       res.close();
+      executeString("DROP TABLE testIgnoreFilesInIntermediateDir PURGE").close();
     }
   }
 
diff --git a/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestDirectOutputCommitter.java b/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestDirectOutputCommitter.java
new file mode 100644
index 0000000000..a739a0316c
--- /dev/null
+++ b/tajo-core-tests/src/test/java/org/apache/tajo/querymaster/TestDirectOutputCommitter.java
@@ -0,0 +1,825 @@
+/**
+ * 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.querymaster;
+
+import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.*;
+import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos;
+import org.apache.tajo.client.TajoClient;
+import org.apache.tajo.ipc.ClientProtos;
+import org.apache.tajo.master.event.QueryEvent;
+import org.apache.tajo.master.event.QueryEventType;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.util.*;
+
+import static org.apache.tajo.TajoConstants.DEFAULT_DATABASE_NAME;
+import static org.junit.Assert.*;
+
+public class TestDirectOutputCommitter {
+
+  private static TajoTestingCluster cluster;
+  private static TajoClient client;
+
+  @Rule
+  public TestName name = new TestName();
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    cluster = TpchTestBase.getInstance().getTestingCluster();
+    client = cluster.newTajoClient();
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    client.unsetSessionVariables(Arrays.asList(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+    client.close();
+  }
+
+  private static void setSessionVariable(boolean value) {
+    Map<String, String> variables = new HashMap<>();
+    variables.put(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname(), Boolean.toString(value));
+    client.updateSessionVariables(variables);
+  }
+
+  @Test(timeout = 10000)
+  public final void testInsertWithMixedSessionVariable() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(false);
+    assertEquals("false", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName + " ( L_ORDERKEY bigint, L_PARTKEY bigint, " +
+      " L_SUPPKEY bigint, L_LINENUMBER bigint, L_QUANTITY double, L_EXTENDEDPRICE double, L_DISCOUNT double, " +
+      " L_TAX double, L_RETURNFLAG text, L_LINESTATUS text, L_SHIPDATE text, L_COMMITDATE text, L_RECEIPTDATE text, " +
+      " L_SHIPINSTRUCT text, L_SHIPMODE text, L_COMMENT text)");
+    res.close();
+
+    TableDesc desc = catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName);
+    assertNotNull(desc);
+
+    // Insert overwrite the data without direct output committer
+    response = client.executeQuery("insert overwrite into " + tableName + " select * from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    // Insert overwrite the data with direct output committer
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    response = client.executeQuery("insert into " + tableName + " select * from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+
+    FileStatus[] files = fs.listStatus(new Path(desc.getUri()));
+    assertEquals(2, files.length);
+
+    // Make sure whether previous data file and new data file exist or not.
+    int existingFileCount = 0;
+    int addedFileCount = 0;
+    for(FileStatus file : files) {
+      if (file.getPath().getName().toString().startsWith(prefix)) {
+        addedFileCount++;
+      } else {
+        existingFileCount++;
+      }
+    }
+    assertEquals(1, existingFileCount);
+    assertEquals(1, addedFileCount);
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testInsertOverwriteWithPartitionedTableByOneColumn() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col1 int4, col2 int4) partition by column(key float8) ");
+    res.close();
+
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    // Insert overwrite the data
+    response = client.executeQuery("insert overwrite into " + tableName
+      + " (col1, col2, key) select l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    // Check the number of partitions and the prefix of each output file
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(6, partitions.size());
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    response = client.executeQuery("insert overwrite into " + tableName
+      + " select l_orderkey, l_partkey, l_quantity from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+
+    }
+    prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+
+    // Make sure whether previous partitions exists or not.
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    protos  = catalog.getAllDirectOutputCommitHistories();
+    historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testInsertOverwriteWithPartitionedTableByThreeColumns() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col4 text) partition by column(col1 int4, col2 int4, col3 float8) ");
+    res.close();
+
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    // Insert overwrite the data
+    response = client.executeQuery("insert overwrite into " + tableName
+      + " select l_returnflag, l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    // Check the number of partitions and the prefix of each output file
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(6, partitions.size());
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    response = client.executeQuery("insert overwrite into " + tableName
+      + " select l_returnflag, l_orderkey, l_partkey, l_quantity from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+
+    }
+    prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+
+    // Make sure whether previous partitions exists or not.
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    protos  = catalog.getAllDirectOutputCommitHistories();
+    historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testRecoverAfterKillQueryWithNonPartitionedTable() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName + " ( L_ORDERKEY bigint, L_PARTKEY bigint, " +
+      " L_SUPPKEY bigint, L_LINENUMBER bigint, L_QUANTITY double, L_EXTENDEDPRICE double, L_DISCOUNT double, " +
+      " L_TAX double, L_RETURNFLAG text, L_LINESTATUS text, L_SHIPDATE text, L_COMMITDATE text, L_RECEIPTDATE text, " +
+      " L_SHIPINSTRUCT text, L_SHIPMODE text, L_COMMENT text)");
+    res.close();
+
+    TableDesc desc = catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName);
+    assertNotNull(desc);
+
+    // Insert overwrite the data
+    response = client.executeQuery("insert overwrite into " + tableName + " select * from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+
+    FileStatus[] files = fs.listStatus(new Path(desc.getUri()));
+    assertEquals(1, files.length);
+    for(FileStatus file : files) {
+      assertTrue(file.getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Kill query when executing inserting data.
+    response = client.executeQuery("insert overwrite into " + tableName + " select * from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+
+      queryMasterTask = cluster.getQueryMasterTask(queryId);
+      if (queryMasterTask != null) {
+        files = fs.listStatus(new Path(desc.getUri()));
+        if (files.length > 1) {
+          queryMasterTask.getEventHandler().handle(new QueryEvent(queryId, QueryEventType.KILL));
+          break;
+        }
+      }
+    }
+    waitUntilQueryFinish(queryMasterTask.getQuery(), 50, 200, queryMasterTask);
+
+    // Make sure whether previous partitions exists or not.
+    files = fs.listStatus(new Path(desc.getUri()));
+    assertEquals(1, files.length);
+    for(FileStatus file : files) {
+      assertTrue(file.getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testRecoverAfterKillQueryWithPartitionedTableByOneColumn() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col1 int4, col2 int4) partition by column(key float8) ");
+    res.close();
+
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    // Insert overwrite the data
+    response = client.executeQuery("insert overwrite into " + tableName
+        + " (col1, col2, key) select l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    // Check the number of partitions and the prefix of each output file
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(6, partitions.size());
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Kill query when executing inserting data.
+    response = client.executeQuery("insert into " + tableName
+      + " select l_orderkey, l_partkey, l_quantity from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+
+      queryMasterTask = cluster.getQueryMasterTask(queryId);
+      if (queryMasterTask != null) {
+        Query query = queryMasterTask.getQuery();
+        if (query != null && query.getPartitions() != null) {
+          if (query.getPartitions().size() > 2) {
+            queryMasterTask.getEventHandler().handle(new QueryEvent(queryId, QueryEventType.KILL));
+            Thread.sleep(100);
+            break;
+          }
+        }
+      }
+    }
+    waitUntilQueryFinish(queryMasterTask.getQuery(), 50, 200, queryMasterTask);
+
+    // Make sure whether previous partitions exists or not.
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testRecoverAfterKillQueryWithPartitionedTableByThreeColumns() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col4 text) partition by column(col1 int4, col2 int4, col3 float8) ");
+    res.close();
+
+    assertTrue(catalog.existsTable(DEFAULT_DATABASE_NAME, tableName));
+
+    // Insert overwrite the data
+    response = client.executeQuery("insert overwrite into " + tableName
+      + " select l_returnflag, l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+      Thread.sleep(100);
+    }
+    assertNotNull(queryInfo);
+
+    // Check the number of partitions and the prefix of each output file
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(6, partitions.size());
+
+    String prefix = "UUID-" + queryId.toString().substring(2).replaceAll("_", "-");
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Kill query when executing inserting data.
+    response = client.executeQuery("insert into " + tableName
+      + " select l_returnflag, l_orderkey, l_partkey, l_quantity from lineitem");
+    queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+
+      queryMasterTask = cluster.getQueryMasterTask(queryId);
+      if (queryMasterTask != null) {
+        Query query = queryMasterTask.getQuery();
+        if (query != null && query.getPartitions() != null) {
+          if (query.getPartitions().size() > 2) {
+            queryMasterTask.getEventHandler().handle(new QueryEvent(queryId, QueryEventType.KILL));
+            Thread.sleep(100);
+            break;
+          }
+        }
+      }
+    }
+    waitUntilQueryFinish(queryMasterTask.getQuery(), 50, 200, queryMasterTask);
+
+    // Make sure whether previous partitions exists or not.
+    for (CatalogProtos.PartitionDescProto partition : partitions) {
+      Path path = new Path(partition.getPath());
+      FileStatus[] statuses = fs.listStatus(path);
+      assertNotNull(statuses);
+      assertEquals(1, statuses.length);
+      assertTrue(statuses[0].getPath().getName().toString().startsWith(prefix));
+    }
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  @Test(timeout = 10000)
+  public final void testRemoveAllFilesAfterKillQueryWithPartitionedTableByOneColumn() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col1 int4, col2 int4) partition by column(key float8) ");
+    res.close();
+
+    TableDesc desc = catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName);
+    assertNotNull(desc);
+
+    // Kill query when executing inserting data.
+    response = client.executeQuery("insert into " + tableName
+      + " select l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+
+      queryMasterTask = cluster.getQueryMasterTask(queryId);
+      if (queryMasterTask != null) {
+        Query query = queryMasterTask.getQuery();
+        if (query != null && query.getPartitions() != null) {
+          if (query.getPartitions().size() > 2) {
+            queryMasterTask.getEventHandler().handle(new QueryEvent(queryId, QueryEventType.KILL));
+            Thread.sleep(100);
+            break;
+          }
+        }
+      }
+    }
+    waitUntilQueryFinish(queryMasterTask.getQuery(), 50, 200, queryMasterTask);
+
+    // Check the number and file count of partitions
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(0, partitions.size());
+    ContentSummary summary = fs.getContentSummary(new Path(desc.getUri()));
+    assertEquals(0, summary.getFileCount());
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+        && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+
+  @Test(timeout = 10000)
+  public final void testRemoveAllFilesAfterKillQueryWithPartitionedTableByThreeColumns() throws Exception {
+    ResultSet res = null;
+    ClientProtos.SubmitQueryResponse response = null;
+    ClientProtos.QueryInfoProto queryInfo = null;
+    QueryMasterTask queryMasterTask = null;
+    CatalogService catalog = cluster.getMaster().getCatalog();
+    FileSystem fs = cluster.getDefaultFileSystem();
+
+    setSessionVariable(true);
+    assertEquals("true", client.getSessionVariable(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED.keyname()));
+
+    // Create partition table
+    String tableName = CatalogUtil.normalizeIdentifier(name.getMethodName());
+
+    res = client.executeQueryAndGetResult("create table " + tableName
+      + " (col4 text) partition by column(col1 int4, col2 int4, col3 float8) ");
+    res.close();
+
+    TableDesc desc = catalog.getTableDesc(DEFAULT_DATABASE_NAME, tableName);
+    assertNotNull(desc);
+
+    // Kill query when executing inserting data.
+    response = client.executeQuery("insert into " + tableName
+      + " select l_returnflag, l_orderkey, l_partkey, l_quantity from lineitem");
+    QueryId queryId = new QueryId(response.getQueryId());
+
+    while (true) {
+      queryInfo = client.getQueryInfo(queryId);
+      if (queryInfo != null && queryInfo.getQueryState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
+        break;
+      }
+
+      queryMasterTask = cluster.getQueryMasterTask(queryId);
+      if (queryMasterTask != null) {
+        Query query = queryMasterTask.getQuery();
+        if (query != null && query.getPartitions() != null) {
+          if (query.getPartitions().size() > 2) {
+            queryMasterTask.getEventHandler().handle(new QueryEvent(queryId, QueryEventType.KILL));
+            Thread.sleep(100);
+            break;
+          }
+        }
+      }
+    }
+    waitUntilQueryFinish(queryMasterTask.getQuery(), 50, 200, queryMasterTask);
+
+    // Check the number and file count of partitions
+    List<CatalogProtos.PartitionDescProto> partitions = catalog.getPartitionsOfTable(DEFAULT_DATABASE_NAME, tableName);
+    assertEquals(0, partitions.size());
+    ContentSummary summary = fs.getContentSummary(new Path(desc.getUri()));
+    assertEquals(0, summary.getFileCount());
+
+    // Check the status of query history from catalog
+    List<CatalogProtos.DirectOutputCommitHistoryProto> protos  = catalog.getAllDirectOutputCommitHistories();
+    boolean historyFound = false;
+    for (CatalogProtos.DirectOutputCommitHistoryProto proto : protos) {
+      if (proto.getQueryId().equals(queryId.toString())
+      && (proto.getQueryState().equals(TajoProtos.QueryState.QUERY_SUCCEEDED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_FAILED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_KILLED.name())
+        || proto.getQueryState().equals(TajoProtos.QueryState.QUERY_ERROR.name())
+      )) {
+        historyFound = true;
+      }
+    }
+    assertTrue(historyFound);
+
+    res = client.executeQueryAndGetResult("DROP TABLE " + tableName + " PURGE");
+    res.close();
+  }
+
+  private void waitUntilQueryFinish(Query query, int delay, int maxRetry, QueryMasterTask queryMasterTask)
+    throws Exception {
+    int i = 0;
+
+    try {
+      while (query == null || (query.getSynchronizedState() != TajoProtos.QueryState.QUERY_KILLED
+        && query.getSynchronizedState() != TajoProtos.QueryState.QUERY_ERROR)) {
+        Thread.sleep(delay);
+        if (++i > maxRetry) {
+          throw new IOException("Exceeded the maximum retry (" + maxRetry + ") actual state: "
+            + query != null ? String.valueOf(query.getSynchronizedState()) : String.valueOf(query));
+        }
+      }
+    } finally {
+      queryMasterTask.stop();
+    }
+  }
+
+}
diff --git a/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result b/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
index 5a6198e08a..8176945a98 100644
--- a/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
+++ b/tajo-core-tests/src/test/resources/results/TestTajoCli/testHelpSessionVars.result
@@ -43,6 +43,7 @@ Available Session Variables:
 \set JOIN_HASH_TABLE_SIZE [int value] - The initial size of hash table for in-memory hash join
 \set INDEX_ENABLED [true or false] - index scan enabled
 \set INDEX_SELECTIVITY_THRESHOLD [real value] - the selectivity threshold for index scan
+\set DIRECT_OUTPUT_COMMITTER_ENABLED [true or false] - If true, a task will write the output data directly to the final location.
 \set PARTITION_NO_RESULT_OVERWRITE_ENABLED [true or false] - If true, a partitioned table is overwritten even if a sub query leads to no result. Otherwise, the table data will be kept if there is no result
 \set ARITHABORT [true or false] - If true, a running query will be terminated when an overflow or divide-by-zero occurs.
 \set FETCH_ROWNUM [int value] - The number of rows to be fetched from Master at a time
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index 2f046a1f0c..600f623e16 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -33,6 +33,7 @@
 import org.apache.hadoop.yarn.util.Clock;
 import org.apache.hadoop.yarn.util.RackResolver;
 import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tajo.TajoProtos;
 import org.apache.tajo.catalog.CatalogServer;
 import org.apache.tajo.catalog.CatalogService;
 import org.apache.tajo.catalog.FunctionDesc;
@@ -40,6 +41,8 @@
 import org.apache.tajo.catalog.proto.CatalogProtos;
 import org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.AlterTablespaceProto.AlterTablespaceCommand;
+import org.apache.tajo.catalog.proto.CatalogProtos.DirectOutputCommitHistoryProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.UpdateDirectOutputCommitHistoryProto;
 import org.apache.tajo.catalog.store.AbstractDBStore;
 import org.apache.tajo.catalog.store.DerbyStore;
 import org.apache.tajo.conf.TajoConf;
@@ -60,6 +63,7 @@
 import org.apache.tajo.service.ServiceTracker;
 import org.apache.tajo.service.ServiceTrackerFactory;
 import org.apache.tajo.session.SessionManager;
+import org.apache.tajo.storage.Tablespace;
 import org.apache.tajo.storage.TablespaceManager;
 import org.apache.tajo.util.*;
 import org.apache.tajo.util.history.HistoryReader;
@@ -371,6 +375,9 @@ public void serviceStart() throws Exception {
     historyWriter.start();
 
     historyReader = new HistoryReader(getMasterName(), context.getConf());
+
+    // Check remaining files of direct output commit
+    checkDirectOutputCommitHistory();
   }
 
   private void writeSystemConf() throws IOException {
@@ -444,6 +451,22 @@ private void checkBaseTBSpaceAndDatabase()
     }
   }
 
+  private void checkDirectOutputCommitHistory() throws IOException, UnsupportedException
+    , UndefinedQueryIdException, InsufficientPrivilegeException{
+    List<DirectOutputCommitHistoryProto> list = catalog.getIncompleteDirectOutputCommitHistories();
+    for (DirectOutputCommitHistoryProto history : list) {
+      Path path = new Path(history.getPath());
+      Tablespace tablespace = TablespaceManager.get(path.toUri());
+      tablespace.clearDirectOutputCommit(history.getQueryId(), path);
+
+      UpdateDirectOutputCommitHistoryProto.Builder builder = UpdateDirectOutputCommitHistoryProto.newBuilder();
+      builder.setQueryId(history.getQueryId());
+      builder.setQueryState(TajoProtos.QueryState.QUERY_SUCCEEDED.name());
+
+      catalog.updateDirectOutputCommitHistoryProto(builder.build());
+    }
+  }
+
   @Override
   public void serviceStop() throws Exception {
     if (haService != null) haService.delete();
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
index d57040e3ff..269b9a59f0 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/Query.java
@@ -37,6 +37,8 @@
 import org.apache.tajo.catalog.*;
 import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.UpdateTableStatsProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.DirectOutputCommitHistoryProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.UpdateDirectOutputCommitHistoryProto;
 import org.apache.tajo.catalog.statistics.StatisticsUtil;
 import org.apache.tajo.catalog.statistics.TableStats;
 import org.apache.tajo.conf.TajoConf;
@@ -47,6 +49,7 @@
 import org.apache.tajo.engine.query.QueryContext;
 import org.apache.tajo.error.Errors.SerializedException;
 import org.apache.tajo.exception.ErrorUtil;
+import org.apache.tajo.exception.UnsupportedException;
 import org.apache.tajo.master.event.*;
 import org.apache.tajo.plan.logical.*;
 import org.apache.tajo.plan.util.PlannerUtil;
@@ -451,15 +454,49 @@ public static class StartTransition
     public void transition(Query query, QueryEvent queryEvent) {
 
       query.setStartTime();
+
+      Stage lastStage = null;
       ExecutionQueue executionQueue = query.newExecutionQueue();
       for (ExecutionBlock executionBlock : executionQueue.first()) {
         Stage stage = new Stage(query.context, query.getPlan(), executionBlock);
         stage.setPriority(query.priority--);
         query.addStage(stage);
 
+        if (stage.getBlock().getPlan().getType() == NodeType.CREATE_TABLE
+          || stage.getBlock().getPlan().getType() == NodeType.INSERT) {
+          lastStage = stage;
+        }
+
         stage.getEventHandler().handle(new StageEvent(stage.getId(), StageEventType.SQ_INIT));
         LOG.debug("Schedule unit plan: \n" + stage.getBlock().getPlan());
       }
+
+      // Add direct output commit history to catalog
+      if (lastStage != null) {
+        addDirectOutputCommitHistory(query, lastStage);
+      }
+    }
+  }
+
+  public static void addDirectOutputCommitHistory(Query query, Stage stage) {
+    try {
+      QueryContext queryContext = query.context.getQueryContext();
+
+      if (queryContext.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)) {
+        CatalogService catalog = stage.getContext().getQueryMasterContext().getWorkerContext().getCatalog();
+
+        DirectOutputCommitHistoryProto.Builder builder = DirectOutputCommitHistoryProto.newBuilder();
+        builder.setQueryId(query.getId().toString());
+        builder.setQueryState(query.getState().name());
+        builder.setStartTime(System.currentTimeMillis());
+        Path path = new Path(queryContext.get(QueryVars.OUTPUT_TABLE_URI));
+        builder.setPath(path.toString());
+
+        catalog.addDirectOutputCommitHistory(builder.build());
+      }
+    } catch (Throwable t) {
+      LOG.error(t.getMessage(), t);
+      query.eventHandler.handle(new QueryEvent(query.getId(), QueryEventType.INTERNAL_ERROR));
     }
   }
 
@@ -486,7 +523,7 @@ public QueryState transition(Query query, QueryEvent queryEvent) {
       // When a query is failed
       if (finalState != QueryState.QUERY_SUCCEEDED) {
         Stage lastStage = query.getStage(stageEvent.getExecutionBlockId());
-        handleQueryFailure(query, lastStage);
+        handleQueryFailure(query, lastStage, finalState);
       }
 
       query.eventHandler.handle(new QueryMasterQueryCompletedEvent(query.getId()));
@@ -496,7 +533,7 @@ public QueryState transition(Query query, QueryEvent queryEvent) {
     }
 
     // handle query failures
-    private void handleQueryFailure(Query query, Stage lastStage) {
+    private void handleQueryFailure(Query query, Stage lastStage, QueryState finalState) {
       QueryContext context = query.context.getQueryContext();
 
       if (lastStage != null && context.hasOutputTableUri()) {
@@ -504,12 +541,39 @@ private void handleQueryFailure(Query query, Stage lastStage) {
         try {
           LogicalRootNode rootNode = lastStage.getMasterPlan().getLogicalPlan().getRootBlock().getRoot();
           space.rollbackTable(rootNode.getChild());
+          clearDirectOutputCommit(query, finalState, lastStage);
         } catch (Throwable e) {
           LOG.warn(query.getId() + ", failed processing cleanup storage when query failed:" + e.getMessage(), e);
         }
       }
     }
 
+    private void clearDirectOutputCommit(Query query, QueryState queryState, Stage lastStage) {
+      try {
+        QueryContext queryContext = query.context.getQueryContext();
+        NodeType type = lastStage.getBlock().getPlan().getType();
+
+        if (queryContext.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)
+          && (type == NodeType.CREATE_TABLE || type == NodeType.INSERT)) {
+          Tablespace space = TablespaceManager.get(queryContext.get(QueryVars.OUTPUT_TABLE_URI, ""));
+          Path path = new Path(queryContext.get(QueryVars.OUTPUT_TABLE_URI));
+          space.clearDirectOutputCommit(query.getId().toString(), path);
+
+          CatalogService catalog = lastStage.getContext().getQueryMasterContext().getWorkerContext().getCatalog();
+
+          UpdateDirectOutputCommitHistoryProto.Builder builder = UpdateDirectOutputCommitHistoryProto.newBuilder();
+          builder.setQueryState(queryState.name());
+          builder.setQueryId(query.getId().toString());
+
+          catalog.updateDirectOutputCommitHistoryProto(builder.build());
+        }
+      } catch (UnsupportedException ue) {
+        // HBaseTableSpace and JDBCTableSpace don't support directOutputCommitter
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+    }
+
     private QueryState finalizeQuery(Query query, QueryCompletedEvent event) {
       Stage lastStage = query.getStage(event.getExecutionBlockId());
 
@@ -525,43 +589,58 @@ private QueryState finalizeQuery(Query query, QueryCompletedEvent event) {
         // In this case, we should use default tablespace.
         Tablespace space = TablespaceManager.get(queryContext.get(QueryVars.OUTPUT_TABLE_URI, ""));
 
+        List<PartitionDescProto> partitions = null;
+        if (queryContext.hasOutputTableUri() && queryContext.hasPartition()) {
+          partitions = query.getPartitions();
+        }
+
         Path finalOutputDir = space.commitTable(
-            query.context.getQueryContext(),
-            lastStage.getId(),
-            lastStage.getMasterPlan().getLogicalPlan(),
-            lastStage.getOutSchema(),
-            tableDesc);
+          query.context.getQueryContext(),
+          lastStage.getId(),
+          lastStage.getMasterPlan().getLogicalPlan(),
+          lastStage.getOutSchema(),
+          tableDesc,
+          partitions);
 
         QueryHookExecutor hookExecutor = new QueryHookExecutor(query.context.getQueryMasterContext());
         hookExecutor.execute(query.context.getQueryContext(), query, event.getExecutionBlockId(), finalOutputDir);
 
         // Add dynamic partitions to catalog for partition table.
-        if (queryContext.hasOutputTableUri() && queryContext.hasPartition()) {
-          List<PartitionDescProto> partitions = query.getPartitions();
-          if (partitions != null) {
-            // Set contents length and file count to PartitionDescProto by listing final output directories.
-            List<PartitionDescProto> finalPartitions = getPartitionsWithContentsSummary(query.systemConf,
-              finalOutputDir, partitions);
-
-            String databaseName, simpleTableName;
-            if (CatalogUtil.isFQTableName(tableDesc.getName())) {
-              String[] split = CatalogUtil.splitFQTableName(tableDesc.getName());
-              databaseName = split[0];
-              simpleTableName = split[1];
-            } else {
-              databaseName = queryContext.getCurrentDatabase();
-              simpleTableName = tableDesc.getName();
-            }
-
-            // Store partitions to CatalogStore using alter table statement.
-            catalog.addPartitions(databaseName, simpleTableName, finalPartitions, true);
-            LOG.info("Added partitions to catalog (total=" + partitions.size() + ")");
+        if (partitions != null) {
+          // Set contents length and file count to PartitionDescProto by listing final output directories.
+          List<PartitionDescProto> finalPartitions = getPartitionsWithContentsSummary(query.systemConf,
+            finalOutputDir, partitions);
+
+          String databaseName, simpleTableName;
+          if (CatalogUtil.isFQTableName(tableDesc.getName())) {
+            String[] split = CatalogUtil.splitFQTableName(tableDesc.getName());
+            databaseName = split[0];
+            simpleTableName = split[1];
           } else {
-            LOG.info("Can't find partitions for adding.");
+            databaseName = queryContext.getCurrentDatabase();
+            simpleTableName = tableDesc.getName();
           }
+
+          // Store partitions to CatalogStore using alter table statement.
+          catalog.addPartitions(databaseName, simpleTableName, finalPartitions, true);
+          LOG.info("Added partitions to catalog (total=" + partitions.size() + ")");
+
           query.clearPartitions();
         }
+
+        // Update the status of direct output commit history
+        if (queryContext.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)
+          && (lastStage.getBlock().getPlan().getType() == NodeType.CREATE_TABLE
+          || lastStage.getBlock().getPlan().getType() == NodeType.INSERT)) {
+          UpdateDirectOutputCommitHistoryProto.Builder builder = UpdateDirectOutputCommitHistoryProto.newBuilder();
+          builder.setQueryState(QueryState.QUERY_SUCCEEDED.name());
+          builder.setQueryId(query.getId().toString());
+
+          catalog.updateDirectOutputCommitHistoryProto(builder.build());
+        }
+
       } catch (Throwable e) {
+        clearDirectOutputCommit(query, QueryState.QUERY_ERROR, lastStage);
         LOG.fatal(e.getMessage(), e);
         query.failureReason = ErrorUtil.convertException(e);
         query.eventHandler.handle(new QueryDiagnosticsUpdateEvent(query.id, ExceptionUtils.getStackTrace(e)));
@@ -816,6 +895,12 @@ private synchronized boolean executeNextBlock(Query query, ExecutionBlockId bloc
         query.addStage(nextStage);
         nextStage.getEventHandler().handle(new StageEvent(nextStage.getId(), StageEventType.SQ_INIT));
 
+        // Add direct output commit history to catalog
+        if (nextStage.getBlock().getPlan().getType() == NodeType.CREATE_TABLE
+          || nextStage.getBlock().getPlan().getType() == NodeType.INSERT) {
+          addDirectOutputCommitHistory(query, nextStage);
+        }
+
         LOG.info("Scheduling Stage:" + nextStage.getId());
         if (LOG.isDebugEnabled()) {
           LOG.debug("Scheduling Stage's Priority: " + nextStage.getPriority());
@@ -864,9 +949,9 @@ public void transition(Query query, QueryEvent event) {
           query.eventHandler.handle(new QueryCompletedEvent(castEvent.getExecutionBlockId(), castEvent.getState()));
         }
         LOG.info(String.format("Complete Stage[%s], State: %s, %d/%d. ",
-            castEvent.getExecutionBlockId().toString(),
-            castEvent.getState().toString(),
-            query.completedStagesCount,
+          castEvent.getExecutionBlockId().toString(),
+          castEvent.getState().toString(),
+          query.completedStagesCount,
             query.stages.size()));
       } catch (Throwable t) {
         LOG.error(t.getMessage(), t);
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
index 920dfe5205..5c9ce0758b 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskImpl.java
@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.tajo.SessionVars;
 import org.apache.tajo.TajoProtos;
 import org.apache.tajo.TajoProtos.FetcherState;
 import org.apache.tajo.TajoProtos.TaskAttemptState;
@@ -148,7 +149,8 @@ public void initPlan() throws IOException {
       }
     } else {
       Path outFilePath = ((FileTablespace) TablespaceManager.get(queryContext.getStagingDir().toUri()))
-          .getAppenderFilePath(getId(), queryContext.getStagingDir());
+          .getAppenderFilePath(getId(), queryContext.getStagingDir(),
+            queryContext.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED));
       LOG.info("Output File Path: " + outFilePath);
       context.setOutputPath(outFilePath);
     }
diff --git a/tajo-docs/src/main/sphinx/table_management.rst b/tajo-docs/src/main/sphinx/table_management.rst
index dd7b9ad4c0..1c629aaf53 100644
--- a/tajo-docs/src/main/sphinx/table_management.rst
+++ b/tajo-docs/src/main/sphinx/table_management.rst
@@ -11,3 +11,4 @@ In Tajo, a table is a logical view of one data sources. Logically, one table con
     table_management/tablespaces
     table_management/data_formats
     table_management/compression
+    table_management/directoutputcommitter
diff --git a/tajo-docs/src/main/sphinx/table_management/directoutputcommitter.rst b/tajo-docs/src/main/sphinx/table_management/directoutputcommitter.rst
new file mode 100644
index 0000000000..e3df8ca571
--- /dev/null
+++ b/tajo-docs/src/main/sphinx/table_management/directoutputcommitter.rst
@@ -0,0 +1,25 @@
+***********
+Direct Output Committer
+***********
+
+The final output of a Tajo query is written to a location in S3 or HDFS. However, each task that is supposed to write the output data into a final location first writes it into a temporary location. It is in a task’s commit phase, when the output data is moved from the temporary location to a final location.
+
+DirectOutputCommitter is a setting, which when enabled lets a task write the output data directly to the final location. In this case, a commit phase just will backup previous output data. Setting DirectOutputCommitter avoids Eventual Consistency issues and its default value is false.
+
+===========================================
+Properties for Direct Output Committer
+===========================================
+
+You can set DirectOutputCommitter in conf/tajo-site.xml file as follows:
+
+.. code-block:: xml
+
+  <name>tajo.query.direct-output-committer.enabled</name>
+  <value>true</value>
+
+
+Or you can set DirectOutputCommitter with ``\set`` command as follows:
+
+.. code-block:: sh
+
+  \set DIRECT_OUTPUT_COMMITTER_ENABLED true
\ No newline at end of file
diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Tablespace.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Tablespace.java
index 00e6d75a12..1b9ec5acb2 100644
--- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Tablespace.java
+++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Tablespace.java
@@ -24,6 +24,7 @@
 import org.apache.tajo.OverridableConf;
 import org.apache.tajo.TaskAttemptId;
 import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.catalog.proto.CatalogProtos.FragmentProto;
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.exception.TajoException;
@@ -147,6 +148,11 @@ public abstract List<Fragment> getSplits(String inputSourceId,
    */
   public abstract void close();
 
+  /**
+   * Clear output files of direct output commit
+   */
+  public abstract void clearDirectOutputCommit(String queryId, Path path) throws IOException, UnsupportedException;
+
 
   /**
    * It is called by a Repartitioner for range shuffling when the SortRangeType of SortNode is USING_STORAGE_MANAGER.
@@ -363,7 +369,8 @@ public void rewritePlan(OverridableConf context, LogicalPlan plan) throws TajoEx
   public abstract Path commitTable(OverridableConf queryContext,
                                    ExecutionBlockId finalEbId,
                                    LogicalPlan plan, Schema schema,
-                                   TableDesc tableDesc) throws IOException;
+                                   TableDesc tableDesc,
+                                   List<PartitionDescProto> partitions) throws IOException;
 
   public abstract void rollbackTable(LogicalNode node) throws IOException, TajoException;
 
diff --git a/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseTablespace.java b/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseTablespace.java
index 132ceff0ae..48a376970a 100644
--- a/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseTablespace.java
+++ b/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseTablespace.java
@@ -39,6 +39,7 @@
 import org.apache.hadoop.mapreduce.task.JobContextImpl;
 import org.apache.tajo.*;
 import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.catalog.statistics.TableStats;
 import org.apache.tajo.common.TajoDataTypes.Type;
 import org.apache.tajo.conf.TajoConf;
@@ -911,7 +912,7 @@ public Pair<Datum, Datum> getIndexablePredicateValue(ColumnMapping columnMapping
   @Override
   public Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId,
                           LogicalPlan plan, Schema schema,
-                          TableDesc tableDesc) throws IOException {
+                          TableDesc tableDesc, List<PartitionDescProto> partitions) throws IOException {
     if (tableDesc == null) {
       throw new IOException("TableDesc is null while calling loadIncrementalHFiles: " + finalEbId);
     }
@@ -1118,4 +1119,9 @@ public void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema) throws Ta
       }
     }
   }
+
+  @Override
+  public void clearDirectOutputCommit(String queryId, Path path) throws IOException, UnsupportedException {
+    throw new UnsupportedException();
+  }
 }
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileTablespace.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileTablespace.java
index ad4574909d..7826587cac 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileTablespace.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileTablespace.java
@@ -32,6 +32,7 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.tajo.*;
 import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.catalog.statistics.TableStats;
 import org.apache.tajo.conf.TajoConf;
 import org.apache.tajo.exception.TajoInternalError;
@@ -50,6 +51,7 @@
 import java.text.NumberFormat;
 import java.util.*;
 
+import static java.lang.String.format;
 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;
 
@@ -64,6 +66,9 @@ public boolean accept(Path p) {
   private static final Log LOG = LogFactory.getLog(FileTablespace.class);
 
   static final String OUTPUT_FILE_PREFIX="part-";
+
+  static final String DIRECT_OUTPUT_FILE_PREFIX="UUID-";
+
   static final ThreadLocal<NumberFormat> OUTPUT_FILE_FORMAT_STAGE =
       new ThreadLocal<NumberFormat>() {
         @Override
@@ -260,17 +265,30 @@ public long calculateSize(Path tablePath) throws IOException {
   // FileInputFormat Area
   /////////////////////////////////////////////////////////////////////////////
   public Path getAppenderFilePath(TaskAttemptId taskAttemptId, Path workDir) {
+    return getAppenderFilePath(taskAttemptId, workDir, false);
+  }
+
+  public Path getAppenderFilePath(TaskAttemptId taskAttemptId, Path workDir, boolean isDirectOutputCommit) {
     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,
+
+    Path outFilePath = null;
+    if (isDirectOutputCommit && !workDir.toString().startsWith(this.stagingRootPath.toString())) {
+      QueryId queryId = taskAttemptId.getTaskId().getExecutionBlockId().getQueryId();
+      outFilePath = StorageUtil.concatPath(workDir, DIRECT_OUTPUT_FILE_PREFIX +
+        queryId.toString().substring(2).replaceAll("_", "-") + "-" +
+        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));
+    } else {
+      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));
+          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;
@@ -699,12 +717,16 @@ public URI getStagingUri(OverridableConf context, String queryId, TableMeta meta
       // 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));
+      if (context.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)) {
+        stagingDir = new Path(context.get(QueryVars.OUTPUT_TABLE_URI));
       } else {
-        stagingDir = fs.makeQualified(new Path(stagingRootPath, queryId));
+        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));
+        }
       }
     }
 
@@ -732,10 +754,17 @@ public URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf co
     // Create Output Directory
     ////////////////////////////////////////////
 
-    if (fs.exists(stagingDir)) {
-      throw new IOException("The staging directory '" + stagingDir + "' already exists");
+    if (!context.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)) {
+      if (fs.exists(stagingDir)) {
+        throw new IOException("The staging directory '" + stagingDir + "' already exists");
+      }
+      fs.mkdirs(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
+    } else {
+      if (!fs.exists(stagingDir)) {
+        fs.mkdirs(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
+      }
     }
-    fs.mkdirs(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
+
     FileStatus fsStatus = fs.getFileStatus(stagingDir);
     String owner = fsStatus.getOwner();
 
@@ -754,8 +783,10 @@ public URI prepareStagingSpace(TajoConf conf, String queryId, OverridableConf co
       fs.setPermission(stagingDir, new FsPermission(STAGING_DIR_PERMISSION));
     }
 
-    Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);
-    fs.mkdirs(stagingResultDir);
+    if (!context.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)) {
+      Path stagingResultDir = new Path(stagingDir, TajoConstants.RESULT_DIR_NAME);
+      fs.mkdirs(stagingResultDir);
+    }
 
     return stagingDir.toUri();
   }
@@ -766,8 +797,14 @@ 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);
+                          Schema schema, TableDesc tableDesc, List<PartitionDescProto> partitions) throws IOException {
+
+    if (!queryContext.get(QueryVars.OUTPUT_TABLE_URI, "").isEmpty()
+      && queryContext.getBool(SessionVars.DIRECT_OUTPUT_COMMITTER_ENABLED)) {
+        return directCommitOutputData(queryContext, finalEbId.getQueryId(), partitions);
+    } else {
+      return commitOutputData(queryContext, true);
+    }
   }
 
   @Override
@@ -960,6 +997,147 @@ protected Path commitOutputData(OverridableConf queryContext, boolean changeFile
     return finalOutputDir;
   }
 
+
+  private Path directCommitOutputData(OverridableConf queryContext, QueryId queryId,
+                                      List<PartitionDescProto> partitions) throws IOException {
+
+    Path finalOutputDir = new Path(queryContext.get(QueryVars.OUTPUT_TABLE_URI));
+
+    Path backupDir = new Path(finalOutputDir, TajoConstants.INSERT_OVERWIRTE_OLD_TABLE_NAME);
+    if (!fs.exists(backupDir)) {
+      fs.mkdirs(backupDir);
+    }
+
+    String queryType = queryContext.get(QueryVars.COMMAND_TYPE);
+    String prefix = DIRECT_OUTPUT_FILE_PREFIX + queryId.toString().substring(2).replaceAll("_", "-");
+
+    try {
+      if (queryType.equals(NodeType.INSERT.name()) && queryContext.getBool(QueryVars.OUTPUT_OVERWRITE, false)) {
+        if (queryContext.get(QueryVars.OUTPUT_PARTITIONS, "").isEmpty()) {
+          // Backup existing files
+          recursiveRenameFilesStartingWith(finalOutputDir, backupDir, prefix);
+        } else {
+          // Backup existing files on added partition directory.
+          if (partitions != null) {
+            for(PartitionDescProto partition : partitions) {
+              String recoverPath = partition.getPath().replaceAll(finalOutputDir.toString(), backupDir.toString());
+              recursiveRenameFilesStartingWith(new Path(partition.getPath()), new Path(recoverPath), prefix);
+            }
+          }
+        }
+      } else if (queryType.equals(NodeType.CREATE_TABLE.name())) {
+        // Backup existing files
+        recursiveRenameFilesStartingWith(finalOutputDir, backupDir, prefix);
+      }
+
+      // Delete backup files
+      fs.delete(backupDir, true);
+
+      // Delete Staging directory
+      Path stagingDir = new Path(finalOutputDir, ".staging");
+      fs.delete(stagingDir, true);
+    } catch (Exception e) {
+      clearDirectOutputCommit(queryId.getId(), finalOutputDir);
+      throw new IOException(e);
+    }
+    return finalOutputDir;
+  }
+
+  @Override
+  public void clearDirectOutputCommit(String queryId, Path path) throws IOException {
+    Path backupDir = new Path(path, TajoConstants.INSERT_OVERWIRTE_OLD_TABLE_NAME);
+    String prefix = DIRECT_OUTPUT_FILE_PREFIX + queryId.substring(2).replaceAll("_", "-");
+
+    // Delete added files
+    if (fs.exists(path)) {
+      deleteOutputFiles(path, path, prefix);
+    }
+
+    // Recover backup files to output directory
+    if (fs.exists(backupDir)) {
+      recursiveRenameFiles(backupDir, path);
+    }
+  }
+
+  private void deleteOutputFiles(Path path, Path rootPath, String prefix) throws IOException {
+    int deleteCount = 0, notDeleteCount = 0;
+    FileStatus[] statuses = fs.listStatus(path);
+    for (FileStatus status : statuses) {
+      if (status.isDirectory()) {
+        deleteOutputFiles(status.getPath(), rootPath, prefix);
+      } else if (status.isFile()) {
+        if (status.getPath().getName().startsWith(prefix)) {
+          fs.delete(status.getPath(), false);
+          deleteCount++;
+        } else {
+          notDeleteCount++;
+        }
+      }
+    }
+
+    // If all files had been deleted, delete parent directory
+    if (deleteCount > 0 && notDeleteCount == 0) {
+      fs.delete(path, false);
+    // If there is no child directories, delete parent directory
+    } else if(fs.isDirectory(path) && fs.listStatus(path).length == 0 && !path.equals(rootPath)) {
+      fs.delete(path, false);
+    }
+  }
+
+  protected void recursiveRenameFilesStartingWith(Path sourcePath, Path targetPath, String prefix) throws
+    IOException {
+    FileStatus[] statuses = fs.listStatus(sourcePath);
+
+    for(FileStatus status : statuses) {
+      if (fs.isDirectory(status.getPath())) {
+        recursiveRenameFilesStartingWith(status.getPath(), targetPath, prefix);
+      } else if(fs.isFile(status.getPath()) && !status.getPath().getName().startsWith(prefix)) {
+        renameDirectory(status.getPath(), new Path(targetPath, status.getPath().getName()));
+      }
+    }
+  }
+
+  protected void recursiveRenameFiles(Path sourcePath, Path targetPath) throws
+    IOException {
+    FileStatus[] statuses = fs.listStatus(sourcePath);
+
+    for(FileStatus status : statuses) {
+      if (fs.isDirectory(status.getPath())) {
+        recursiveRenameFiles(status.getPath(), targetPath);
+      } else {
+        renameDirectory(status.getPath(), new Path(targetPath, status.getPath().getName()));
+      }
+    }
+  }
+
+  protected void renameDirectory(Path sourcePath, Path targetPath) throws IOException {
+    try {
+      if (!fs.exists(targetPath.getParent())) {
+        createDirectory(targetPath.getParent());
+      }
+      if (!rename(sourcePath, targetPath)) {
+        throw new IOException(format("Failed to rename %s to %s: rename returned false", sourcePath, targetPath));
+      }
+    } catch (IOException e) {
+      e.printStackTrace();
+      throw new IOException(format("Failed to rename %s to %s", sourcePath, targetPath), e);
+    }
+  }
+
+  protected void createDirectory(Path path) throws IOException {
+    try {
+      if (!fs.mkdirs(path)) {
+        throw new IOException(format("mkdirs %s returned false", path));
+      }
+    } catch (IOException e) {
+      throw new IOException("Failed to create directory:" + path, e);
+    }
+  }
+
+  protected boolean rename(Path sourcePath, Path targetPath) throws IOException {
+    return fs.rename(sourcePath, targetPath);
+  }
+
   /**
    * Attach the sequence number to the output file name and than move the file into the final result path.
    *
diff --git a/tajo-storage/tajo-storage-jdbc/src/main/java/org/apache/tajo/storage/jdbc/JdbcTablespace.java b/tajo-storage/tajo-storage-jdbc/src/main/java/org/apache/tajo/storage/jdbc/JdbcTablespace.java
index fa6cf486e2..c2f6574b39 100644
--- a/tajo-storage/tajo-storage-jdbc/src/main/java/org/apache/tajo/storage/jdbc/JdbcTablespace.java
+++ b/tajo-storage/tajo-storage-jdbc/src/main/java/org/apache/tajo/storage/jdbc/JdbcTablespace.java
@@ -27,6 +27,7 @@
 import org.apache.tajo.ExecutionBlockId;
 import org.apache.tajo.OverridableConf;
 import org.apache.tajo.catalog.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto;
 import org.apache.tajo.exception.NotImplementedException;
 import org.apache.tajo.exception.TajoInternalError;
 import org.apache.tajo.exception.TajoRuntimeException;
@@ -178,7 +179,7 @@ public void prepareTable(LogicalNode node) throws IOException {
 
   @Override
   public Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId, LogicalPlan plan, Schema schema,
-                          TableDesc tableDesc) throws IOException {
+                          TableDesc tableDesc, List<PartitionDescProto> partitions) throws IOException {
     throw new TajoRuntimeException(new NotImplementedException());
   }
 
@@ -200,6 +201,11 @@ public abstract Scanner getScanner(TableMeta meta,
                             Fragment fragment,
                             @Nullable Schema target) throws IOException;
 
+  @Override
+  public void clearDirectOutputCommit(String queryId, Path path) throws IOException, UnsupportedException {
+    throw new UnsupportedException();
+  }
+
   public DatabaseMetaData getDatabaseMetaData() {
     try {
       return conn.getMetaData();