[SPARK-37961][SQL] Override maxRows/maxRowsPerPartition for some logical operators

[zhengruifeng]

What changes were proposed in this pull request?

1, override maxRowsPerPartition in Sort,Expand,Sample,CollectMetrics;
2, override maxRows in Except,Expand,CollectMetrics;

Why are the changes needed?

to provide an accurate value if possible

Does this PR introduce any user-facing change?

No

How was this patch tested?

added testsuites

[zhengruifeng]

I am not sure whether it is wrong, sampling with replacement should not generate more rows than the input dataset.

But we can not impl a strict sampling with replacement, so PoissonSampler is used instead, which can not guarantee this attribute.

scala> val df = spark.range(0, 1000)
df: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> df.count
res0: Long = 1000

scala> df.sample(true, 0.999999, 10).count
res1: Long = 1004

[cloud-fan]

interesting, cc @sigmod @maryannxue @srielau

this seems like a correctness issue and we should fix it ASAP. @zhengruifeng can you open a new PR for the sample fix only so that we can backport?

[zhengruifeng]

@cloud-fan Sure

[HyukjinKwon]

cc @wangyum FYI

[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!

[cloud-fan]

Somehow this PR lost track. @zhengruifeng do you want to reopen it and get it in?

[zhengruifeng]

@cloud-fan Sure, Let me update this PR

[cloud-fan]

LGTM if tests pass

[zhengruifeng]

@cloud-fan the tests passed after I rebased the PR, it should be ready now.

[cloud-fan]

thanks, merging to master!