[SPARK-43838][SQL] Fix subquery on single table with having clause can't be optimized

[Hisoka-X]

What changes were proposed in this pull request?

Eg:

sql("create view t(c1, c2) as values (0, 1), (0, 2), (1, 2)")

sql("select c1, c2, (select count(*) cnt from t t2 where t1.c1 = t2.c1 " +
"having cnt = 0) from t t1").show() 

The error will throw:

[PLAN_VALIDATION_FAILED_RULE_IN_BATCH] Rule org.apache.spark.sql.catalyst.optimizer.RewriteCorrelatedScalarSubquery in batch Operator Optimization before Inferring Filters generated an invalid plan: The plan becomes unresolved: 'Project [toprettystring(c1#224, Some(America/Los_Angeles)) AS toprettystring(c1)#238, toprettystring(c2#225, Some(America/Los_Angeles)) AS toprettystring(c2)#239, toprettystring(cnt#246L, Some(America/Los_Angeles)) AS toprettystring(scalarsubquery(c1))#240]
+- 'Project [c1#224, c2#225, CASE WHEN isnull(alwaysTrue#245) THEN 0 WHEN NOT (cnt#222L = 0) THEN null ELSE cnt#222L END AS cnt#246L]
   +- 'Join LeftOuter, (c1#224 = c1#224#244)
      :- Project [col1#226 AS c1#224, col2#227 AS c2#225]
      :  +- LocalRelation [col1#226, col2#227]
      +- Project [cnt#222L, c1#224#244, cnt#222L, c1#224, true AS alwaysTrue#245]
         +- Project [cnt#222L, c1#224 AS c1#224#244, cnt#222L, c1#224]
            +- Aggregate [c1#224], [count(1) AS cnt#222L, c1#224]
               +- Project [col1#228 AS c1#224]
                  +- LocalRelation [col1#228, col2#229]The previous plan: Project [toprettystring(c1#224, Some(America/Los_Angeles)) AS toprettystring(c1)#238, toprettystring(c2#225, Some(America/Los_Angeles)) AS toprettystring(c2)#239, toprettystring(scalar-subquery#223 [c1#224 && (c1#224 = c1#224#244)], Some(America/Los_Angeles)) AS toprettystring(scalarsubquery(c1))#240]
:  +- Project [cnt#222L, c1#224 AS c1#224#244]
:     +- Filter (cnt#222L = 0)
:        +- Aggregate [c1#224], [count(1) AS cnt#222L, c1#224]
:           +- Project [col1#228 AS c1#224]
:              +- LocalRelation [col1#228, col2#229]
+- Project [col1#226 AS c1#224, col2#227 AS c2#225]
   +- LocalRelation [col1#226, col2#227] 

The reason of error is the unresolved expression in Join node which generate by subquery decorrelation. The duplicateResolved in Join node are false. That's meaning the Join left and right have same Attribute, in this eg is c1#224. The right c1#224 Attribute generated by having Inputs, because there are wrong having Inputs.

This problem only occurs when there contain having clause.

also do some code format fix.

Why are the changes needed?

Fix subquery bug on single table when use having clause

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add new test

[Hisoka-X]

cc @cloud-fan @jchen5

[cloud-fan]

I think a better fix is to let DeduplicateRelations handle Project with alias as well, which is also a source of conflicting attribute ids.

[Hisoka-X]

I think a better fix is to let DeduplicateRelations handle Project with alias as well, which is also a source of conflicting attribute ids.

@cloud-fan Hi, I already implement DeduplicateRelations to handle with Project. Should I revert change from RewriteCorrelatedScalarSubquery? I think it's a hidden danger

[allisonwang-db]

cc @jchen5

[cloud-fan]

I'm reading the doc of the collectConflictPlans function in this class. I think the problem we have now is there are more plan nodes than the leaf node that can produce new attributes, and we need to handle all of them. Project is not the only way, let's follow what plan nodes collectConflictPlans handles.

[Hisoka-X]

I add all plan node which in collectConflictPlans. Please check again, but there are another problem. How to test it, seem produce an negative case not easy.

[Hisoka-X]

This test unnecessary. Because we can deduplicate those attributes in anti-join / semi-join is a self-join. Please refer #39131

[Hisoka-X]

make sure the AnalysisException will be throw, not AssertionError

[cloud-fan]

Can we really reach this code branch?

[Hisoka-X]

Yes. Usually this error will be thrown by checkAnalysis, but we may call datatype in DeduplicateRelations to cause this exception to be thrown. This change ensures that the thrown exception is consistent.

Change before:

Caused by: sbt.ForkMain$ForkError: java.lang.AssertionError: assertion failed: Scalar subquery should have only one column
	at scala.Predef$.assert(Predef.scala:223)
	at org.apache.spark.sql.catalyst.expressions.ScalarSubquery.dataType(subquery.scala:274)
	at org.apache.spark.sql.catalyst.expressions.Alias.toAttribute(namedExpressions.scala:194)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$$anonfun$findAliases$1.applyOrElse(DeduplicateRelations.scala:530)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$$anonfun$findAliases$1.applyOrElse(DeduplicateRelations.scala:530)
	at scala.PartialFunction.$anonfun$runWith$1$adapted(PartialFunction.scala:145)
	at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
	at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
	at scala.collection.TraversableLike.collect(TraversableLike.scala:407)
	at scala.collection.TraversableLike.collect$(TraversableLike.scala:405)
	at scala.collection.AbstractTraversable.collect(Traversable.scala:108)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$.findAliases(DeduplicateRelations.scala:530)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$.org$apache$spark$sql$catalyst$analysis$DeduplicateRelations$$renewDuplicatedRelations(DeduplicateRelations.scala:120)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$.apply(DeduplicateRelations.scala:40)
	at org.apache.spark.sql.catalyst.analysis.DeduplicateRelations$.apply(DeduplicateRelations.scala:38)

[cloud-fan]

The k8s failure is unrelated, I'm merging it to master, thanks!

[Hisoka-X]

Thanks @cloud-fan for your help and @allisonwang-db