You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2021/08/25 05:55:26 UTC
[spark] branch branch-3.1 updated: [SPARK-36352][SQL][3.1] Spark
should check result plan's output schema name
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new ce66e41 [SPARK-36352][SQL][3.1] Spark should check result plan's output schema name
ce66e41 is described below
commit ce66e41bd74dcfd8da757afd9d041a50908ee0fc
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Wed Aug 25 13:54:39 2021 +0800
[SPARK-36352][SQL][3.1] Spark should check result plan's output schema name
### What changes were proposed in this pull request?
Spark should check result plan's output schema name
### Why are the changes needed?
In current code, some optimizer rule may change plan's output schema, since in the code we always use semantic equal to check output, but it may change the plan's output schema.
For example, for SchemaPruning, if we have a plan
```
Project[a, B]
|--Scan[A, b, c]
```
the origin output schema is `a, B`, after SchemaPruning. it become
```
Project[A, b]
|--Scan[A, b]
```
It change the plan's schema. when we use CTAS, the schema is same as query plan's output.
Then since we change the schema, it not consistent with origin SQL. So we need to check final result plan's schema with origin plan's schema
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
existed UT
Closes #33703 from AngersZhuuuu/SPARK-36352-3.1.
Authored-by: Angerszhuuuu <an...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 6 ++++--
.../spark/sql/catalyst/optimizer/Optimizer.scala | 23 +++++++++-------------
.../spark/sql/catalyst/rules/RuleExecutor.scala | 6 +++---
.../org/apache/spark/sql/types/DataType.scala | 2 +-
.../org/apache/spark/sql/util/SchemaUtils.scala | 11 +++++++++++
.../sql/catalyst/trees/RuleExecutorSuite.scala | 8 ++++++--
.../sql/execution/adaptive/AQEOptimizer.scala | 12 +++++++----
.../execution/datasources/DataSourceStrategy.scala | 2 +-
.../sql/execution/datasources/SchemaPruning.scala | 10 ++++++----
.../datasources/v2/V2ScanRelationPushDown.scala | 3 ++-
10 files changed, 51 insertions(+), 32 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index b2af21d..62bfd53 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -157,8 +157,10 @@ class Analyzer(override val catalogManager: CatalogManager)
private val v1SessionCatalog: SessionCatalog = catalogManager.v1SessionCatalog
- override protected def isPlanIntegral(plan: LogicalPlan): Boolean = {
- !Utils.isTesting || LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(plan)
+ override protected def isPlanIntegral(
+ previousPlan: LogicalPlan,
+ currentPlan: LogicalPlan): Boolean = {
+ !Utils.isTesting || LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(currentPlan)
}
override def isView(nameParts: Seq[String]): Boolean = v1SessionCatalog.isView(nameParts)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index c3e6310..6584401 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.SchemaUtils._
import org.apache.spark.util.Utils
/**
@@ -44,10 +45,14 @@ abstract class Optimizer(catalogManager: CatalogManager)
// - is still resolved
// - only host special expressions in supported operators
// - has globally-unique attribute IDs
- override protected def isPlanIntegral(plan: LogicalPlan): Boolean = {
- !Utils.isTesting || (plan.resolved &&
- plan.find(PlanHelper.specialExpressionsInUnsupportedOperator(_).nonEmpty).isEmpty &&
- LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(plan))
+ // - optimized plan have same schema with previous plan.
+ override protected def isPlanIntegral(
+ previousPlan: LogicalPlan,
+ currentPlan: LogicalPlan): Boolean = {
+ !Utils.isTesting || (currentPlan.resolved &&
+ currentPlan.find(PlanHelper.specialExpressionsInUnsupportedOperator(_).nonEmpty).isEmpty &&
+ LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(currentPlan) &&
+ DataType.equalsIgnoreNullability(previousPlan.schema, currentPlan.schema))
}
override protected val excludedOnceBatches: Set[String] =
@@ -488,16 +493,6 @@ object RemoveRedundantAliases extends Rule[LogicalPlan] {
* Remove no-op operators from the query plan that do not make any modifications.
*/
object RemoveNoopOperators extends Rule[LogicalPlan] {
- def restoreOriginalOutputNames(
- projectList: Seq[NamedExpression],
- originalNames: Seq[String]): Seq[NamedExpression] = {
- projectList.zip(originalNames).map {
- case (attr: Attribute, name) => attr.withName(name)
- case (alias: Alias, name) => alias.withName(name)
- case (other, _) => other
- }
- }
-
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
// Eliminate no-op Projects
case p @ Project(projectList, child) if child.sameOutput(p) =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
index d5b0884..45910f2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala
@@ -156,7 +156,7 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
* `Optimizer`, so we can catch rules that return invalid plans. The check function returns
* `false` if the given plan doesn't pass the structural integrity check.
*/
- protected def isPlanIntegral(plan: TreeType): Boolean = true
+ protected def isPlanIntegral(previousPlan: TreeType, currentPlan: TreeType): Boolean = true
/**
* Util method for checking whether a plan remains the same if re-optimized.
@@ -196,7 +196,7 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
val beforeMetrics = RuleExecutor.getCurrentMetrics()
// Run the structural integrity checker against the initial input
- if (!isPlanIntegral(plan)) {
+ if (!isPlanIntegral(plan, plan)) {
val message = "The structural integrity of the input plan is broken in " +
s"${this.getClass.getName.stripSuffix("$")}."
throw new TreeNodeException(plan, message, null)
@@ -229,7 +229,7 @@ abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {
tracker.foreach(_.recordRuleInvocation(rule.ruleName, runTime, effective))
// Run the structural integrity checker against the plan after each rule.
- if (effective && !isPlanIntegral(result)) {
+ if (effective && !isPlanIntegral(plan, result)) {
val message = s"After applying rule ${rule.ruleName} in batch ${batch.name}, " +
"the structural integrity of the plan is broken."
throw new TreeNodeException(result, message, null)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
index 9e820f0..016d0bc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataType.scala
@@ -276,7 +276,7 @@ object DataType {
/**
* Compares two types, ignoring nullability of ArrayType, MapType, StructType.
*/
- private[types] def equalsIgnoreNullability(left: DataType, right: DataType): Boolean = {
+ private[sql] def equalsIgnoreNullability(left: DataType, right: DataType): Boolean = {
(left, right) match {
case (ArrayType(leftElementType, _), ArrayType(rightElementType, _)) =>
equalsIgnoreNullability(leftElementType, rightElementType)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
index 4aae7bd..ccbe698 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
@@ -21,6 +21,7 @@ import java.util.Locale
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis._
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, NamedExpression}
import org.apache.spark.sql.connector.expressions.{BucketTransform, FieldReference, NamedTransform, Transform}
import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType}
@@ -201,6 +202,16 @@ private[spark] object SchemaUtils {
}
}
+ def restoreOriginalOutputNames(
+ projectList: Seq[NamedExpression],
+ originalNames: Seq[String]): Seq[NamedExpression] = {
+ projectList.zip(originalNames).map {
+ case (attr: Attribute, name) => attr.withName(name)
+ case (alias: Alias, name) => alias.withName(name)
+ case (other, _) => other
+ }
+ }
+
/**
* Returns the given column's ordinal within the given `schema`. The length of the returned
* position will be as long as how nested the column is.
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/RuleExecutorSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/RuleExecutorSuite.scala
index 60dd9a1..5855289 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/RuleExecutorSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/trees/RuleExecutorSuite.scala
@@ -75,7 +75,9 @@ class RuleExecutorSuite extends SparkFunSuite {
test("structural integrity checker - verify initial input") {
object WithSIChecker extends RuleExecutor[Expression] {
- override protected def isPlanIntegral(expr: Expression): Boolean = expr match {
+ override protected def isPlanIntegral(
+ previousPlan: Expression,
+ currentPlan: Expression): Boolean = currentPlan match {
case IntegerLiteral(_) => true
case _ => false
}
@@ -93,7 +95,9 @@ class RuleExecutorSuite extends SparkFunSuite {
test("structural integrity checker - verify rule execution result") {
object WithSICheckerForPositiveLiteral extends RuleExecutor[Expression] {
- override protected def isPlanIntegral(expr: Expression): Boolean = expr match {
+ override protected def isPlanIntegral(
+ previousPlan: Expression,
+ currentPlan: Expression): Boolean = currentPlan match {
case IntegerLiteral(i) if i > 0 => true
case _ => false
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
index 04b8ade..bddb81c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.adaptive
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, LogicalPlanIntegrity, PlanHelper}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.DataType
import org.apache.spark.util.Utils
/**
@@ -55,9 +56,12 @@ class AQEOptimizer(conf: SQLConf) extends RuleExecutor[LogicalPlan] {
}
}
- override protected def isPlanIntegral(plan: LogicalPlan): Boolean = {
- !Utils.isTesting || (plan.resolved &&
- plan.find(PlanHelper.specialExpressionsInUnsupportedOperator(_).nonEmpty).isEmpty &&
- LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(plan))
+ override protected def isPlanIntegral(
+ previousPlan: LogicalPlan,
+ currentPlan: LogicalPlan): Boolean = {
+ !Utils.isTesting || (currentPlan.resolved &&
+ currentPlan.find(PlanHelper.specialExpressionsInUnsupportedOperator(_).nonEmpty).isEmpty &&
+ LogicalPlanIntegrity.checkIfExprIdsAreGloballyUnique(currentPlan) &&
+ DataType.equalsIgnoreNullability(previousPlan.schema, currentPlan.schema))
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
index e4f001d..439fe75 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
@@ -471,7 +471,7 @@ object DataSourceStrategy
*/
protected[sql] def normalizeExprs(
exprs: Seq[Expression],
- attributes: Seq[AttributeReference]): Seq[Expression] = {
+ attributes: Seq[Attribute]): Seq[Expression] = {
exprs.map { e =>
e transform {
case a: AttributeReference =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala
index 76a6a48..55ae49e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+import org.apache.spark.sql.util.SchemaUtils._
/**
* Prunes unnecessary physical columns given a [[PhysicalOperation]] over a data source relation.
@@ -83,8 +84,8 @@ object SchemaPruning extends Rule[LogicalPlan] {
val prunedRelation = leafNodeBuilder(prunedDataSchema)
val projectionOverSchema = ProjectionOverSchema(prunedDataSchema)
- Some(buildNewProjection(normalizedProjects, normalizedFilters, prunedRelation,
- projectionOverSchema))
+ Some(buildNewProjection(projects, normalizedProjects, normalizedFilters,
+ prunedRelation, projectionOverSchema))
} else {
None
}
@@ -126,6 +127,7 @@ object SchemaPruning extends Rule[LogicalPlan] {
*/
private def buildNewProjection(
projects: Seq[NamedExpression],
+ normalizedProjects: Seq[NamedExpression],
filters: Seq[Expression],
leafNode: LeafNode,
projectionOverSchema: ProjectionOverSchema): Project = {
@@ -144,7 +146,7 @@ object SchemaPruning extends Rule[LogicalPlan] {
// Construct the new projections of our Project by
// rewriting the original projections
- val newProjects = projects.map(_.transformDown {
+ val newProjects = normalizedProjects.map(_.transformDown {
case projectionOverSchema(expr) => expr
}).map { case expr: NamedExpression => expr }
@@ -152,7 +154,7 @@ object SchemaPruning extends Rule[LogicalPlan] {
logDebug(s"New projects:\n${newProjects.map(_.treeString).mkString("\n")}")
}
- Project(newProjects, projectionChild)
+ Project(restoreOriginalOutputNames(newProjects, projects.map(_.name)), projectionChild)
}
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
index d218056..daef757 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.connector.read.{Scan, V1Scan}
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
import org.apache.spark.sql.sources
import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.SchemaUtils._
object V2ScanRelationPushDown extends Rule[LogicalPlan] {
import DataSourceV2Implicits._
@@ -79,7 +80,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] {
val newProjects = normalizedProjects
.map(projectionFunc)
.asInstanceOf[Seq[NamedExpression]]
- Project(newProjects, withFilter)
+ Project(restoreOriginalOutputNames(newProjects, project.map(_.name)), withFilter)
} else {
withFilter
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org