[FLINK-7204] [core] CombineHint.NONE

[greghogan]

FLINK-3477 added a hash-combine preceding the reducer configured with CombineHint.HASH or CombineHint.SORT (default). In some cases it may be useful to disable the combiner in ReduceNode by specifying a new CombineHint.NONE value.

We don't currently have unit tests for this part of the code but I'm happy to discuss and add some.

[fhueske]

Hi @greghogan, I think this is definitely a valuable improvement.
However, I looked at the plan of a query with a ReduceFunction with CombineHint.None and found that it still has a combine node:

[IN] -FWD-> [COMBINE(SORTED_REDUCE)] -HASH-> [REDUCE(SORTED_REDUCE)] -> [OUT]

I think we should

1. set combinerStrategy to DriverStrategy.NONE in ReduceNode for CombineHint.NONE.
2. change ReduceProperties.instantiate() with a check for combinerStrategy == NONE and not add a node for the combiner in that case.
3. add a test in ReduceCompilationTest similar to testGroupedReduceWithHint() to check that the resulting plan has no combine node.

Best, Fabian

[greghogan]

@fhueske updated and manually verified. I had misinterpreted DriverStrategy to imply that SORTED_REDUCE would not use a combiner.

[fhueske]

Hi @greghogan, the changes look good and the PR is ready to be merged.

If you like, you can simplify the test before merging by using field position keys (as in testGroupedReduceWithFieldPositionKey()) instead of a KeySelector. That would remove the key extractor and projector map operators and make the plan a bit easier to verify.

[greghogan]

Test updated.

[fhueske]

+1 to merge