diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index 39f554c137c988cdddf5db11eef0637722a822f9..7701fd04510411c9b977a9a3eaade14477ce22d9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -137,6 +137,12 @@ trait CheckAnalysis {
}
}
+ case s @ SetOperation(left, right) if left.output.length != right.output.length =>
+ failAnalysis(
+ s"${s.nodeName} can only be performed on tables with the same number of columns, " +
+ s"but the left table has ${left.output.length} columns and the right has " +
+ s"${right.output.length}")
+
case _ => // Fallbacks to the following checks
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
index 2cb067f4aac91588dce4280d470335304ff23d90..a1aa2a2b2c6800435df5f183d52bbd4eba4d938b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
@@ -203,6 +203,7 @@ object HiveTypeCoercion {
planName: String,
left: LogicalPlan,
right: LogicalPlan): (LogicalPlan, LogicalPlan) = {
+ require(left.output.length == right.output.length)
val castedTypes = left.output.zip(right.output).map {
case (lhs, rhs) if lhs.dataType != rhs.dataType =>
@@ -229,15 +230,10 @@ object HiveTypeCoercion {
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
case p if p.analyzed => p
- case u @ Union(left, right) if u.childrenResolved && !u.resolved =>
- val (newLeft, newRight) = widenOutputTypes(u.nodeName, left, right)
- Union(newLeft, newRight)
- case e @ Except(left, right) if e.childrenResolved && !e.resolved =>
- val (newLeft, newRight) = widenOutputTypes(e.nodeName, left, right)
- Except(newLeft, newRight)
- case i @ Intersect(left, right) if i.childrenResolved && !i.resolved =>
- val (newLeft, newRight) = widenOutputTypes(i.nodeName, left, right)
- Intersect(newLeft, newRight)
+ case s @ SetOperation(left, right) if s.childrenResolved
+ && left.output.length == right.output.length && !s.resolved =>
+ val (newLeft, newRight) = widenOutputTypes(s.nodeName, left, right)
+ s.makeCopy(Array(newLeft, newRight))
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
index 73b8261260acba889f288cbdf248f114d5407dc9..722f69cdca827b69b9fe3f785226b1d33e771280 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -89,13 +89,21 @@ case class Filter(condition: Expression, child: LogicalPlan) extends UnaryNode {
override def output: Seq[Attribute] = child.output
}
-case class Union(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
+abstract class SetOperation(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
// TODO: These aren't really the same attributes as nullability etc might change.
- override def output: Seq[Attribute] = left.output
+ final override def output: Seq[Attribute] = left.output
- override lazy val resolved: Boolean =
+ final override lazy val resolved: Boolean =
childrenResolved &&
- left.output.zip(right.output).forall { case (l, r) => l.dataType == r.dataType }
+ left.output.length == right.output.length &&
+ left.output.zip(right.output).forall { case (l, r) => l.dataType == r.dataType }
+}
+
+private[sql] object SetOperation {
+ def unapply(p: SetOperation): Option[(LogicalPlan, LogicalPlan)] = Some((p.left, p.right))
+}
+
+case class Union(left: LogicalPlan, right: LogicalPlan) extends SetOperation(left, right) {
override def statistics: Statistics = {
val sizeInBytes = left.statistics.sizeInBytes + right.statistics.sizeInBytes
@@ -103,6 +111,10 @@ case class Union(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
}
}
+case class Intersect(left: LogicalPlan, right: LogicalPlan) extends SetOperation(left, right)
+
+case class Except(left: LogicalPlan, right: LogicalPlan) extends SetOperation(left, right)
+
case class Join(
left: LogicalPlan,
right: LogicalPlan,
@@ -142,15 +154,6 @@ case class BroadcastHint(child: LogicalPlan) extends UnaryNode {
override def output: Seq[Attribute] = child.output
}
-
-case class Except(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
- override def output: Seq[Attribute] = left.output
-
- override lazy val resolved: Boolean =
- childrenResolved &&
- left.output.zip(right.output).forall { case (l, r) => l.dataType == r.dataType }
-}
-
case class InsertIntoTable(
table: LogicalPlan,
partition: Map[String, Option[String]],
@@ -160,7 +163,7 @@ case class InsertIntoTable(
extends LogicalPlan {
override def children: Seq[LogicalPlan] = child :: Nil
- override def output: Seq[Attribute] = child.output
+ override def output: Seq[Attribute] = Seq.empty
assert(overwrite || !ifNotExists)
override lazy val resolved: Boolean = childrenResolved && child.output.zip(table.output).forall {
@@ -440,10 +443,3 @@ case object OneRowRelation extends LeafNode {
override def statistics: Statistics = Statistics(sizeInBytes = 1)
}
-case class Intersect(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
- override def output: Seq[Attribute] = left.output
-
- override lazy val resolved: Boolean =
- childrenResolved &&
- left.output.zip(right.output).forall { case (l, r) => l.dataType == r.dataType }
-}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
index 7065adce04bf83567472b8069f837b0fc1293a87..fbdd3a7776f50fb1ff554cfbf9e600f718715049 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala
@@ -145,6 +145,24 @@ class AnalysisErrorSuite extends AnalysisTest {
UnresolvedTestPlan(),
"unresolved" :: Nil)
+ errorTest(
+ "union with unequal number of columns",
+ testRelation.unionAll(testRelation2),
+ "union" :: "number of columns" :: testRelation2.output.length.toString ::
+ testRelation.output.length.toString :: Nil)
+
+ errorTest(
+ "intersect with unequal number of columns",
+ testRelation.intersect(testRelation2),
+ "intersect" :: "number of columns" :: testRelation2.output.length.toString ::
+ testRelation.output.length.toString :: Nil)
+
+ errorTest(
+ "except with unequal number of columns",
+ testRelation.except(testRelation2),
+ "except" :: "number of columns" :: testRelation2.output.length.toString ::
+ testRelation.output.length.toString :: Nil)
+
errorTest(
"SPARK-9955: correct error message for aggregate",
// When parse SQL string, we will wrap aggregate expressions with UnresolvedAlias.
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
index bbe8c1911bf868ce6ee6538623819262c2285d3a..98d21aa76d64e75a229ff25ec3cf79e7e61e1245 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala
@@ -751,7 +751,7 @@ private[hive] case class InsertIntoHiveTable(
extends LogicalPlan {
override def children: Seq[LogicalPlan] = child :: Nil
- override def output: Seq[Attribute] = child.output
+ override def output: Seq[Attribute] = Seq.empty
val numDynamicPartitions = partition.values.count(_.isEmpty)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
index 12c667e6e92da7f3a616cec50cf4cb858fe7a541..62efda613a176aae866fc5564a7ffabba852656d 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
@@ -61,7 +61,7 @@ case class InsertIntoHiveTable(
serializer
}
- def output: Seq[Attribute] = child.output
+ def output: Seq[Attribute] = Seq.empty
def saveAsHiveFile(
rdd: RDD[InternalRow],