justinuang commented on a change in pull request #20303: [SPARK-23128][SQL] A new approach to do adaptive execution in Spark SQL URL: https://github.com/apache/spark/pull/20303#discussion_r252017917
########## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/QueryStageTrigger.scala ########## @@ -0,0 +1,160 @@ +/* + * 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.spark.sql.execution.adaptive + +import scala.collection.mutable.{HashMap, HashSet, ListBuffer} +import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, Future} + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.util.{EventLoop, ThreadUtils} + +/** + * This class triggers [[QueryStage]] bottom-up, apply planner rules for query stages and + * materialize them. It triggers as many query stages as possible at the same time, and triggers + * the parent query stage when all its child stages are materialized. + */ +class QueryStageTrigger(session: SparkSession, callback: QueryStageTriggerCallback) + extends EventLoop[QueryStageTriggerEvent]("QueryStageTrigger") { + + private val stageToParentStages = HashMap.empty[Int, ListBuffer[QueryStage]] + + private val idToUpdatedStage = HashMap.empty[Int, QueryStage] + + private val stageToNumPendingChildStages = HashMap.empty[Int, Int] + + private val submittedStages = HashSet.empty[Int] + + private val readyStages = HashSet.empty[Int] + + private val planner = new QueryStagePlanner(session.sessionState.conf) + + def trigger(stage: QueryStage): Unit = { + post(SubmitStage(stage)) + } + + private implicit def executionContext: ExecutionContextExecutorService = { + QueryStageTrigger.executionContext + } + + override protected def onReceive(event: QueryStageTriggerEvent): Unit = event match { + case SubmitStage(stage) => + // We may submit a query stage multiple times, because of stage reuse. Here we avoid + // re-submitting a query stage. + if (!submittedStages.contains(stage.id)) { + submittedStages += stage.id + val pendingChildStages = stage.plan.collect { + // The stage being submitted may have child stages that are already ready, if the child + // stage is a reused stage. + case stage: QueryStage if !readyStages.contains(stage.id) => stage + } + if (pendingChildStages.isEmpty) { + // This is a leaf stage, or all its child stages are ready, we can plan it now. + post(PlanStage(stage)) + } else { + // This stage has some pending child stages, we store the connection of this stage and + // its child stages, and submit all the child stages, so that we can plan this stage + // later when all its child stages are ready. + stageToNumPendingChildStages(stage.id) = pendingChildStages.length + pendingChildStages.foreach { child => + // a child may have multiple parents, because of query stage reuse. + val parentStages = stageToParentStages.getOrElseUpdate(child.id, new ListBuffer) + parentStages += stage + post(SubmitStage(child)) + } + } + } + + case PlanStage(stage) => + Future { Review comment: We found when applying this to our fork that we needed to preserve all local properties since we are now changing the threading model. https://github.com/palantir/spark/pull/443/commits/f477a242d00c5ddcea5c5d29eb3c6926d2bff3a0 ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
