[SPARK-41220][SQL] Range partitioner sample supports column pruning

[ulysses-you]

What changes were proposed in this pull request?

Make RangePartitioning take a new parameter planForSample, so we can pass the it to ShuffleExchangeExec.

Add a new rule OptimizeSampleForRangePartitioning to infer the planForSample if has benefit at preparations phase.

Why are the changes needed?

When do a global sort or repartition by range, firstly we do sample to get the range bounds, then use the range partitioner to do shuffle exchange.

The issue is, the plan for sample is coupled with the original query. What we need for sample plan is the columns for sort order but the original query plan contains all data columns. So we can do column pruning for the sample plan to only fetch the ordering columns.

A common example is: OPTIMIZE table ZORDER BY columns

Does this PR introduce any user-facing change?

no it's improve performance

How was this patch tested?

add tests

performace test:

val rows = 1000000
val columns = (0 until(200)).map(i => s"uuid() c$i")
spark.range(rows).selectExpr(columns: _*).repartition(30).write.format("parquet").saveAsTable("t")
spark.sql("SELECT * FROM t ORDER BY c0, c1, c2").write.format("noop").mode("overwrite").save()

after do column pruning, the stage for sample from 7s -> 0.4s, the whole query 1.4X

[ulysses-you]

how about this idea ? cc @yaooqinn @wangyum @HyukjinKwon @cloud-fan

[cloud-fan]

I'd rather have a nested query execution for sample, instead of changing the query plan tree.

[ulysses-you]

@cloud-fan how about this updating ? I make a new query execution before shuffle, so that we can have a optimized plan to sample.

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!