[SPARK-35203][SQL] Improve Repartition statistics estimation

[wangyum]

What changes were proposed in this pull request?

This PR improves Repartition and RepartitionByExpr statistics estimation using child statistics.

Why are the changes needed?

The current implementation will missing column stat. For example:

CREATE TABLE t1 USING parquet AS SELECT id % 10 AS key FROM range(100);
ANALYZE TABLE t1 COMPUTE STATISTICS FOR ALL COLUMNS;
set spark.sql.cbo.enabled=true;
EXPLAIN COST SELECT key FROM (SELECT key FROM t1 DISTRIBUTE BY key) t GROUP BY key;

Before this PR:

== Optimized Logical Plan ==
Aggregate [key#2950L], [key#2950L], Statistics(sizeInBytes=1600.0 B)
+- RepartitionByExpression [key#2950L], Statistics(sizeInBytes=1600.0 B, rowCount=100)
   +- Relation default.t1[key#2950L] parquet, Statistics(sizeInBytes=1600.0 B, rowCount=100)

After this PR:

== Optimized Logical Plan ==
Aggregate [key#2950L], [key#2950L], Statistics(sizeInBytes=160.0 B, rowCount=10)
+- RepartitionByExpression [key#2950L], Statistics(sizeInBytes=1600.0 B, rowCount=100)
   +- Relation default.t1[key#2950L] parquet, Statistics(sizeInBytes=1600.0 B, rowCount=100)

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42380/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42380/

[SparkQA]

Test build #137850 has finished for PR 32309 at commit a010dea.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42392/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42392/

[SparkQA]

Test build #137863 has finished for PR 32309 at commit 9d4c349.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44265/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44265/

[SparkQA]

Test build #139740 has finished for PR 32309 at commit 5553429.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44356/

[maropu]

Q: we need to fall back into the size-based one instead of just calling p.child.stats?

[wangyum]

For better maintenance, if BasicStatsPlanVisitor and SizeInBytesOnlyStatsPlanVisitor have the same implementation, then fallback. For example visitLocalLimit:

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/BasicStatsPlanVisitor.scala

Line 76 in 5553429

override def visitLocalLimit(p: LocalLimit): Statistics = fallback(p)

[maropu]

Ah, I see. It's trivial and sgtm.

[maropu]

Looks fine otherwise.

[maropu]

Thank you, @wangyum . Merged to master.