[FLINK-27336] Avoid merging when there is only one record

[SteNicholas]

If there is just one record, still use MergeFunction to merge. This is not necessary, just output directly.

The brief change log

• Updates the merge in the SortBufferMemTable and SortMergeReader to output directly if there is only one record, and use MergeFunction to merge in the case of multiple records.

[tsreaper]

Thanks for your contribution! Overall this is a good optimization but I have some doubts on the implementation.

I think we should introduce an abstract merger class to deal with this optimization instead of implementing it in the caller, otherwise we'll have to repeat the implementation every time we use a merger.

Also there is no tests verifying this change. Please add some tests.

[JingsongLi]

Thanks @SteNicholas for the contribution. Left two comments.

Currently we have to change the func wherever we use it. To avoid this, can we create a MergeFunctionHelper (maybe there is a better name) class which is responsible for avoiding calling MergeFunction when there is only one record in it.

We can also add tests against this class.

[SteNicholas]

@JingsongLi @tsreaper, IMO, the current tests could cover the execution logic branch, hence I have not added some extra tests to verify. WDYT?

[JingsongLi]

@JingsongLi @tsreaper, IMO, the current tests could cover the execution logic branch, hence I have not added some extra tests to verify. WDYT?

At least, you can add some unit tests on the new Helper.

[LadyForest]

Hi @SteNicholas, are you still following up?

[SteNicholas]

@JingsongLi, I have addressed above comments and the failure of the UTCase/ITCase isn't related to this change.
PTAL.

[JingsongLi]

Thanks for the update! Looks good to me!