[SPARK-47094][SQL] SPJ : fix bucket reducer function #47126
Conversation
|
@szehon-ho please take a look. |
| @@ -1558,7 +1558,7 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase { | |||
| } | |||
| } | |||
|
|
|||
| test("SPARK-47094: Support compatible buckets with common divisor") { | |||
| test("SPARK-47094: SPJ:Support compatible buckets with common divisor") { | |||
There was a problem hiding this comment.
Nit: space before Support ? Maybe SPJ is redundant here actually
There was a problem hiding this comment.
All the other ones in SPJ has this SPJ: , can remove it too.
There was a problem hiding this comment.
I see, its fine then, maybe just put the space
|
|
||
| Seq((table1, partition1), (table2, partition2)).foreach { case (tab, part) => | ||
| createTable(tab, columns2, part) | ||
| val insertStr = s"INSERT INTO testcat.ns.$tab VALUES " + |
There was a problem hiding this comment.
I think I added a ton of values to test the positive case, but do you think we can just have the bare minimum values on each side?
| } | ||
|
|
||
|
|
||
| test("SPARK-47094: SPJ: Support compatible buckets common divisor is one of the numbers") { |
There was a problem hiding this comment.
I feel this is already covered by "Support compatible bucket" , and would just say no need to add this test?
There was a problem hiding this comment.
I wanted to check if we can test the fact that there is no grouping happening on the side which has same number of bucket as gcd. Not sure whether we can assert that or not.
There was a problem hiding this comment.
I wanted to check if we can test the fact that there is no grouping happening on the side which has same number of bucket as gcd. Not sure whether we can assert that or not.
Do you mean expected scans?
I'm still not seeing the difference, this test the case that one divides into the other (4, 8), (3, 9), what's the difference with the existing test of (2, 4), (2, 6) above? (Support compatible bucket) It also asserts the expected scans up there too.
| @@ -101,8 +101,8 @@ object BucketFunction extends ScalarFunction[Int] with ReducibleFunction[Int, In | |||
|
|
|||
| if (otherFunc == BucketFunction) { | |||
| val gcd = this.gcd(thisNumBuckets, otherNumBuckets) | |||
| if (gcd != thisNumBuckets) { | |||
| return BucketReducer(thisNumBuckets, gcd) | |||
| if (gcd>1 && gcd!=thisNumBuckets) { | |||
There was a problem hiding this comment.
Nit: spacing (gcd > 1 && gcd != thisNumBuckets)
HyukjinKwon
changed the title
What changes were proposed in this pull request?
SPJ compatible bucket issue has an implementation of reducible function. This patch fixes the implementation and make it same as in apache iceberg one.
Why are the changes needed?
With this fix, incompatible number of buckets do not return 1 as GCD, hence the buckets do not reduce to 1 when it used in incompatible number of buckets.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
With unit tests
Was this patch authored or co-authored using generative AI tooling?
No.