You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by kl0u <gi...@git.apache.org> on 2017/03/09 17:17:39 UTC
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
GitHub user kl0u opened a pull request:
https://github.com/apache/flink/pull/3501
[FLINK-5874] Restrict key types in the DataStream API.
Rejects a type from being a key in `DataStream.keyBy()` if it is:
1. it is a POJO type but does not override the `hashCode()` and
relies on the `Object.hashCode()` implementation.
2. it is an array of any type.
This was also discussed with @fhueske
R @aljoscha
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kl0u/flink array-keys
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3501.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3501
----
commit bd00f9d37d5e3e1960b8bf1fff506293784b3039
Author: kl0u <kk...@gmail.com>
Date: 2017-03-08T11:11:07Z
[FLINK-5874] Restrict key types in the DataStream API.
Reject a type from being a key in keyBy() if it is:
1. it is a POJO type but does not override the hashCode() and
relies on the Object.hashCode() implementation.
2. it is an array of any type.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3501: [FLINK-5874] Restrict key types in the DataStream API.
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3501
This should work and is well tested, good job. Had a bunch of minor comments, but nothing critical.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105240936
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.setMaxParallelism(1);
+
+ DataStream<Integer> input = env.fromElements(new Integer(10000));
+
+ TypeInformation<?> expectedTypeInfo = IntegerTypeInfo.INT_TYPE_INFO;
+
+ KeyedStream<Integer, Integer> keyedStream = input.keyBy(new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ Assert.assertEquals(expectedTypeInfo, keyedStream.getKeyType());
+
+ keyedStream.addSink(new SinkFunction<Integer>() {
+ @Override
+ public void invoke(Integer value) throws Exception {
+ Assert.assertEquals(10000L, (long) value);
+ }
+ });
+
+ env.execute();
+ }
+
+ private static class TestClass {
+
+ private final int id;
+
+ TestClass(int id) {
+ this.id = id;
+ }
+ }
+
+ public static class POJOnoHashCode {
--- End diff --
no being lower case appears to violate the naming convention.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/3501
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105241358
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.setMaxParallelism(1);
+
+ DataStream<Integer> input = env.fromElements(new Integer(10000));
+
+ TypeInformation<?> expectedTypeInfo = IntegerTypeInfo.INT_TYPE_INFO;
+
+ KeyedStream<Integer, Integer> keyedStream = input.keyBy(new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ Assert.assertEquals(expectedTypeInfo, keyedStream.getKeyType());
+
+ keyedStream.addSink(new SinkFunction<Integer>() {
+ @Override
+ public void invoke(Integer value) throws Exception {
+ Assert.assertEquals(10000L, (long) value);
+ }
+ });
+
+ env.execute();
+ }
+
+ private static class TestClass {
+
+ private final int id;
+
+ TestClass(int id) {
+ this.id = id;
+ }
+ }
+
+ public static class POJOnoHashCode {
+
+ private int[] id;
+
+ public POJOnoHashCode() {}
+
+ public POJOnoHashCode(int[] id) {
+ this.id = id;
+ }
+
+ public int[] getId() {
+ return id;
+ }
+
+ public void setId(int[] id) {
+ this.id = id;
+ }
+ }
+
+ public static class POJOwithHashCode {
--- End diff --
You could extend PojoNoHashCode and only implement hashCode.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105240820
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
--- End diff --
I would shorten this to read ```returns true if the type overrides the hashcode implementation```. The details can be container in the general javadoc of the method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3501: [FLINK-5874] Restrict key types in the DataStream API.
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3501
+1 to merge.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105239799
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link #hashCode()} method and relies on
+ * the {@link Object#hashCode()} implementation.</li>
+ * <li>it is an array of any type (see {@link PrimitiveArrayTypeInfo}, {@link BasicArrayTypeInfo},
+ * {@link ObjectArrayTypeInfo}).</li>
+ * </ol>,
+ * {@code true} otherwise.
+ */
+ private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo) ?
--- End diff --
? and : are typically at the start of a new line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3501: [FLINK-5874] Restrict key types in the DataStream API.
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3501
Thanks for the thorough review @zentol . I integrated the comments that I agree with and commented on the rest. If you agree with the changes, I can merge it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105238197
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
--- End diff --
type: Arround -> Around
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105407025
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,243 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Object[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Object[]>() {
+
+ @Override
+ public Object[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ Object[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<Object[], Object> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ Object[].class, new GenericTypeInfo<>(Object.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage(new StringStartsWith("Type " + expectedKeyType + " cannot be used as key."));
+
+ input.keyBy(keySelector);
+ }
+
+ //////////////// Composite Key Tests : POJOs ////////////////
+
+ @Test
+ public void testPOJOWithNestedArrayNoHashCodeKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOWithHashCode> input = env.fromElements(
+ new POJOWithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage(new StringStartsWith("Type " + expectedTypeInfo + " cannot be used as key."));
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testPOJOWithNestedArrayAndHashCodeWorkAround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOWithHashCode> input = env.fromElements(
+ new POJOWithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOWithHashCode, POJOWithHashCode>() {
+ @Override
+ public POJOWithHashCode getKey(POJOWithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOWithHashCode>() {
+ @Override
+ public void invoke(POJOWithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOWithoutHashCode, POJOWithoutHashCode> keySelector =
+ new KeySelector<POJOWithoutHashCode, POJOWithoutHashCode>() {
+ @Override
+ public POJOWithoutHashCode getKey(POJOWithoutHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOWithoutHashCode> input = env.fromElements(
+ new POJOWithoutHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ //////////////// Composite Key Tests : Tuples ////////////////
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage(new StringStartsWith("Type " + expectedTypeInfo + " cannot be used as key."));
+
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyAcceptance() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.setMaxParallelism(1);
+
+ DataStream<Integer> input = env.fromElements(new Integer(10000));
+
+ KeyedStream<Integer, Object> keyedStream = input.keyBy(new KeySelector<Integer, Object>() {
+ @Override
+ public Object getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ keyedStream.addSink(new SinkFunction<Integer>() {
+ @Override
+ public void invoke(Integer value) throws Exception {
+ Assert.assertEquals(10000L, (long) value);
+ }
+ });
+
+ env.execute();
--- End diff --
We don't need to execute a job, correct? just the keyBy passing should be enough.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105242176
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
--- End diff --
The focus of this test appears to be to verify that you can use an array as a key by wrapping it in a Pojo that implements HashCode. (based on the naming).
We should probably focus more on the Pojo-with-hashCode implementation part; doesn't really matter what is contained. We are never checking that anyway.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105241262
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
--- End diff --
The name should be changed; as it stands i expected that this test verifies that a primitive key is rejected.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105406561
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,243 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Object[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Object[]>() {
+
+ @Override
+ public Object[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ Object[] ks = new TestClass[value.f0.length];
--- End diff --
Can we use plain objects here? Then we wouldn't need the TestClass at all.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105378932
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
--- End diff --
We can go for `Object[]`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105406325
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
--- End diff --
Well it may be more concise, but a try catch block makes it way more obvious that this part of the code throws an exception we care about.
try-catch blocks are also more commonin the codebase, and I'm wondering whether we want to introduce an inconsistency there.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105377789
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
--- End diff --
I think it is worth having it also here for users of the method. The more the places that the user can find the required information, the better.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105239687
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
--- End diff --
Let's include ```typeInfo``` in this exception as well to narrow it down for the user. We may even want to delay the exception until we scanned the entire type and report all invalid keys at once.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105380399
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
--- End diff --
The name is totally misleading, you are right. This test only verifies that we did not break anything and that even if you leave the default signature for the `keySelector` (returning `Object` although your key is of a different type) it still works.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105239060
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
--- End diff --
First time I'm seeing this pattern and i can't help but feel that a simple try-catch block would be more readable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105241562
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
--- End diff --
this commit will not be seen as it is.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105241450
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.setMaxParallelism(1);
+
+ DataStream<Integer> input = env.fromElements(new Integer(10000));
+
+ TypeInformation<?> expectedTypeInfo = IntegerTypeInfo.INT_TYPE_INFO;
+
+ KeyedStream<Integer, Integer> keyedStream = input.keyBy(new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ Assert.assertEquals(expectedTypeInfo, keyedStream.getKeyType());
+
+ keyedStream.addSink(new SinkFunction<Integer>() {
+ @Override
+ public void invoke(Integer value) throws Exception {
+ Assert.assertEquals(10000L, (long) value);
+ }
+ });
+
+ env.execute();
+ }
+
+ private static class TestClass {
+
+ private final int id;
+
+ TestClass(int id) {
+ this.id = id;
+ }
+ }
+
+ public static class POJOnoHashCode {
--- End diff --
the naming is also inconsistent with the other class; this class name doesn't contain "with".
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105407208
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
--- End diff --
The thing is that if I remember correctly we discussed about how to check for exceptions earlier this year. And if I remember correctly the outcome was sth along the lines of using JUnit facilities. Feel free to correct me on this one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3501: [FLINK-5874] Restrict key types in the DataStream API.
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/3501
Done! Let me know if you have any additional comments @zentol
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105378808
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
--- End diff --
This is the proper way of doing it. A `try catch` is of course an option, but this is more concise, and built in JUnit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105239223
--- Diff: flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java ---
@@ -736,7 +736,7 @@ public void restoreState(List<Integer> state) throws Exception {
static final ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>("seen", Boolean.class, false);
private static final long serialVersionUID = 1L;
- private ValueState<Boolean> operatorState;
+ private transient ValueState<Boolean> operatorState;
--- End diff --
unrelated change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105238457
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, TestClass[]>() {
+
+ @Override
+ public TestClass[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ TestClass[] ks = new TestClass[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = new TestClass(value.f0[i]);
+ }
+ return ks;
+ }
+ };
+
+ ObjectArrayTypeInfo<TestClass[], TestClass> keyTypeInfo = ObjectArrayTypeInfo.getInfoFor(
+ TestClass[].class, new GenericTypeInfo<>(TestClass.class));
+
+ testKeyRejection(keySelector, keyTypeInfo);
+ }
+
+ private <K> void testKeyRejection(KeySelector<Tuple2<Integer[], String>, K> keySelector, TypeInformation<K> expectedKeyType) {
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "barfoo")
+ );
+
+ Assert.assertEquals(expectedKeyType, TypeExtractor.getKeySelectorTypes(keySelector, input.getType()));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedKeyType + ") cannot be used as key.");
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : POJOs
+
+ @Test
+ public void testPOJONestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple1<int[]>>(
+ PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy("id");
+ }
+
+ @Test
+ public void testNestedArrayWorkArround() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOwithHashCode> input = env.fromElements(
+ new POJOwithHashCode(new int[] {1, 2}));
+
+ input.keyBy(new KeySelector<POJOwithHashCode, POJOwithHashCode>() {
+ @Override
+ public POJOwithHashCode getKey(POJOwithHashCode value) throws Exception {
+ return value;
+ }
+ }).addSink(new SinkFunction<POJOwithHashCode>() {
+ @Override
+ public void invoke(POJOwithHashCode value) throws Exception {
+ Assert.assertEquals(value.getId(), new int[]{1, 2});
+ }
+ });
+ }
+
+ @Test
+ public void testPOJOnoHashCodeKeyRejection() {
+
+ KeySelector<POJOnoHashCode, POJOnoHashCode> keySelector =
+ new KeySelector<POJOnoHashCode, POJOnoHashCode>() {
+ @Override
+ public POJOnoHashCode getKey(POJOnoHashCode value) throws Exception {
+ return value;
+ }
+ };
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<POJOnoHashCode> input = env.fromElements(
+ new POJOnoHashCode(new int[] {1, 2}));
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+
+ input.keyBy(keySelector);
+ }
+
+ // composite key tests : Tuples
+
+ @Test
+ public void testTupleNestedArrayKeyRejection() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer[], String>> input = env.fromElements(
+ new Tuple2<>(new Integer[] {1, 2}, "test-test"));
+
+ TypeInformation<?> expectedTypeInfo = new TupleTypeInfo<Tuple2<Integer[], String>>(
+ BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+ // adjust the rule
+ expectedException.expect(InvalidProgramException.class);
+ expectedException.expectMessage("This type (" + expectedTypeInfo + ") cannot be used as key.");
+
+ input.keyBy(new KeySelector<Tuple2<Integer[],String>, Tuple2<Integer[],String>>() {
+ @Override
+ public Tuple2<Integer[], String> getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value;
+ }
+ });
+ }
+
+ @Test
+ public void testPrimitiveKeyRejection() throws Exception {
--- End diff --
what do we actually verify here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105241140
--- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java ---
@@ -906,6 +919,256 @@ public void testChannelSelectors() {
}
/////////////////////////////////////////////////////////////
+ // KeyBy testing
+ /////////////////////////////////////////////////////////////
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ @Test
+ public void testPrimitiveArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, int[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, int[]>() {
+
+ @Override
+ public int[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ int[] ks = new int[value.f0.length];
+ for (int i = 0; i < ks.length; i++) {
+ ks[i] = value.f0[i];
+ }
+ return ks;
+ }
+ };
+
+ testKeyRejection(keySelector, PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testBasicArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, Integer[]> keySelector =
+ new KeySelector<Tuple2<Integer[], String>, Integer[]>() {
+
+ @Override
+ public Integer[] getKey(Tuple2<Integer[], String> value) throws Exception {
+ return value.f0;
+ }
+ };
+
+ testKeyRejection(keySelector, BasicArrayTypeInfo.INT_ARRAY_TYPE_INFO);
+ }
+
+ @Test
+ public void testObjectArrayKeyRejection() {
+
+ KeySelector<Tuple2<Integer[], String>, TestClass[]> keySelector =
--- End diff --
Do we need a separate class here or could we just throw in an Object[]?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105240510
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link #hashCode()} method and relies on
+ * the {@link Object#hashCode()} implementation.</li>
+ * <li>it is an array of any type (see {@link PrimitiveArrayTypeInfo}, {@link BasicArrayTypeInfo},
+ * {@link ObjectArrayTypeInfo}).</li>
+ * </ol>,
+ * {@code true} otherwise.
+ */
+ private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo) ?
--- End diff --
we should also simply break this if statement down into multiple blocks.
i.e.
```
if (type isntanceof PojoTypeInfo) {
return //find hashCode
}
if (tpye instance off XArrayTypeInfo ... ) {
return false;
}
return true;
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105377617
--- Diff: flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java ---
@@ -736,7 +736,7 @@ public void restoreState(List<Integer> state) throws Exception {
static final ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>("seen", Boolean.class, false);
private static final long serialVersionUID = 1L;
- private ValueState<Boolean> operatorState;
+ private transient ValueState<Boolean> operatorState;
--- End diff --
true but it is wrong and someone has to fix it at some point.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105405543
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link #hashCode()} method and relies on
+ * the {@link Object#hashCode()} implementation.</li>
+ * <li>it is an array of any type (see {@link PrimitiveArrayTypeInfo}, {@link BasicArrayTypeInfo},
+ * {@link ObjectArrayTypeInfo}).</li>
+ * </ol>,
+ * {@code true} otherwise.
+ */
+ private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo) ?
--- End diff --
So i looked through the code base, and found 5 different patterns, so you're right, "typical" doesn't really fit:
1:
```
<condition> ? <then> : <else>
```
2:
```
<condition> ?
<then> : <else>
```
3:
```
<condition> ?
<then>
: <else>
```
4:
```
<condition>
? <then>
: <else>
```
5:
```
if <condition> ?
<then> :
<else>
```
In my opinion, only 1 and 4 should be used; and 1 only for short statements. Both of these options are symmetrical; the if and else statement are formatted the same way, and place important information (? and \:) at the start of the line.
The point here is pattern recognition.
When seeing this
```
X
? Y
: Z
```
you can immediately see what is going on without looking at the details of the expression, nor scanning any line; the very first character gives it away. The position of ? and \: is also static, it's on the next line with an additional indent, compared to other options which may hide ? at the end of a 100 character expression (which gets especially funky with option 2 which also hides \: somewhere, so at a glace it could also just be a method call with many arguments).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105240608
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link #hashCode()} method and relies on
+ * the {@link Object#hashCode()} implementation.</li>
+ * <li>it is an array of any type (see {@link PrimitiveArrayTypeInfo}, {@link BasicArrayTypeInfo},
+ * {@link ObjectArrayTypeInfo}).</li>
+ * </ol>,
+ * {@code true} otherwise.
+ */
+ private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo) ?
+ !type.getTypeClass().getMethod("hashCode").getDeclaringClass().equals(Object.class) :
--- End diff --
this method would be more readable by not inverting here, and returning true as the default.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105377853
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java ---
@@ -114,9 +121,53 @@ public KeyedStream(DataStream<T> dataStream, KeySelector<T, KEY> keySelector, Ty
dataStream.getTransformation(),
new KeyGroupStreamPartitioner<>(keySelector, StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)));
this.keySelector = keySelector;
- this.keyType = keyType;
+ this.keyType = validateKeyType(keyType);
}
-
+
+ private TypeInformation<KEY> validateKeyType(TypeInformation<KEY> keyType) {
+ Stack<TypeInformation<?>> stack = new Stack<>();
+ stack.push(keyType);
+
+ while (!stack.isEmpty()) {
+ TypeInformation<?> typeInfo = stack.pop();
+
+ if (!validateKeyTypeIsHashable(typeInfo)) {
+ throw new InvalidProgramException("This type (" + keyType + ") cannot be used as key.");
+ }
+
+ if (typeInfo instanceof TupleTypeInfoBase) {
+ for (int i = 0; i < typeInfo.getArity(); i++) {
+ stack.push(((TupleTypeInfoBase) typeInfo).getTypeAt(i));
+ }
+ }
+ }
+ return keyType;
+ }
+
+ /**
+ * Validates that a given type of element (as encoded by the provided {@link TypeInformation}) can be
+ * used as a key in the {@code DataStream.keyBy()} operation.
+ *
+ * @return {@code false} if:
+ * <ol>
+ * <li>it is a POJO type but does not override the {@link #hashCode()} method and relies on
+ * the {@link Object#hashCode()} implementation.</li>
+ * <li>it is an array of any type (see {@link PrimitiveArrayTypeInfo}, {@link BasicArrayTypeInfo},
+ * {@link ObjectArrayTypeInfo}).</li>
+ * </ol>,
+ * {@code true} otherwise.
+ */
+ private boolean validateKeyTypeIsHashable(TypeInformation<?> type) {
+ try {
+ return (type instanceof PojoTypeInfo) ?
--- End diff --
Really? I am not sure about it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3501: [FLINK-5874] Restrict key types in the DataStream ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/3501#discussion_r105400311
--- Diff: flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java ---
@@ -736,7 +736,7 @@ public void restoreState(List<Integer> state) throws Exception {
static final ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>("seen", Boolean.class, false);
private static final long serialVersionUID = 1L;
- private ValueState<Boolean> operatorState;
+ private transient ValueState<Boolean> operatorState;
--- End diff --
well feel free to open a new PR or create a hotfix.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---