TAJO-2087: Support DirectOutputCommitter for AWS S3 file system

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

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

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

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

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

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

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 {