[FLINK-7204] [core] CombineHint.NONE

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.

flink-core/src/main/java/org/apache/flink/api/common/operators/base/ReduceOperatorBase.java

@@ -76,7 +76,13 @@ public enum CombineHint {
  * Use a hash-based strategy. This should be faster in most cases, especially if the number
  * of different keys is small compared to the number of input elements (eg. 1/10).
  */
- HASH
+ HASH,
+
+ /**
+ * Disable the use of a combiner. This can be faster in cases when the number of different keys
+ * is very small compared to the number of input elements (eg. 1/100).
+ */
+ NONE
  }
 
  private CombineHint hint;

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/simple/directed/Simplify.java

@@ -19,6 +19,7 @@
 package org.apache.flink.graph.asm.simple.directed;
 
 import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.operators.base.ReduceOperatorBase.CombineHint;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.graph.Edge;
 import org.apache.flink.graph.Graph;
@@ -44,6 +45,7 @@ public Graph<K, VV, EV> runInternal(Graph<K, VV, EV> input)
  .setParallelism(parallelism)
  .name("Remove self-loops")
  .distinct(0, 1)
+ .setCombineHint(CombineHint.NONE)
  .setParallelism(parallelism)
  .name("Remove duplicate edges");
 

flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/simple/undirected/Simplify.java

@@ -19,6 +19,7 @@
 package org.apache.flink.graph.asm.simple.undirected;
 
 import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.operators.base.ReduceOperatorBase.CombineHint;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.graph.Edge;
 import org.apache.flink.graph.Graph;
@@ -74,6 +75,7 @@ public Graph<K, VV, EV> runInternal(Graph<K, VV, EV> input)
  .setParallelism(parallelism)
  .name("Remove self-loops")
  .distinct(0, 1)
+ .setCombineHint(CombineHint.NONE)
  .setParallelism(parallelism)
  .name("Remove duplicate edges");
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/dag/ReduceNode.java

@@ -63,6 +63,9 @@ public ReduceNode(ReduceOperatorBase<?, ?> operator) {
  case HASH:
  combinerStrategy = DriverStrategy.HASHED_PARTIAL_REDUCE;
  break;
+ case NONE:
+ combinerStrategy = DriverStrategy.SORTED_REDUCE;
+ break;
  default:
  throw new RuntimeException("Unknown CombineHint");
  }

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/GroupReduceWithCombineProperties.java

@@ -105,7 +105,7 @@ public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
  in.setLocalStrategy(LocalStrategy.COMBININGSORT, in.getLocalStrategyKeys(),
  in.getLocalStrategySortOrder());
  }
- return new SingleInputPlanNode(node, "Reduce("+node.getOperator().getName()+")", in,
+ return new SingleInputPlanNode(node, "Reduce ("+node.getOperator().getName()+")", in,
  DriverStrategy.SORTED_GROUP_REDUCE, this.keyList);
  } else {
  // non forward case. all local properties are killed anyways, so we can safely plug in a combiner

flink-optimizer/src/main/java/org/apache/flink/optimizer/operators/ReduceProperties.java

@@ -95,7 +95,7 @@ public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {
  in.getShipStrategySortOrder(), in.getDataExchangeMode());
  toReducer.setLocalStrategy(LocalStrategy.SORT, in.getLocalStrategyKeys(), in.getLocalStrategySortOrder());
 
- return new SingleInputPlanNode(node, "Reduce("+node.getOperator().getName()+")", toReducer,
+ return new SingleInputPlanNode(node, "Reduce ("+node.getOperator().getName()+")", toReducer,
  DriverStrategy.SORTED_REDUCE, this.keyList);
  }
  }