From 7798a81f88c8c0f0be039ee8533467b5b1cb61ba Mon Sep 17 00:00:00 2001 From: zml1206 Date: Wed, 19 Jun 2024 13:35:40 +0800 Subject: [PATCH 1/3] [CORE] Add custom cost evaluator for optimize buildSide of shuffled hash join --- .../org/apache/gluten/GlutenPlugin.scala | 6 +++ .../org/apache/gluten/GlutenConfig.scala | 15 +++++++ .../sql/execution/adaptive/GlutenCost.scala | 43 +++++++++++++++++++ .../adaptive/GlutenCostEvaluator.scala | 26 +++++++++++ .../adaptive/GlutenCostEvaluator.scala | 29 +++++++++++++ .../adaptive/GlutenCostEvaluator.scala | 29 +++++++++++++ .../adaptive/GlutenCostEvaluator.scala | 29 +++++++++++++ 7 files changed, 177 insertions(+) create mode 100644 shims/common/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCost.scala create mode 100644 shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala create mode 100644 shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala create mode 100644 shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala create mode 100644 shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala diff --git a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala index cafed66eb8f0..2860e3ced072 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala @@ -143,6 +143,12 @@ private[gluten] class GlutenDriverPlugin extends DriverPlugin with Logging { } conf.set(SPARK_SESSION_EXTS_KEY, extensions) + // adaptive custom cost evaluator class + if (GlutenConfig.getConf.enableGluten && GlutenConfig.getConf.enableGlutenCostEvaluator) { + val costEvaluator = "org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator" + conf.set(SQLConf.ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS.key, costEvaluator) + } + // check memory off-heap enabled and size val minOffHeapSize = "1MB" if ( diff --git a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala index eb6118ffc74e..0aa3441391a6 100644 --- a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala +++ b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala @@ -437,6 +437,8 @@ class GlutenConfig(conf: SQLConf) extends Logging { def enableCastAvgAggregateFunction: Boolean = conf.getConf(COLUMNAR_NATIVE_CAST_AGGREGATE_ENABLED) + def enableGlutenCostEvaluator: Boolean = conf.getConf(COST_EVALUATOR_ENABLED) + def dynamicOffHeapSizingEnabled: Boolean = conf.getConf(DYNAMIC_OFFHEAP_SIZING_ENABLED) @@ -595,6 +597,8 @@ object GlutenConfig { val GLUTEN_DYNAMIC_OFFHEAP_SIZING_MEMORY_FRACTION = "spark.gluten.memory.dynamic.offHeap.sizing.memory.fraction" + val GLUTEN_COST_EVALUATOR_ENABLED = "spark.gluten.sql.execution.adaptive.costEvaluator.enabled" + var ins: GlutenConfig = _ def getConf: GlutenConfig = { @@ -1962,6 +1966,17 @@ object GlutenConfig { .booleanConf .createWithDefault(true) + val COST_EVALUATOR_ENABLED = + buildConf(GlutenConfig.GLUTEN_COST_EVALUATOR_ENABLED) + .internal() + .doc( + "If true and gluten enabled, use " + + "org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost " + + "evaluator class, else follow the configuration " + + "org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator.") + .booleanConf + .createWithDefault(true) + val DYNAMIC_OFFHEAP_SIZING_ENABLED = buildConf(GlutenConfig.GLUTEN_DYNAMIC_OFFHEAP_SIZING_ENABLED) .internal() diff --git a/shims/common/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCost.scala b/shims/common/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCost.scala new file mode 100644 index 000000000000..9436f469ba79 --- /dev/null +++ b/shims/common/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCost.scala @@ -0,0 +1,43 @@ +/* + * 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 org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.execution.SparkPlan + +class GlutenCost(val eval: CostEvaluator, val plan: SparkPlan) extends Cost { + override def compare(that: Cost): Int = that match { + case that: GlutenCost if plan eq that.plan => + 0 + case that: GlutenCost if plan == that.plan => + // Plans are identical. Considers the newer one as having lower cost. + -(plan.id - that.plan.id) + case that: GlutenCost => + // Plans are different. Use the delegated cost evaluator. + assert(eval == that.eval) + eval.evaluateCost(plan).compare(eval.evaluateCost(that.plan)) + case _ => + throw QueryExecutionErrors.cannotCompareCostWithTargetCostError(that.toString) + } + + override def hashCode(): Int = throw new UnsupportedOperationException() + + override def equals(obj: Any): Boolean = obj match { + case that: Cost => compare(that) == 0 + case _ => false + } +} diff --git a/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala new file mode 100644 index 000000000000..2016352ad86b --- /dev/null +++ b/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -0,0 +1,26 @@ +/* + * 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 org.apache.spark.sql.execution.SparkPlan + +/** This [[CostEvaluator]] is to force use the new physical plan when cost is equal. */ +case class GlutenCostEvaluator() extends CostEvaluator { + override def evaluateCost(plan: SparkPlan): Cost = { + new GlutenCost(SimpleCostEvaluator, plan) + } +} diff --git a/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala new file mode 100644 index 000000000000..6b52dd772940 --- /dev/null +++ b/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -0,0 +1,29 @@ +/* + * 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 org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.internal.SQLConf + +/** This [[CostEvaluator]] is to force use the new physical plan when cost is equal. */ +case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { + override def evaluateCost(plan: SparkPlan): Cost = { + val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } +} diff --git a/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala new file mode 100644 index 000000000000..6b52dd772940 --- /dev/null +++ b/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -0,0 +1,29 @@ +/* + * 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 org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.internal.SQLConf + +/** This [[CostEvaluator]] is to force use the new physical plan when cost is equal. */ +case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { + override def evaluateCost(plan: SparkPlan): Cost = { + val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } +} diff --git a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala new file mode 100644 index 000000000000..6b52dd772940 --- /dev/null +++ b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -0,0 +1,29 @@ +/* + * 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 org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.internal.SQLConf + +/** This [[CostEvaluator]] is to force use the new physical plan when cost is equal. */ +case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { + override def evaluateCost(plan: SparkPlan): Cost = { + val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } +} From 51b4911b7fff51c5b76368aa44eabab38679dc7d Mon Sep 17 00:00:00 2001 From: zml1206 Date: Wed, 19 Jun 2024 16:42:29 +0800 Subject: [PATCH 2/3] fix doc --- .../common/src/main/scala/org/apache/gluten/GlutenConfig.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala index 0aa3441391a6..442b319d57ae 100644 --- a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala +++ b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala @@ -1973,7 +1973,7 @@ object GlutenConfig { "If true and gluten enabled, use " + "org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator as custom cost " + "evaluator class, else follow the configuration " + - "org.apache.spark.sql.execution.adaptive.GlutenCostEvaluator.") + "spark.sql.adaptive.customCostEvaluatorClass.") .booleanConf .createWithDefault(true) From f75f34663b142f2f555898eef0f898bf56eab0a2 Mon Sep 17 00:00:00 2001 From: zml1206 Date: Wed, 19 Jun 2024 16:58:30 +0800 Subject: [PATCH 3/3] make as a query level config --- .../src/main/scala/org/apache/gluten/GlutenConfig.scala | 2 +- .../sql/execution/adaptive/GlutenCostEvaluator.scala | 8 +++++++- .../sql/execution/adaptive/GlutenCostEvaluator.scala | 8 +++++++- .../sql/execution/adaptive/GlutenCostEvaluator.scala | 8 +++++++- .../sql/execution/adaptive/GlutenCostEvaluator.scala | 8 +++++++- 5 files changed, 29 insertions(+), 5 deletions(-) diff --git a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala index 442b319d57ae..462032488548 100644 --- a/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala +++ b/shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala @@ -597,7 +597,7 @@ object GlutenConfig { val GLUTEN_DYNAMIC_OFFHEAP_SIZING_MEMORY_FRACTION = "spark.gluten.memory.dynamic.offHeap.sizing.memory.fraction" - val GLUTEN_COST_EVALUATOR_ENABLED = "spark.gluten.sql.execution.adaptive.costEvaluator.enabled" + val GLUTEN_COST_EVALUATOR_ENABLED = "spark.gluten.sql.adaptive.costEvaluator.enabled" var ins: GlutenConfig = _ diff --git a/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala index 2016352ad86b..a6f066462a1a 100644 --- a/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala +++ b/shims/spark32/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -16,11 +16,17 @@ */ package org.apache.spark.sql.execution.adaptive +import org.apache.gluten.GlutenConfig + import org.apache.spark.sql.execution.SparkPlan /** This [[CostEvaluator]] is to force use the new physical plan when cost is equal. */ case class GlutenCostEvaluator() extends CostEvaluator { override def evaluateCost(plan: SparkPlan): Cost = { - new GlutenCost(SimpleCostEvaluator, plan) + if (GlutenConfig.getConf.enableGluten) { + new GlutenCost(SimpleCostEvaluator, plan) + } else { + SimpleCostEvaluator.evaluateCost(plan) + } } } diff --git a/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala index 6b52dd772940..8fcfa735f463 100644 --- a/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala +++ b/shims/spark33/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.sql.execution.adaptive +import org.apache.gluten.GlutenConfig + import org.apache.spark.sql.catalyst.SQLConfHelper import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.internal.SQLConf @@ -24,6 +26,10 @@ import org.apache.spark.sql.internal.SQLConf case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { override def evaluateCost(plan: SparkPlan): Cost = { val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) - new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + if (GlutenConfig.getConf.enableGluten) { + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } else { + SimpleCostEvaluator(forceOptimizeSkewedJoin).evaluateCost(plan) + } } } diff --git a/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala index 6b52dd772940..8fcfa735f463 100644 --- a/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala +++ b/shims/spark34/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.sql.execution.adaptive +import org.apache.gluten.GlutenConfig + import org.apache.spark.sql.catalyst.SQLConfHelper import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.internal.SQLConf @@ -24,6 +26,10 @@ import org.apache.spark.sql.internal.SQLConf case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { override def evaluateCost(plan: SparkPlan): Cost = { val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) - new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + if (GlutenConfig.getConf.enableGluten) { + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } else { + SimpleCostEvaluator(forceOptimizeSkewedJoin).evaluateCost(plan) + } } } diff --git a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala index 6b52dd772940..8fcfa735f463 100644 --- a/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala +++ b/shims/spark35/src/main/scala/org/apache/spark/sql/execution/adaptive/GlutenCostEvaluator.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.sql.execution.adaptive +import org.apache.gluten.GlutenConfig + import org.apache.spark.sql.catalyst.SQLConfHelper import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.internal.SQLConf @@ -24,6 +26,10 @@ import org.apache.spark.sql.internal.SQLConf case class GlutenCostEvaluator() extends CostEvaluator with SQLConfHelper { override def evaluateCost(plan: SparkPlan): Cost = { val forceOptimizeSkewedJoin = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) - new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + if (GlutenConfig.getConf.enableGluten) { + new GlutenCost(SimpleCostEvaluator(forceOptimizeSkewedJoin), plan) + } else { + SimpleCostEvaluator(forceOptimizeSkewedJoin).evaluateCost(plan) + } } }