[FLINK-3281] IndexOutOfBoundsException when range partition on empty DataSet

flink-runtime/src/main/java/org/apache/flink/runtime/operators/udf/RangeBoundaryBuilder.java

@@ -60,13 +60,15 @@ public class RangeBoundaryBuilder<T> extends RichMapPartitionFunction<T, Object[
 
 		int boundarySize = parallelism - 1;
 		Object[][] boundaries = new Object[boundarySize][];
-		double avgRange = sampledData.size() / (double) parallelism;
-		int numKey = comparator.getFlatComparators().length;
-		for (int i = 1; i < parallelism; i++) {
-			T record = sampledData.get((int) (i * avgRange));
-			Object[] keys = new Object[numKey];
-			comparator.extractKeys(record, keys, 0);
-			boundaries[i-1] = keys;
+		if (sampledData.size() > 0) {
+			double avgRange = sampledData.size() / (double) parallelism;
+			int numKey = comparator.getFlatComparators().length;
+			for (int i = 1; i < parallelism; i++) {
+				T record = sampledData.get((int) (i * avgRange));
+				Object[] keys = new Object[numKey];
+				comparator.extractKeys(record, keys, 0);
+				boundaries[i-1] = keys;
+			}
 		}
 
 		out.collect(boundaries);

flink-tests/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala

@@ -49,6 +49,38 @@ class PartitionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(
     TestBaseUtils.compareResultsByLinesInMemory(expected, resultPath)
   }
 
+  @Test
+  def testEmptyHashPartition(): Unit = {
+    /*
+     * Test hash partition by tuple field
+     */
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val ds = env.fromCollection(Seq[Tuple1[String]]())
+
+    val unique = ds.partitionByHash(0)
+
+    unique.writeAsText(resultPath, WriteMode.OVERWRITE)
+    env.execute()
+
+    expected = ""
+  }
+
+  @Test
+  def testEmptyRangePartition(): Unit = {
+    /*
+     * Test hash partition by tuple field
+     */
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val ds = env.fromCollection(Seq[Tuple1[String]]())
+
+    val unique = ds.partitionByRange(0)
+
+    unique.writeAsText(resultPath, WriteMode.OVERWRITE)
+    env.execute()
+
+    expected = ""
+  }
+
   @Test
   def testHashPartitionByTupleField(): Unit = {
     /*