[FLINK-3179] Combiner is not injected if Reduce or GroupReduce input is explicitly partitioned

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.10.1
• Fix Version/s: 1.0.1, 1.1.0
• Component/s: API / DataSet
• Assignee: ramkrishna.s.vasudevan
• Reporter: Fabian Hueske
• Created: 16/Dec/15 20:02
• Updated: 28/Feb/19 14:30
• Resolved: 22/Mar/16 13:46

Description

The optimizer does not inject a combiner if the input of a Reducer or GroupReducer is explicitly partitioned as in the following example

DataSet<Tuple2<String,Integer>> words = ...
DataSet<Tuple2<String,Integer>> counts = words
  .partitionByHash(0)
  .groupBy(0)
  .sum(1);

Explicit partitioning can be useful to enforce partitioning on a subset of keys or to use a different partitioning method (custom or range partitioning).

This issue should be fixed by changing the instantiate() methods of the ReduceProperties and GroupReduceWithCombineProperties classes such that a combine is injected in front of a PartitionPlanNode if it is the input of a Reduce or GroupReduce operator. This should only happen, if the Reducer is the only successor of the Partition operator.