[FLINK-2997] Support range partition with user customized data distribution. #1776

gallenvara · 2016-03-09T02:24:55Z

Sometime user have better knowledge of the source data, and they can build customized data distribution to do range partition more efficiently.

gallenvara · 2016-03-09T14:09:44Z

@fhueske @ChengXiangLi Can you please help me with review work? The error of CI build failure is not relevant.

ChengXiangLi · 2016-03-10T06:44:21Z

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CustomDistributionITCase.java

+
+		String expected = "[(1), (4), (2), (3), (5), (12)]";
+		assertEquals(expected, out1.collect().toString());
+	}


Would you add a test which use 2 fields as partition key?

OK, i will modify the test code.

ChengXiangLi · 2016-03-10T06:45:46Z

Just a minor comment, mostly looks good to me. @fhueske , do you want to take a look at this PR?

fhueske · 2016-03-10T10:20:23Z

flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java

 		this.pKeys = pKeys;
 		this.partitionLocationName = partitionLocationName;
 		this.customPartitioner = customPartitioner;
+		this.distribution = distribution;


Please add a check that a distribution is only set if pMethod == PartitionMethod.RANGE

Other pMethod like hashPartition and customPartition don't need the distribution and the distribution use its default value: null, i think no check here also can make sense.

The check would prevent an invalid parameter combination and also shows readers of the code that such a combination is not valid. Hence, I think the check should be added.

We should check here that the key types of distribution.getKeyTypes() are equal to pKeys.getKeyFieldTypes().

fhueske · 2016-03-10T10:44:24Z

Thanks @gallenvara for opening the PR. Looks mostly good but the tests need to be improved, IMO.

It would also be good to extend the DataDistribution interface by adding a method TypeInformation[] getKeyTypes(). This will allow us to validate the key types of the DataDistribution and the specified keys in the PartitionOperator.

gallenvara · 2016-03-14T10:30:22Z

Hi, @fhueske . I have modified the relevant code. I still use the generic class CustomDistribution for the tests because it is not flexible with build-in data, IMO. Also it is not good if mix the logic of distribution to the tests. Can you please help with review?

fhueske · 2016-03-15T22:49:16Z

Sorry @gallenvara, I was busy the last days. I will look into this PR tomorrow.

fhueske · 2016-03-16T13:30:50Z

flink-optimizer/src/main/java/org/apache/flink/optimizer/traversals/RangePartitionRewriter.java

 		prPlanNode.setParallelism(targetParallelism);
 		GlobalProperties globalProperties = new GlobalProperties();
-		globalProperties.setRangePartitioned(new Ordering(0, null, Order.ASCENDING));
+		globalProperties.setRangePartitioned(new Ordering(0, null, Order.ASCENDING), channel.getDataDistribution());


can you pass null instead of channel.getDataDistribution() to make it clear that no provided data distribution is available?

fhueske · 2016-03-18T15:05:59Z

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CustomDistributionITCase.java

+						checkPartition = false;
+					}
+				}
+				out.collect(checkPartition);


No need to emit something if assert is done in function.

fhueske · 2016-03-18T15:16:07Z

I had a few comments, but we are getting closer :-)

gallenvara · 2016-03-19T09:56:08Z

@fhueske , thanks a lot for review work, codes have been modified based on your advice. I change the second test with modifying the range boundary from (bucketNum+1)*7,bucketNum*2+3 to (bucketNum+1)*(bucketNum+2)/2,bucketNum+1. The reason for this is for the record Tuple3(15, 5L, "Comment#9"), it will be emitted to partition:2 with first field and partition:1 with second field and the test will be failed.

fhueske · 2016-03-21T22:14:53Z

flink-java/src/main/java/org/apache/flink/api/java/operators/PartitionOperator.java

+		if (distribution != null) {
+			Preconditions.checkArgument(distribution.getNumberOfFields() == pKeys.getNumberOfKeyFields(), "The number of key fields in the distribution and range partitioner should be the same.");
+			for (int i = 0; i < pKeys.getNumberOfKeyFields(); i++) {
+				Preconditions.checkArgument(distribution.getKeyTypes()[i].equals(pKeys.getKeyFieldTypes()[i]), "The types of key from the distribution and range partitioner are not equal.");


can you fetch the key type arrays only once and not for every key?

code modified and rebase the new commit with previous one.( you must have stayed up late last night :) )

…n the second test.

fhueske · 2016-03-22T10:11:16Z

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CustomDistributionITCase.java

+	/**
+	 * The class is used to do the tests of range partition with customed data distribution.
+	 */
+	public static class TestDataDist implements DataDistribution {


The two distributions defined by this class have not a lot in common. How about we define two classes, one for each distribution?

gallenvara · 2016-03-23T08:57:59Z

@fhueske , PR has been updated.

fhueske · 2016-03-23T11:06:23Z

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/CustomDistributionITCase.java

+	public static class TestDataDist2 implements DataDistribution {
+
+		public int rightBoundary[] = new int[]{6, 4, 9, 1, 2};
+		private int dim;


dim should always be 2. Remove the field, the constructor, and update write() and read().

fhueske · 2016-03-23T11:09:08Z

Thanks for the quick updates @gallenvara! I think we are almost done. The test distribution and input data looks good. Only the boundary checks need to be fixed.
Thanks, Fabian

gallenvara · 2016-03-23T11:59:36Z

@fhueske codes has been modified :)

…with lower and upper boundary.

fhueske · 2016-03-23T13:11:50Z

Thanks for the fast update. The PR looks good. :-)
I will merge it.

… data distribution. This closes apache#1776

gallenvara force-pushed the flink-2997 branch 2 times, most recently from a3cd265 to e91407d Compare March 9, 2016 04:30

ChengXiangLi reviewed Mar 10, 2016
View reviewed changes

gallenvara added 2 commits March 10, 2016 17:20

Enable range partition with custom data distribution.

c337073

Modify the first test of range partition with two key fields.

1970474

gallenvara force-pushed the flink-2997 branch from 95c981a to 1970474 Compare March 10, 2016 09:21

fhueske reviewed Mar 10, 2016
View reviewed changes

Modify the test to validate the partitioning correctness.

7e2783f

gallenvara force-pushed the flink-2997 branch from 1f90403 to 7e2783f Compare March 14, 2016 13:10

fhueske reviewed Mar 16, 2016
View reviewed changes

fhueske reviewed Mar 18, 2016
View reviewed changes

gallenvara force-pushed the flink-2997 branch from 8e29ab3 to 284fe28 Compare March 19, 2016 09:38

gallenvara force-pushed the flink-2997 branch 4 times, most recently from 7094543 to ecdb437 Compare March 20, 2016 19:16

fhueske reviewed Mar 21, 2016
View reviewed changes

Move TestDataDist class into the test and modify the range boundary i…

a2ec639

…n the second test.

gallenvara force-pushed the flink-2997 branch from 7eb9b87 to a2ec639 Compare March 21, 2016 23:40

fhueske reviewed Mar 22, 2016
View reviewed changes

gallenvara force-pushed the flink-2997 branch from d6ca8be to acbf64e Compare March 23, 2016 09:01

fhueske reviewed Mar 23, 2016
View reviewed changes

gallenvara force-pushed the flink-2997 branch from d9a8a83 to 28b3570 Compare March 23, 2016 12:03

Modify data distribution of the second test and check the partitions …

da3f857

…with lower and upper boundary.

gallenvara force-pushed the flink-2997 branch from 11c90b6 to da3f857 Compare March 23, 2016 12:42

asfgit closed this in b8299bf Mar 23, 2016

fijolekProjects pushed a commit to fijolekProjects/flink that referenced this pull request May 1, 2016

[FLINK-2997] [dataSet] Add support for range partitioning with custom…

4672388

… data distribution. This closes apache#1776

rmetzger added the component=<none> label Mar 14, 2019

[FLINK-2997] Support range partition with user customized data distribution. #1776

[FLINK-2997] Support range partition with user customized data distribution. #1776

Uh oh!

Conversation

gallenvara commented Mar 9, 2016

Uh oh!

gallenvara commented Mar 9, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ChengXiangLi commented Mar 10, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske commented Mar 10, 2016

Uh oh!

gallenvara commented Mar 14, 2016

Uh oh!

fhueske commented Mar 15, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske commented Mar 18, 2016

Uh oh!

gallenvara commented Mar 19, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gallenvara commented Mar 23, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske commented Mar 23, 2016

Uh oh!

gallenvara commented Mar 23, 2016

Uh oh!

fhueske commented Mar 23, 2016

Uh oh!

Uh oh!