Presto查询提交流程

public Response postStatement(
        String statement,
        @Context HttpServletRequest servletRequest,
        @Context HttpHeaders httpHeaders,
        @Context UriInfo uriInfo)
{
    if (isNullOrEmpty(statement)) {
        throw badRequest(BAD_REQUEST, "SQL statement is empty");
    }

    String remoteAddress = servletRequest.getRemoteAddr();
    Optional<Identity> identity = Optional.ofNullable((Identity) servletRequest.getAttribute(AUTHENTICATED_IDENTITY));
    MultivaluedMap<String, String> headers = httpHeaders.getRequestHeaders();
		// 创建请求的session context
    SessionContext sessionContext = new HttpRequestSessionContext(headers, remoteAddress, identity, groupProvider);
  	// 创建Query对象，构造QueryId
    Query query = new Query(statement, sessionContext, dispatchManager);
    queries.put(query.getQueryId(), query);

    // let authentication filter know that identity lifecycle has been handed off
    servletRequest.setAttribute(AUTHENTICATED_IDENTITY, null);

    return createQueryResultsResponse(query.getQueryResults(query.getLastToken(), uriInfo), compressionEnabled);
}

// io.prestosql.execution.QueryIdGenerator#createNextQueryId
public synchronized QueryId createNextQueryId()
{
    // only generate 100,000 ids per day
    if (counter > 99_999) {
        // wait for the second to rollover
        while (MILLISECONDS.toSeconds(nowInMillis()) == lastTimeInSeconds) {
            Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
        }
        counter = 0;
    }

    // if it has been a second since the last id was generated, generate a new timestamp
    long now = nowInMillis();
    if (MILLISECONDS.toSeconds(now) != lastTimeInSeconds) {
        // generate new timestamp
        lastTimeInSeconds = MILLISECONDS.toSeconds(now);
        lastTimestamp = TIMESTAMP_FORMAT.print(now);

        // if the day has rolled over, restart the counter
        if (MILLISECONDS.toDays(now) != lastTimeInDays) {
            lastTimeInDays = MILLISECONDS.toDays(now);
            counter = 0;
        }
    }

    return new QueryId(format("%s_%05d_%s", lastTimestamp, counter++, coordinatorId));
}

@ResourceSecurity(PUBLIC)
@GET
@Path("queued/{queryId}/{slug}/{token}")
@Produces(APPLICATION_JSON)
public void getStatus(
        @PathParam("queryId") QueryId queryId,
        @PathParam("slug") String slug,
        @PathParam("token") long token,
        @QueryParam("maxWait") Duration maxWait,
        @Context UriInfo uriInfo,
        @Suspended AsyncResponse asyncResponse)
{
    Query query = getQuery(queryId, slug, token);

    // 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<QueryResults> queryResultsFuture = Futures.transform(
            futureStateChange,
            ignored -> query.getQueryResults(token, uriInfo),
            responseExecutor);

    // transform to Response
    ListenableFuture<Response> response = Futures.transform(
            queryResultsFuture,
            queryResults -> createQueryResultsResponse(queryResults, compressionEnabled),
            directExecutor());
    bindAsyncResponse(asyncResponse, response, responseExecutor);
}

private <C> void createQueryInternal(QueryId queryId, Slug slug, SessionContext sessionContext, String query, ResourceGroupManager<C> resourceGroupManager)
{
    Session session = null;
    PreparedQuery preparedQuery = null;
    try {
      	// 检查查询长度是否超过限制，超过报错
        if (query.length() > maxQueryLength) {
            int queryLength = query.length();
            query = query.substring(0, maxQueryLength);
            throw new PrestoException(QUERY_TEXT_TOO_LARGE, format("Query text length (%s) exceeds the maximum length (%s)", queryLength, maxQueryLength));
        }

        // 创建查询会话
        session = sessionSupplier.createSession(queryId, sessionContext);

        // 检查查询执行权限
        accessControl.checkCanExecuteQuery(sessionContext.getIdentity());

        // sql解析，生成抽象语法树
        preparedQuery = queryPreparer.prepareQuery(session, query);

        // 选择Query并执行对应的ResourceGroup
        Optional<String> queryType = getQueryType(preparedQuery.getStatement().getClass()).map(Enum::name);
        SelectionContext<C> selectionContext = resourceGroupManager.selectGroup(new SelectionCriteria(
                sessionContext.getIdentity().getPrincipal().isPresent(),
                sessionContext.getIdentity().getUser(),
                sessionContext.getIdentity().getGroups(),
                Optional.ofNullable(sessionContext.getSource()),
                sessionContext.getClientTags(),
                sessionContext.getResourceEstimates(),
                queryType));

        // apply system default session properties (does not override user set properties)
        session = sessionPropertyDefaults.newSessionWithDefaultProperties(session, queryType, selectionContext.getResourceGroupId());

        // mark existing transaction as active
        transactionManager.activateTransaction(session, isTransactionControlStatement(preparedQuery.getStatement()), accessControl);

      	// 生成DispatchQuery
        DispatchQuery dispatchQuery = dispatchQueryFactory.createDispatchQuery(
                session,
                query,
                preparedQuery,
                slug,
                selectionContext.getResourceGroupId());

      	// 向ResourceGroupManager提交查询请求并执行
        boolean queryAdded = queryCreated(dispatchQuery);
        if (queryAdded && !dispatchQuery.isDone()) {
            try {
                resourceGroupManager.submit(dispatchQuery, selectionContext, dispatchExecutor);
            }
            catch (Throwable e) {
                // dispatch query has already been registered, so just fail it directly
                dispatchQuery.fail(e);
            }
        }
    }
    catch (Throwable throwable) {
        // creation must never fail, so register a failed query in this case
        if (session == null) {
            session = Session.builder(new SessionPropertyManager())
                    .setQueryId(queryId)
                    .setIdentity(sessionContext.getIdentity())
                    .setSource(sessionContext.getSource())
                    .build();
        }
        Optional<String> preparedSql = Optional.ofNullable(preparedQuery).flatMap(PreparedQuery::getPrepareSql);
        DispatchQuery failedDispatchQuery = failedDispatchQueryFactory.createFailedDispatchQuery(session, query, preparedSql, Optional.empty(), throwable);
        queryCreated(failedDispatchQuery);
    }
}

public class FunctionCall
        extends Expression
{
  	// 函数名
    private final QualifiedName name;
    private final Optional<Window> window;
    private final Optional<Expression> filter;
    private final Optional<OrderBy> orderBy;
    private final boolean distinct;
    private final Optional<NullTreatment> nullTreatment;
  	// 函数参数
    private final List<Expression> arguments;

public interface Operator
        extends AutoCloseable
{
    ListenableFuture<?> NOT_BLOCKED = Futures.immediateFuture(null);

    OperatorContext getOperatorContext();

    /**
     * Returns a future that will be completed when the operator becomes
     * unblocked.  If the operator is not blocked, this method should return
     * {@code NOT_BLOCKED}.
     */
    default ListenableFuture<?> isBlocked()
    {
        return NOT_BLOCKED;
    }

    /**
     * Returns true if and only if this operator can accept an input page.
     */
    boolean needsInput();

    /**
     * Adds an input page to the operator.  This method will only be called if
     * {@code needsInput()} returns true.
     */
    void addInput(Page page);

    /**
     * Gets an output page from the operator.  If no output data is currently
     * available, return null.
     */
    Page getOutput();

    /**
     * After calling this method operator should revoke all reserved revocable memory.
     * As soon as memory is revoked returned future should be marked as done.
     * <p>
     * Spawned threads cannot modify OperatorContext because it's not thread safe.
     * For this purpose implement {@link #finishMemoryRevoke()}
     * <p>
     * Since memory revoking signal is delivered asynchronously to the Operator, implementation
     * must gracefully handle the case when there no longer is any revocable memory allocated.
     * <p>
     * After this method is called on Operator the Driver is disallowed to call any
     * processing methods on it (isBlocked/needsInput/addInput/getOutput) until
     * {@link #finishMemoryRevoke()} is called.
     */
    default ListenableFuture<?> startMemoryRevoke()
    {
        return NOT_BLOCKED;
    }

    /**
     * Clean up and release resources after completed memory revoking. Called by driver
     * once future returned by startMemoryRevoke is completed.
     */
    default void finishMemoryRevoke()
    {
    }

    /**
     * Notifies the operator that no more pages will be added and the
     * operator should finish processing and flush results. This method
     * will not be called if the Task is already failed or canceled.
     */
    void finish();

    /**
     * Is this operator completely finished processing and no more
     * output pages will be produced.
     */
    boolean isFinished();

    /**
     * This method will always be called before releasing the Operator reference.
     */
    @Override
    default void close()
            throws Exception
    {
    }
}

private ListenableFuture<?> processInternal(OperationTimer operationTimer)
for (int i = 0; i < activeOperators.size() - 1 && !driverContext.isDone(); i++) {
    Operator current = activeOperators.get(i);
    Operator next = activeOperators.get(i + 1);

    // skip blocked operator
    if (getBlockedFuture(current).isPresent()) {
        continue;
    }

    // 如果当前算子未完成任务，且下一个算子未被阻塞且需要输入
    if (!current.isFinished() && getBlockedFuture(next).isEmpty() && next.needsInput()) {
        // 从当前算子中获取计算输出的Page
        Page page = current.getOutput();
        current.getOperatorContext().recordGetOutput(operationTimer, page);

        // 如果能拿到一个Page，将其给到下一个算子
        if (page != null && page.getPositionCount() != 0) {
            next.addInput(page);
            next.getOperatorContext().recordAddInput(operationTimer, page);
            movedPage = true;
        }

        if (current instanceof SourceOperator) {
            movedPage = true;
        }
    }

    // if current operator is finished...
    if (current.isFinished()) {
        // let next operator know there will be no more data
        next.finish();
        next.getOperatorContext().recordFinish(operationTimer);
    }
}  

public abstract class Node
{
  protected <R, C> R accept(AstVisitor<R, C> visitor, C context)
    {
        return visitor.visitNode(this, context);
    }
}

public abstract class Statement
        extends Node
{
    @Override
    public <R, C> R accept(AstVisitor<R, C> visitor, C context)
    {
        return visitor.visitStatement(this, context);
    }
}

private Node invokeParser(String name, String sql, Function<SqlBaseParser, ParserRuleContext> parseFunction, ParsingOptions parsingOptions)
{
    try {
      	// 从sql字符串构建字节流，忽略大小写
        SqlBaseLexer lexer = new SqlBaseLexer(new CaseInsensitiveStream(CharStreams.fromString(sql)));
        CommonTokenStream tokenStream = new CommonTokenStream(lexer);
        SqlBaseParser parser = new SqlBaseParser(tokenStream);
        initializer.accept(lexer, parser);

        // Override the default error strategy to not attempt inserting or deleting a token.
        // Otherwise, it messes up error reporting
        parser.setErrorHandler(new DefaultErrorStrategy()
        {
            @Override
            public Token recoverInline(Parser recognizer)
                    throws RecognitionException
            {
                if (nextTokensContext == null) {
                    throw new InputMismatchException(recognizer);
                }
                else {
                    throw new InputMismatchException(recognizer, nextTokensState, nextTokensContext);
                }
            }
        });
				// PostProcessor用来显示捕捉某种语法错误并给出精确的错误提示
      	// 使用观察者模式，在语法分析的过程中触发回调
        parser.addParseListener(new PostProcessor(Arrays.asList(parser.getRuleNames()), parser));

        lexer.removeErrorListeners();
        lexer.addErrorListener(LEXER_ERROR_LISTENER);

        parser.removeErrorListeners();
        parser.addErrorListener(PARSER_ERROR_HANDLER);

        ParserRuleContext tree;
        try {
            // first, try parsing with potentially faster SLL mode
            parser.getInterpreter().setPredictionMode(PredictionMode.SLL);
  					// 生成语法分析树对象
            tree = parseFunction.apply(parser);
        }
        catch (ParseCancellationException ex) {
            // if we fail, parse with LL mode
            tokenStream.seek(0); // rewind input stream
            parser.reset();

            parser.getInterpreter().setPredictionMode(PredictionMode.LL);
            tree = parseFunction.apply(parser);
        }
				// 使用访问者模式生成以Node为父类的抽象语法树结构
        return new AstBuilder(parsingOptions).visit(tree);
    }
    catch (StackOverflowError e) {
        throw new ParsingException(name + " is too large (stack overflow while parsing)");
    }
}

  @Override
  protected Type visitDereferenceExpression(DereferenceExpression node, StackableAstVisitorContext<Context> context)
  {
    	// 尝试转换成QualifiedName，这种情况下要求每个部分都是表示符
      QualifiedName qualifiedName = DereferenceExpression.getQualifiedName(node);

      // 如果这个Dereference结构看起来像是带有库表限定的列名，先尝试匹配到列
      if (qualifiedName != null) {
          Scope scope = context.getContext().getScope();
        	// 尝试识别该字段
          Optional<ResolvedField> resolvedField = scope.tryResolveField(node, qualifiedName);
          if (resolvedField.isPresent()) {
              return handleResolvedField(node, resolvedField.get(), context);
          }
        	// 是一个非法的结构
          if (!scope.isColumnReference(qualifiedName)) {
              throw missingAttributeException(node, qualifiedName);
          }
      }
			
    	// 前缀存在列引用，进入递归解析，递归分析base结构
      Type baseType = process(node.getBase(), context);
      if (!(baseType instanceof RowType)) {
          throw semanticException(TYPE_MISMATCH, node.getBase(), "Expression %s is not of type ROW", node.getBase());
      }
			
    	// baseType一定是RowType
      RowType rowType = (RowType) baseType;
    	// 根据fieldName确定子列名称
      String fieldName = node.getField().getValue();

      Type rowFieldType = null;
    	// 验证fieldNmae是否真的存在
      for (RowType.Field rowField : rowType.getFields()) {
          if (fieldName.equalsIgnoreCase(rowField.getName().orElse(null))) {
              rowFieldType = rowField.getType();
              break;
          }
      }

      if (rowFieldType == null) {
          throw missingAttributeException(node, qualifiedName);
      }

      return setExpressionType(node, rowFieldType);
  }

public class Symbol
        implements Comparable<Symbol>
{
    private final String name;

    public static Symbol from(Expression expression)
    {
      	// 必须是SymbolReference类型的表达式
        checkArgument(expression instanceof SymbolReference, "Unexpected expression: %s", expression);
        return new Symbol(((SymbolReference) expression).getName());
    }

    @JsonCreator
    public Symbol(String name)
    {
        requireNonNull(name, "name is null");
        this.name = name;
    }

    @JsonValue
  	// 生成Symbol，name是唯一的
    public String getName()
    {
        return name;
    }
		// Symbol转SymbolReference
    public SymbolReference toSymbolReference()
    {
        return new SymbolReference(name);
    }

// Symbol的wrapper，复用Expression体系
public class SymbolReference
        extends Expression
{
    private final String name;

    public SymbolReference(String name)
    {
        super(Optional.empty());
        this.name = name;
    }  

public abstract class PlanNode
{
    private final PlanNodeId id;

    protected PlanNode(PlanNodeId id)
    {
        requireNonNull(id, "id is null");
        this.id = id;
    }

    @JsonProperty("id")
    public PlanNodeId getId()
    {
        return id;
    }

    public abstract List<PlanNode> getSources();

    public abstract List<Symbol> getOutputSymbols();

    public abstract PlanNode replaceChildren(List<PlanNode> newChildren);

    public <R, C> R accept(PlanVisitor<R, C> visitor, C context)
    {
        return visitor.visitPlan(this, context);
    }
}

public class PushPartialAggregationThroughExchange
        implements Rule<AggregationNode>
{
    private final Metadata metadata;

    public PushPartialAggregationThroughExchange(Metadata metadata)
    {
        this.metadata = requireNonNull(metadata, "metadata is null");
    }

    private static final Capture<ExchangeNode> EXCHANGE_NODE = Capture.newCapture();

  	// pattern，目标节点为AggregationNode，且子节点为ExchangeNode，且数据交换节点不能有排序要求
  	// 因为在排序情况下聚合函数无法进行预聚合优化，只能在单个任务中完成聚合，captureAs捕获命中的exchange节点
    private static final Pattern<AggregationNode> PATTERN = aggregation()
            .with(source().matching(
                    exchange()
                            .matching(node -> node.getOrderingScheme().isEmpty())
                            .capturedAs(EXCHANGE_NODE)));

    @Override
    public Pattern<AggregationNode> getPattern()
    {
        return PATTERN;
    }

    @Override
    public Result apply(AggregationNode aggregationNode, Captures captures, Context context)
    {
        ExchangeNode exchangeNode = captures.get(EXCHANGE_NODE);

        boolean decomposable = aggregationNode.isDecomposable(metadata);

public class IterativeOptimizer
        implements PlanOptimizer
{
    private final RuleStatsRecorder stats;
    private final StatsCalculator statsCalculator;
    private final CostCalculator costCalculator;
    private final List<PlanOptimizer> legacyRules;
    private final RuleIndex ruleIndex;
    private final Predicate<Session> useLegacyRules;

    public IterativeOptimizer(RuleStatsRecorder stats, StatsCalculator statsCalculator, CostCalculator costCalculator, Set<Rule<?>> rules)
    {
        this(stats, statsCalculator, costCalculator, session -> false, ImmutableList.of(), rules);
    }

    public IterativeOptimizer(RuleStatsRecorder stats, StatsCalculator statsCalculator, CostCalculator costCalculator, Predicate<Session> useLegacyRules, List<PlanOptimizer> legacyRules, Set<Rule<?>> newRules)
    {
        this.stats = requireNonNull(stats, "stats is null");
        this.statsCalculator = requireNonNull(statsCalculator, "statsCalculator is null");
        this.costCalculator = requireNonNull(costCalculator, "costCalculator is null");
        this.useLegacyRules = requireNonNull(useLegacyRules, "useLegacyRules is null");
        this.legacyRules = ImmutableList.copyOf(legacyRules);
      	// 初始化，构造ruleIndex
        this.ruleIndex = RuleIndex.builder()
                .register(newRules)
                .build();

        stats.registerAll(newRules);
    }

  	// 优化逻辑，重写PlanOptimizer中的方法
    @Override
    public PlanNode optimize(PlanNode plan, Session session, TypeProvider types, SymbolAllocator symbolAllocator, PlanNodeIdAllocator idAllocator, WarningCollector warningCollector)
    {
        // only disable new rules if we have legacy rules to fall back to
        if (useLegacyRules.test(session) && !legacyRules.isEmpty()) {
            for (PlanOptimizer optimizer : legacyRules) {
                plan = optimizer.optimize(plan, session, symbolAllocator.getTypes(), symbolAllocator, idAllocator, warningCollector);
            }

            return plan;
        }
				// 转换成Memo
        Memo memo = new Memo(idAllocator, plan);
        Lookup lookup = Lookup.from(planNode -> Stream.of(memo.resolve(planNode)));
				// 超时控制
        Duration timeout = SystemSessionProperties.getOptimizerTimeout(session);
        Context context = new Context(memo, lookup, idAllocator, symbolAllocator, System.nanoTime(), timeout.toMillis(), session, warningCollector);
        exploreGroup(memo.getRootGroup(), context);
				// 从Memo结构转换成执行计划
        return memo.extract();
    }

    private boolean exploreGroup(int group, Context context)
    {
				// progress变量跟踪当前Group或者子节点是否由于应用了优化规则而发生变换
      	// 对当前节点进行优化
        boolean progress = exploreNode(group, context);
				// 递归遍历子节点进行优化
        while (exploreChildren(group, context)) {
            progress = true;
						如果子节点发生变更，再次尝试对当前节点进行优化，也许能应用新的规则
            if (!exploreNode(group, context)) {
                // 当前节点无更新，退出
                break;
            }
        }

        return progress;
    }

    private boolean exploreNode(int group, Context context)
    {
      	// 根据GroupID获取真正的执行计划节点
        PlanNode node = context.memo.getNode(group);

        boolean done = false;
        boolean progress = false;

        while (!done) {
            context.checkTimeoutNotExhausted();

            done = true;
          	// getCandicates可以快速过滤无关的优化规则
            Iterator<Rule<?>> possiblyMatchingRules = ruleIndex.getCandidates(node).iterator();
            while (possiblyMatchingRules.hasNext()) {
                Rule<?> rule = possiblyMatchingRules.next();

                if (!rule.isEnabled(context.session)) {
                    continue;
                }
								// 对当前节点应用一个优化规则
                Rule.Result result = transform(node, rule, context);
								// 成功应用了规则
                if (result.getTransformedPlan().isPresent()) {
                  	// 用优化后的结果替换当前节点
                    node = context.memo.replace(group, result.getTransformedPlan().get(), rule.getClass().getName());

                    done = false;
                    progress = true;
                }
            }
        }

        return progress;
    }

    private <T> Rule.Result transform(PlanNode node, Rule<T> rule, Context context)
    {
        Capture<T> nodeCapture = newCapture();
        Pattern<T> pattern = rule.getPattern().capturedAs(nodeCapture);
      	// 调用match函数进行模式匹配
        Iterator<Match> matches = pattern.match(node, context.lookup).iterator();
        while (matches.hasNext()) {
            Match match = matches.next();
            long duration;
            Rule.Result result;
            try {
                long start = System.nanoTime();
              	// 应用单条优化规则
                result = rule.apply(match.capture(nodeCapture), match.captures(), ruleContext(context));
                duration = System.nanoTime() - start;
            }
            catch (RuntimeException e) {
                stats.recordFailure(rule);
                throw e;
            }
            stats.record(rule, duration, !result.isEmpty());

            if (result.getTransformedPlan().isPresent()) {
                return result;
            }
        }

        return Rule.Result.empty();
    }

    private boolean exploreChildren(int group, Context context)
    {
        boolean progress = false;

        PlanNode expression = context.memo.getNode(group);
        for (PlanNode child : expression.getSources()) {
            checkState(child instanceof GroupReference, "Expected child to be a group reference. Found: " + child.getClass().getName());

            if (exploreGroup(((GroupReference) child).getGroupId(), context)) {
              	// 记录执行计划是否发生变更
                progress = true;
            }
        }

        return progress;
    }