You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by fhueske <gi...@git.apache.org> on 2018/03/21 21:52:57 UTC
[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...
GitHub user fhueske opened a pull request:
https://github.com/apache/flink/pull/5742
[FLINK-9031] Fix DataSet Union operator translation bug.
## What is the purpose of the change
- Fixes a bug reported in [FLINK-9031](https://issues.apache.org/jira/browse/FLINK-9031)
- Union nodes had a partitioning strategy on the outgoing channel, that was (intentionally) not translated by the `JobGraphGenerator` because the `JobGraphGenerator` assumed that Union nodes would always have outgoing FORWARD strategies.
- Not translating the partitioning resulted in an incorrect result because data was not correctly distributed.
## Brief change log
- Add a check in `JobGraphGenerator` to fail if a union node with non-FORWARD outgoing strategy is found.
- Add a pre-optimization plan traversal that fixes the strategy of union outputs to FORWARD.
- Add a test based on a simplified version of the reported program.
## Verifying this change
- Run the added test.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **n/a**
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/fhueske/flink dataSetUnionBug
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5742.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 #5742
----
commit 3939f06bacaf1595844de56fe58651732b89592e
Author: Fabian Hueske <fh...@...>
Date: 2018-03-21T19:54:05Z
[FLINK-9031] Fix DataSet Union operator translation bug.
- Adds a pass over the pre-optimized plan that fixes the output strategy of union nodes to FORWARD.
----
---
[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5742#discussion_r176340977
--- Diff: flink-optimizer/src/test/java/org/apache/flink/optimizer/testfunctions/IdentityFilter.java ---
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.testfunctions;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+
+/**
+ * FilterFunction for optimizer tests.
+ */
+public class IdentityFilter<T> implements FilterFunction<T> {
--- End diff --
Good point.
Didn't think about that as the other tests follow that pattern.
Will remove the class.
---
[GitHub] flink issue #5742: [FLINK-9031] Fix DataSet Union operator translation bug.
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/5742
The fix is still broken. Some tests are failing.
Need to look into it a bit more.
---
[GitHub] flink issue #5742: [FLINK-9031] Fix DataSet Union operator translation bug.
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/5742
Thanks for updating this.
The code looks good and the fix conceptually make perfect sense.
Merging this...
---
[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5742
---
[GitHub] flink issue #5742: [FLINK-9031] Fix DataSet Union operator translation bug.
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/5742
Updated the PR.
Tests are passing (mod an unrelated failure).
---
[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/5742#discussion_r176339259
--- Diff: flink-optimizer/src/test/java/org/apache/flink/optimizer/testfunctions/IdentityFilter.java ---
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.testfunctions;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+
+/**
+ * FilterFunction for optimizer tests.
+ */
+public class IdentityFilter<T> implements FilterFunction<T> {
--- End diff --
How about just using lambdas (filter always works) instead of adding this class?
---