[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/5742


---

[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...

[fhueske]

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 implements FilterFunction {
--- End diff --

Good point. 
Didn't think about that as the other tests follow that pattern.
Will remove the class.


---

[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...

[StephanEwen]

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 implements FilterFunction {
--- End diff --

How about just using lambdas (filter always works) instead of adding this 
class?


---

[GitHub] flink pull request #5742: [FLINK-9031] Fix DataSet Union operator translatio...

[fhueske]

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 
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.




---