Core: Fix performance issue when combining tasks by partition

[aokolnychyi]

This PR fixes a substantial performance issue in combining tasks by partition for SPJ, which was caused by repetitive creation of PartitionData, which triggered expensive initialization.

Before

Benchmark                                                    Mode  Cnt   Score   Error  Units
TaskGroupPlanningBenchmark.planTaskGroups                      ss    5   2.879 ± 0.209   s/op
TaskGroupPlanningBenchmark.planTaskGroups:async                ss          NaN            ---
TaskGroupPlanningBenchmark.planTaskGroupsWithGrouping          ss    5  26.564 ± 0.410   s/op
TaskGroupPlanningBenchmark.planTaskGroupsWithGrouping:async    ss          NaN            ---

After

Benchmark                                                    Mode  Cnt  Score   Error  Units
TaskGroupPlanningBenchmark.planTaskGroups                      ss    5  2.830 ± 0.211   s/op
TaskGroupPlanningBenchmark.planTaskGroups:async                ss         NaN            ---
TaskGroupPlanningBenchmark.planTaskGroupsWithGrouping          ss    5  5.167 ± 0.464   s/op
TaskGroupPlanningBenchmark.planTaskGroupsWithGrouping:async    ss         NaN            ---

[aokolnychyi]

Moved from set to avoid calling that public method in the new constructor.

[aokolnychyi]

This line was causing the issue.

[amogh-jahagirdar]

Just for my understanding, the repeated calling of this constructor was expensive because the repeated data array initialization (and possibly the determining of the partitionType via AvroSchemaUtil.convert)? And now we avoid repeatedly performing those computations by having a PartitionData be initialized with a "template" just once for all tasks and then replace the actual data in PartitionData#copyFor?

[aokolnychyi]

Yep, correct, @amogh-jahagirdar.

[aokolnychyi]

An alternative to this is to create a new StructLike class simply backed by an array and populate it in the table scan utility. I went with this approach because we do a similar trick in other places.

[rdblue]

I find this API to be a bit awkward. I'd rather have a more straightforward two-step process to copy the key without copying values (like emptyCopy()) and then fill that with values. Otherwise "copy" doesn't really describe this action, which copies behavior but not data.

[aokolnychyi]

I am not entirely happy too. I followed what we did in other places like StructLikeProjection.

Do we really need PartitionData here? What if we simply create a container struct backed by an array and use it?

My worry with emptyCopy() is that the result object is in a weird state, I am not sure it is actually better than copyFor().

[aokolnychyi]

We only need to a serializable projection used as a grouping key. Not more.

[advancedxy]

An alternative to this is to create a new StructLike class simply backed by an array and populate it in the table scan utility

There's already org.apache.iceberg.io.StructCopy class, maybe we can leverage that.

[rdblue]

Thanks for the fix, @aokolnychyi! I think it's important to get this into 1.5 so I merged this. The method name should be okay.