[SPARK-48030][SQL] SPJ: cache rowOrdering and structType for InternalRowComparableWrapper

[advancedxy]

What changes were proposed in this pull request?

Cache rowOrdering and structType for InternalRowComparableWrapper

Why are the changes needed?

For performance improvement

Does this PR introduce any user-facing change?

NO

How was this patch tested?

Added a new benchmark to verify the performance improvement

Was this patch authored or co-authored using generative AI tooling?

NO

[advancedxy]

Before applying changes in this PR, the benchmark code(in this PR) took:

[info] OpenJDK 64-Bit Server VM 21.0.2 on Mac OS X 14.4.1
[info] Apple M1
[info] internal row comparable wrapper:          Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] toSet                                             10785          11369         825          0.0       53925.3       1.0X
[info] mergePartitions                                   21073          21447         528          0.0      105364.5       0.5X

After this PR:

[info] OpenJDK 64-Bit Server VM 21.0.2 on Mac OS X 14.4.1
[info] Apple M1
[info] internal row comparable wrapper:          Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] toSet                                                85            103          25          2.4         425.0       1.0X
[info] mergePartitions                                     163            167           4          1.2         816.3       0.5X

[advancedxy]

@sunchao, @szehon-ho and @yabola would you mind to take a look at this and help review this one.

[advancedxy]

1024 should be sufficient, it could be a SQL configuration though.

[szehon-ho]

Makes sense to me to have a config, though not familiar with spark preference for these things.

[yabola]

After a quick look, this looks good to me. In addition, is there a better expireAfterAccess configuration for NonFateSharingCache?

[advancedxy]

In addition, is there a better expireAfterAccess configuration for NonFateSharingCache?

hmmm, maybe. However, the memory usage of cache should be relatively low. Let's wait for other people's opinions

[yabola]

@advancedxy It appears that this PR can enhance performance when there are a large number of partitions. Could you please share the test results from a real DatasourceV2 table, such as Iceberg?

[advancedxy]

It appears that this PR can enhance performance when there are a large number of partitions. Could you please share the test results from a real DatasourceV2 table, such as Iceberg?

I cannot share the exact numbers. However, I have described the estimated number of partitions(~N00_000, where N <=2)and time to plan in the jira. After this patch, the planning time should be dropped to seconds(from tens of minutes). Hope that helps.

[szehon-ho]

Sounds like a good find and promising results on real table.

[szehon-ho]

Makes sense to me to have a config, though not familiar with spark preference for these things.

[sunchao]

LGTM. It would be nice to have something to configure this but I don't think it is super important. I feel the default value should be more than enough for most use cases? Similarly for expireAfterAccess.

[sunchao]

Thanks! merged to master.

[advancedxy]

LGTM. It would be nice to have something to configure this but I don't think it is super important. I feel the default value should be more than enough for most use cases? Similarly for expireAfterAccess.

Thanks for reviewing this.