[FLINK-3064] [core] Add size check in GroupReduceOperatorBase

This closes apache#1396.
abhishekagarwal87 · Nov 24, 2015 · d2b4391 · d2b4391
1 parent 66b5def
commit d2b4391
Showing 1 changed file with 20 additions and 18 deletions.
diff --git a/...ore/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java b/...ore/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java
@@ -195,28 +195,30 @@ public int compare(IN o1, IN o2) {
 
 		ArrayList<OUT> result = new ArrayList<OUT>();
 
-		if (keyColumns.length == 0) {
-			final TypeSerializer<IN> inputSerializer = inputType.createSerializer(executionConfig);
-			TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
-			List<IN> inputDataCopy = new ArrayList<IN>(inputData.size());
-			for (IN in: inputData) {
-				inputDataCopy.add(inputSerializer.copy(in));
-			}
-			CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
+		if (inputData.size() > 0) {
+			if (keyColumns.length == 0) {
+				final TypeSerializer<IN> inputSerializer = inputType.createSerializer(executionConfig);
+				TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
+				List<IN> inputDataCopy = new ArrayList<IN>(inputData.size());
+				for (IN in : inputData) {
+					inputDataCopy.add(inputSerializer.copy(in));
+				}
+				CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
 
-			function.reduce(inputDataCopy, collector);
-		} else {
-			final TypeSerializer<IN> inputSerializer = inputType.createSerializer(executionConfig);
-			boolean[] keyOrderings = new boolean[keyColumns.length];
-			final TypeComparator<IN> comparator = getTypeComparator(inputType, keyColumns, keyOrderings, executionConfig);
+				function.reduce(inputDataCopy, collector);
+			} else {
+				final TypeSerializer<IN> inputSerializer = inputType.createSerializer(executionConfig);
+				boolean[] keyOrderings = new boolean[keyColumns.length];
+				final TypeComparator<IN> comparator = getTypeComparator(inputType, keyColumns, keyOrderings, executionConfig);
 
-			ListKeyGroupedIterator<IN> keyedIterator = new ListKeyGroupedIterator<IN>(inputData, inputSerializer, comparator);
+				ListKeyGroupedIterator<IN> keyedIterator = new ListKeyGroupedIterator<IN>(inputData, inputSerializer, comparator);
 
-			TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
-			CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
+				TypeSerializer<OUT> outSerializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
+				CopyingListCollector<OUT> collector = new CopyingListCollector<OUT>(result, outSerializer);
 
-			while (keyedIterator.nextKey()) {
-				function.reduce(keyedIterator.getValues(), collector);
+				while (keyedIterator.nextKey()) {
+					function.reduce(keyedIterator.getValues(), collector);
+				}
 			}
 		}