[SPARK-47319][SQL] Improve missingInput calculation

### What changes were proposed in this pull request? This PR improves `QueryPlan.missingInput()` calculation. ### Why are the changes needed? This seems to be the root cause of `DeduplicateRelations` slowness in some cases. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#45424 from peter-toth/fix-missinginput. Authored-by: Peter Toth <[email protected]> Signed-off-by: Kent Yao <[email protected]>
panbingkun · Mar 8, 2024 · f659f8d · f659f8d
1 parent 54f1572
commit f659f8d
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 8 deletions.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AttributeSet.scala
@@ -104,13 +104,19 @@ class AttributeSet private (private val baseSet: mutable.LinkedHashSet[Attribute
    * in `other`.
    */
   def --(other: Iterable[NamedExpression]): AttributeSet = {
-    other match {
-      // SPARK-32755: `--` method behave differently under scala 2.12 and 2.13,
-      // use a Scala 2.12 based code to maintains the insertion order in Scala 2.13
-      case otherSet: AttributeSet =>
-        new AttributeSet(baseSet.clone() --= otherSet.baseSet)
-      case _ =>
-        new AttributeSet(baseSet.clone() --= other.map(a => new AttributeEquals(a.toAttribute)))
+    if (isEmpty) {
+      AttributeSet.empty
+    } else if (other.isEmpty) {
+      this
+    } else {
+      other match {
+        // SPARK-32755: `--` method behave differently under scala 2.12 and 2.13,
+        // use a Scala 2.12 based code to maintains the insertion order in Scala 2.13
+        case otherSet: AttributeSet =>
+          new AttributeSet(baseSet.clone() --= otherSet.baseSet)
+        case _ =>
+          new AttributeSet(baseSet.clone() --= other.map(a => new AttributeEquals(a.toAttribute)))
+      }
     }
   }
 

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -102,7 +102,13 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]]
   /**
    * Attributes that are referenced by expressions but not provided by this node's children.
    */
-  final def missingInput: AttributeSet = references -- inputSet
+  final def missingInput: AttributeSet = {
+    if (references.isEmpty) {
+      AttributeSet.empty
+    } else {
+      references -- inputSet
+    }
+  }
 
   /**
    * Runs [[transformExpressionsDown]] with `rule` on all expressions present