flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-4460) Side Outputs in Flink
Date Fri, 10 Mar 2017 14:18:05 GMT

    [ https://issues.apache.org/jira/browse/FLINK-4460?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15905146#comment-15905146
] 

ASF GitHub Bot commented on FLINK-4460:
---------------------------------------

Github user kl0u commented on a diff in the pull request:

    https://github.com/apache/flink/pull/3484#discussion_r105398702
  
    --- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
---
    @@ -333,32 +356,40 @@ public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID,
int ty
     				downStreamVertexID,
     				typeNumber,
     				null,
    -				new ArrayList<String>());
    +				new ArrayList<String>(), null);
     
     	}
     
     	private void addEdgeInternal(Integer upStreamVertexID,
     			Integer downStreamVertexID,
     			int typeNumber,
     			StreamPartitioner<?> partitioner,
    -			List<String> outputNames) {
    -
    +			List<String> outputNames,
    +			OutputTag outputTag) {
     
    -		if (virtualSelectNodes.containsKey(upStreamVertexID)) {
    +		if (virtualSideOutputNodes.containsKey(upStreamVertexID)) {
    +			int virtualId = upStreamVertexID;
    +			upStreamVertexID = virtualSideOutputNodes.get(virtualId).f0;
    +			if (outputTag == null) {
    +				outputTag = virtualSideOutputNodes.get(virtualId).f1;
    +			}
    +			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, null,
outputTag);
    +		} else if (virtualSelectNodes.containsKey(upStreamVertexID)) {
     			int virtualId = upStreamVertexID;
     			upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
     			if (outputNames.isEmpty()) {
     				// selections that happen downstream override earlier selections
     				outputNames = virtualSelectNodes.get(virtualId).f1;
     			}
    -			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);
    +			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames,
outputTag);
     		} else if (virtualPartitionNodes.containsKey(upStreamVertexID)) {
     			int virtualId = upStreamVertexID;
     			upStreamVertexID = virtualPartitionNodes.get(virtualId).f0;
     			if (partitioner == null) {
     				partitioner = virtualPartitionNodes.get(virtualId).f1;
     			}
    -			addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);
    +
    --- End diff --
    
    Remove line for uniformity with the `if() ...` above.


> Side Outputs in Flink
> ---------------------
>
>                 Key: FLINK-4460
>                 URL: https://issues.apache.org/jira/browse/FLINK-4460
>             Project: Flink
>          Issue Type: New Feature
>          Components: Core, DataStream API
>    Affects Versions: 1.2.0, 1.1.3
>            Reporter: Chen Qin
>            Assignee: Chen Qin
>              Labels: latearrivingevents, sideoutput
>
> https://docs.google.com/document/d/1vg1gpR8JL4dM07Yu4NyerQhhVvBlde5qdqnuJv4LcV4/edit?usp=sharing



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Mime
View raw message