presto源码学习2——RESTFul服务

骑猪看日落 2023-01-14 10:56 134阅读 0赞

## presto源码学习2——RESTFul服务 ##

> presto集群中的数据传输、节点通信、计算调度和监控等都是通过接口间的互相调用来实现的。其中，主要的的查询相关的接口有Statement、Query、Stage、Task这几类，都位于presto-main模块中

### Statement接口 ###

#### 接口概览 ####

Statement接口在QueuedStatementResource和ExecutingStatementResource类中声明

QueuedStatementResource中有以下几个操作接口

*  POST /v1/statement：所有查询的入口，客户端直接调用的接口
 *  GET /v1/statement/queued/\{queryId\}/\{token\}：提交queryId代表的一个查询
 *  DELETE /v1/statement/queued/\{queryId\}/\{token\}：取消等待查询队列中的某个查询

ExecutingStatementResource类中是对正在执行的查询的操作

*  GET /v1/statement/executing/\{queryId\}/\{token\}：执行queryId代表的查询，并将结果返回
 *  DELETE /v1/statement/executing/\{queryId\}/\{token\}：取消正在执行的查询

#### Statement语句的处理流程 ####

##### 1、statement会以参数的形式发送至/v1/statement接口，客户端连接信息（user、catalog、schema等）都会携带在请求的header中。 #####

@POST
        @Path("/v1/statement")
        @Produces(APPLICATION_JSON)
        public Response postStatement(
                String statement,
                @HeaderParam(X_FORWARDED_PROTO) String xForwardedProto,
                @Context HttpServletRequest servletRequest,
                @Context UriInfo uriInfo)
        {
        
            if (isNullOrEmpty(statement)) {
        
                throw badRequest(BAD_REQUEST, "SQL statement is empty");
            }
    
            SessionContext sessionContext = new HttpRequestSessionContext(servletRequest);
            Query query = new Query(statement, sessionContext, dispatchManager, queryResultsProvider, timeoutExecutor);
            queries.put(query.getQueryId(), query);
    
            return Response.ok(query.getInitialQueryResults(uriInfo, xForwardedProto)).build();
        }

在这个接口中，会生成一个Query（QueuedStatementResource的一个子类）对象，里面初封装了查询内容、会话信息、queryId以及查询相关的线程池等信息。

同时，生成的Query会放在QueuedStatementResource声明的ConcurrentMap<QueryId, Query> queries私有对象中，后面所有跟Query相关的调用都会使用对应的queryId去这个map中获取正在执行的Query

接着，接口会调用query.getInitialQueryResults(uriInfo, xForwardedProto)将生成的Query提交到/v1/statement/queued接口，等待执行

##### 2、Query提交到/v1/statement/queued接口 #####

在这个接口中，会起一个线程任务futureStateChange用来处理提交的Query：包括对Query的解析、调度等；并且会创建queryResultsFuture线程用来当futureStateChange执行完成之后对结果进行封装处理，并异步返回处理结果

这里启动的futureStateChange线程是用的ScheduledExecutorService起的周期性的线程池，如果一定时间内没有完成就会不断的调起新的，直到确认已经完成了任务调度

@GET
        @Path("/v1/statement/queued/{queryId}/{token}")
        @Produces(APPLICATION_JSON)
        public void getStatus(
                @PathParam("queryId") QueryId queryId,
                @PathParam("token") long token,
                @QueryParam("slug") String slug,
                @QueryParam("maxWait") Duration maxWait,
                @HeaderParam(X_FORWARDED_PROTO) String xForwardedProto,
                @Context UriInfo uriInfo,
                @Suspended AsyncResponse asyncResponse)
        {
        
            Query query = getQuery(queryId, slug);
    
            // wait for query to be dispatched, up to the wait timeout
            ListenableFuture<?> futureStateChange = addTimeout(
                    query.waitForDispatched(),
                    () -> null,
                    WAIT_ORDERING.min(MAX_WAIT_TIME, maxWait),
                    timeoutExecutor);
    
            // when state changes, fetch the next result
            ListenableFuture<Response> queryResultsFuture = transformAsync(
                    futureStateChange,
                    ignored -> query.toResponse(token, uriInfo, xForwardedProto, WAIT_ORDERING.min(MAX_WAIT_TIME, maxWait)),
                    responseExecutor);
            bindAsyncResponse(asyncResponse, queryResultsFuture, responseExecutor);
        }

### Query接口 ###

Query相关的接口在com.facebook.presto.server.QueryResource类中声明，主要是对执行中的Query进行的一些查询、取消等操作

当query被提交到/v1/statement/queued/\{queryId\}/\{token\}接口之后，server端就会调用waitForDispatched()方法，生成query派发到节点执行

private ListenableFuture<?> waitForDispatched() {
        
                // if query query submission has not finished, wait for it to finish
                synchronized (this) {
        
                    if (querySubmissionFuture == null) {
        
                        querySubmissionFuture = dispatchManager.createQuery(queryId, slug, sessionContext, query);
                    }
                    if (!querySubmissionFuture.isDone()) {
        
                        return querySubmissionFuture;
                    }
                }
    
                // otherwise, wait for the query to finish
                return dispatchManager.waitForDispatched(queryId);
            }

### Stage接口 ###

stage相关的接口在com.facebook.presto.server.StageResource类中声明，目前只有一个取消制定stage的接口

@Path("/v1/stage")
    @RolesAllowed(USER)
    public class StageResource
    {
        
        private final QueryManager queryManager;
    
        @Inject
        public StageResource(QueryManager queryManager)
        {
        
            this.queryManager = requireNonNull(queryManager, "queryManager is null");
        }
    
        @DELETE
        @Path("{stageId}")
        public void cancelStage(@PathParam("stageId") StageId stageId)
        {
        
            requireNonNull(stageId, "stageId is null");
            queryManager.cancelStage(stageId);
        }
    }

### Task接口 ###

task相关的接口在com.facebook.presto.server.TaskResource类中声明，包括对task的创建、查询、删除等操作接口