Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
78f41e9c
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
78f41e9c
编写于
8月 31, 2014
作者:
S
sebastian kunert
提交者:
Fabian Hueske
1月 28, 2015
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-1328] Integrated forwarded Fields into the optimizer (incomplete)
This closes #83
上级
82c42002
变更
25
展开全部
隐藏空白更改
内联
并排
Showing
25 changed file
with
1460 addition
and
209 deletion
+1460
-209
flink-compiler/src/main/java/org/apache/flink/compiler/dag/AbstractPartialSolutionNode.java
...pache/flink/compiler/dag/AbstractPartialSolutionNode.java
+3
-2
flink-compiler/src/main/java/org/apache/flink/compiler/dag/BinaryUnionNode.java
...n/java/org/apache/flink/compiler/dag/BinaryUnionNode.java
+6
-2
flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
...java/org/apache/flink/compiler/dag/BulkIterationNode.java
+4
-3
flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
...main/java/org/apache/flink/compiler/dag/DataSinkNode.java
+5
-3
flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
...in/java/org/apache/flink/compiler/dag/DataSourceNode.java
+3
-2
flink-compiler/src/main/java/org/apache/flink/compiler/dag/FilterNode.java
...c/main/java/org/apache/flink/compiler/dag/FilterNode.java
+9
-3
flink-compiler/src/main/java/org/apache/flink/compiler/dag/OptimizerNode.java
...ain/java/org/apache/flink/compiler/dag/OptimizerNode.java
+9
-12
flink-compiler/src/main/java/org/apache/flink/compiler/dag/SingleInputNode.java
...n/java/org/apache/flink/compiler/dag/SingleInputNode.java
+7
-20
flink-compiler/src/main/java/org/apache/flink/compiler/dag/TwoInputNode.java
...main/java/org/apache/flink/compiler/dag/TwoInputNode.java
+9
-35
flink-compiler/src/main/java/org/apache/flink/compiler/dag/UnaryOperatorNode.java
...java/org/apache/flink/compiler/dag/UnaryOperatorNode.java
+8
-6
flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
...a/org/apache/flink/compiler/dag/WorksetIterationNode.java
+5
-4
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/GlobalProperties.java
...pache/flink/compiler/dataproperties/GlobalProperties.java
+70
-28
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/InterestingProperties.java
.../flink/compiler/dataproperties/InterestingProperties.java
+13
-4
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/LocalProperties.java
...apache/flink/compiler/dataproperties/LocalProperties.java
+57
-23
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/RequestedGlobalProperties.java
...nk/compiler/dataproperties/RequestedGlobalProperties.java
+46
-12
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/RequestedLocalProperties.java
...ink/compiler/dataproperties/RequestedLocalProperties.java
+29
-11
flink-compiler/src/main/java/org/apache/flink/compiler/postpass/GenericFlatTypePostPass.java
...ache/flink/compiler/postpass/GenericFlatTypePostPass.java
+4
-1
flink-compiler/src/test/java/org/apache/flink/compiler/DOPChangeTest.java
...rc/test/java/org/apache/flink/compiler/DOPChangeTest.java
+5
-2
flink-compiler/src/test/java/org/apache/flink/compiler/SemanticPropOptimizerTest.java
.../org/apache/flink/compiler/SemanticPropOptimizerTest.java
+941
-0
flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java
...ink/api/common/operators/DualInputSemanticProperties.java
+79
-4
flink-core/src/main/java/org/apache/flink/api/common/operators/Ordering.java
.../java/org/apache/flink/api/common/operators/Ordering.java
+1
-1
flink-core/src/main/java/org/apache/flink/api/common/operators/SemanticProperties.java
...apache/flink/api/common/operators/SemanticProperties.java
+15
-2
flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java
...k/api/common/operators/SingleInputSemanticProperties.java
+45
-2
flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
...apache/flink/examples/java/graph/ConnectedComponents.java
+2
-4
flink-java/src/main/java/org/apache/flink/api/java/record/functions/FunctionAnnotation.java
...e/flink/api/java/record/functions/FunctionAnnotation.java
+85
-23
未找到文件。
flink-compiler/src/main/java/org/apache/flink/compiler/dag/AbstractPartialSolutionNode.java
浏览文件 @
78f41e9c
...
...
@@ -24,6 +24,7 @@ import java.util.List;
import
java.util.Map
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.compiler.DataStatistics
;
import
org.apache.flink.compiler.costs.CostEstimator
;
import
org.apache.flink.compiler.plan.PlanNode
;
...
...
@@ -86,8 +87,8 @@ public abstract class AbstractPartialSolutionNode extends OptimizerNode {
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
false
;
public
SemanticProperties
getSemanticProperties
(
)
{
return
null
;
}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/BinaryUnionNode.java
浏览文件 @
78f41e9c
...
...
@@ -24,6 +24,8 @@ import java.util.HashSet;
import
java.util.List
;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.DualInputSemanticProperties
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.Union
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.DataStatistics
;
...
...
@@ -252,8 +254,10 @@ public class BinaryUnionNode extends TwoInputNode {
protected
void
readStubAnnotations
()
{}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
true
;
public
SemanticProperties
getSemanticProperties
()
{
DualInputSemanticProperties
sprops
=
new
DualInputSemanticProperties
();
sprops
.
setAllFieldsConstant
(
true
);
return
sprops
;
}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/BulkIterationNode.java
浏览文件 @
78f41e9c
...
...
@@ -24,6 +24,7 @@ import java.util.Iterator;
import
java.util.List
;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.base.BulkIterationBase
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.compiler.CompilerException
;
...
...
@@ -182,10 +183,10 @@ public class BulkIterationNode extends SingleInputNode implements IterationNode
public
String
getName
()
{
return
"Bulk Iteration"
;
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
false
;
public
SemanticProperties
getSemanticProperties
(
)
{
return
null
;
}
protected
void
readStubAnnotations
()
{}
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSinkNode.java
浏览文件 @
78f41e9c
...
...
@@ -28,6 +28,7 @@ import org.apache.flink.api.common.distributions.DataDistribution;
import
org.apache.flink.api.common.operators.GenericDataSinkBase
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.Ordering
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.DataStatistics
;
import
org.apache.flink.compiler.costs.CostEstimator
;
...
...
@@ -230,9 +231,10 @@ public class DataSinkNode extends OptimizerNode {
// --------------------------------------------------------------------------------------------
// Function Annotation Handling
// --------------------------------------------------------------------------------------------
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
false
;
@Override
public
SemanticProperties
getSemanticProperties
()
{
return
null
;
}
// --------------------------------------------------------------------------------------------
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/DataSourceNode.java
浏览文件 @
78f41e9c
...
...
@@ -30,6 +30,7 @@ import org.apache.flink.api.common.io.NonParallelInput;
import
org.apache.flink.api.common.io.statistics.BaseStatistics
;
import
org.apache.flink.api.common.operators.GenericDataSourceBase
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.compiler.DataStatistics
;
import
org.apache.flink.compiler.PactCompiler
;
import
org.apache.flink.compiler.costs.CostEstimator
;
...
...
@@ -193,8 +194,8 @@ public class DataSourceNode extends OptimizerNode {
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
false
;
public
SemanticProperties
getSemanticProperties
(
)
{
return
null
;
}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/FilterNode.java
浏览文件 @
78f41e9c
...
...
@@ -21,6 +21,8 @@ package org.apache.flink.compiler.dag;
import
java.util.Collections
;
import
java.util.List
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.SingleInputSemanticProperties
;
import
org.apache.flink.api.common.operators.base.FilterOperatorBase
;
import
org.apache.flink.compiler.DataStatistics
;
import
org.apache.flink.compiler.operators.FilterDescriptor
;
...
...
@@ -47,10 +49,14 @@ public class FilterNode extends SingleInputNode {
public
String
getName
()
{
return
"Filter"
;
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
true
;
public
SemanticProperties
getSemanticProperties
()
{
SingleInputSemanticProperties
sprops
=
new
SingleInputSemanticProperties
();
sprops
.
setAllFieldsConstant
(
true
);
return
sprops
;
}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/OptimizerNode.java
浏览文件 @
78f41e9c
...
...
@@ -31,6 +31,7 @@ import java.util.Set;
import
org.apache.flink.api.common.operators.AbstractUdfOperator
;
import
org.apache.flink.api.common.operators.CompilerHints
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.DataStatistics
;
...
...
@@ -268,16 +269,8 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
*/
@Override
public
abstract
void
accept
(
Visitor
<
OptimizerNode
>
visitor
);
/**
* Checks whether a field is modified by the user code or whether it is kept unchanged.
*
* @param input The input number.
* @param fieldNumber The position of the field.
*
* @return True if the field is not changed by the user function, false otherwise.
*/
public
abstract
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
);
public
abstract
SemanticProperties
getSemanticProperties
();
// ------------------------------------------------------------------------
// Getters / Setters
...
...
@@ -688,8 +681,12 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
return
null
;
}
for
(
int
keyColumn
:
keyColumns
)
{
if
(!
isFieldConstant
(
input
,
keyColumn
))
{
return
null
;
FieldSet
fs
=
getSemanticProperties
()
==
null
?
null
:
getSemanticProperties
().
getForwardFields
(
input
,
keyColumn
);
if
(
fs
==
null
)
{
return
null
;
}
else
if
(!
fs
.
contains
(
keyColumn
))
{
return
null
;
}
}
return
keyColumns
;
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/SingleInputNode.java
浏览文件 @
78f41e9c
...
...
@@ -31,8 +31,8 @@ import java.util.Map;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.SingleInputOperator
;
import
org.apache.flink.api.common.operators.SingleInputSemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.PactCompiler
;
...
...
@@ -144,22 +144,8 @@ public abstract class SingleInputNode extends OptimizerNode {
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
if
(
input
!=
0
)
{
throw
new
IndexOutOfBoundsException
();
}
SingleInputOperator
<?,
?,
?>
c
=
getPactContract
();
SingleInputSemanticProperties
semanticProperties
=
c
.
getSemanticProperties
();
if
(
semanticProperties
!=
null
)
{
FieldSet
fs
;
if
((
fs
=
semanticProperties
.
getForwardedField
(
fieldNumber
))
!=
null
)
{
return
fs
.
contains
(
fieldNumber
);
}
}
return
false
;
public
SemanticProperties
getSemanticProperties
()
{
return
((
SingleInputOperator
<?,?,?>)
getPactContract
()).
getSemanticProperties
();
}
...
...
@@ -444,10 +430,11 @@ public abstract class SingleInputNode extends OptimizerNode {
LocalProperties
lProps
=
in
.
getLocalProperties
().
clone
();
gProps
=
dps
.
computeGlobalProperties
(
gProps
);
lProps
=
dps
.
computeLocalProperties
(
lProps
);
SemanticProperties
props
=
this
.
getSemanticProperties
();
// filter by the user code field copies
gProps
=
gProps
.
filterBy
NodesConstantSet
(
thi
s
,
0
);
lProps
=
lProps
.
filterBy
NodesConstantSet
(
thi
s
,
0
);
gProps
=
gProps
.
filterBy
SemanticProperties
(
prop
s
,
0
);
lProps
=
lProps
.
filterBy
SemanticProperties
(
prop
s
,
0
);
// apply
node
.
initProperties
(
gProps
,
lProps
);
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/TwoInputNode.java
浏览文件 @
78f41e9c
...
...
@@ -31,10 +31,9 @@ import java.util.Map;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.DualInputOperator
;
import
org.apache.flink.api.common.operators.DualInputSemanticProperties
;
import
org.apache.flink.api.common.operators.Operator
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.PactCompiler
;
import
org.apache.flink.compiler.costs.CostEstimator
;
...
...
@@ -554,13 +553,14 @@ public abstract class TwoInputNode extends OptimizerNode {
DualInputPlanNode
node
=
operator
.
instantiate
(
in1
,
in2
,
this
);
node
.
setBroadcastInputs
(
broadcastChannelsCombination
);
GlobalProperties
gp1
=
in1
.
getGlobalProperties
().
clone
().
filterByNodesConstantSet
(
this
,
0
);
GlobalProperties
gp2
=
in2
.
getGlobalProperties
().
clone
().
filterByNodesConstantSet
(
this
,
1
);
SemanticProperties
props
=
this
.
getSemanticProperties
();
GlobalProperties
gp1
=
in1
.
getGlobalProperties
().
clone
().
filterBySemanticProperties
(
props
,
0
);
GlobalProperties
gp2
=
in2
.
getGlobalProperties
().
clone
().
filterBySemanticProperties
(
props
,
1
);
GlobalProperties
combined
=
operator
.
computeGlobalProperties
(
gp1
,
gp2
);
LocalProperties
lp1
=
in1
.
getLocalProperties
().
clone
().
filterBy
NodesConstantSet
(
thi
s
,
0
);
LocalProperties
lp2
=
in2
.
getLocalProperties
().
clone
().
filterBy
NodesConstantSet
(
thi
s
,
1
);
LocalProperties
lp1
=
in1
.
getLocalProperties
().
clone
().
filterBy
SemanticProperties
(
prop
s
,
0
);
LocalProperties
lp2
=
in2
.
getLocalProperties
().
clone
().
filterBy
SemanticProperties
(
prop
s
,
1
);
LocalProperties
locals
=
operator
.
computeLocalProperties
(
lp1
,
lp2
);
node
.
initProperties
(
combined
,
locals
);
...
...
@@ -690,37 +690,11 @@ public abstract class TwoInputNode extends OptimizerNode {
}
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
DualInputOperator
<?,
?,
?,
?>
c
=
getPactContract
();
DualInputSemanticProperties
semanticProperties
=
c
.
getSemanticProperties
();
switch
(
input
)
{
case
0
:
if
(
semanticProperties
!=
null
)
{
FieldSet
fs
;
if
((
fs
=
semanticProperties
.
getForwardedField1
(
fieldNumber
))
!=
null
)
{
return
fs
.
contains
(
fieldNumber
);
}
}
break
;
case
1
:
if
(
semanticProperties
!=
null
)
{
FieldSet
fs
;
if
((
fs
=
semanticProperties
.
getForwardedField2
(
fieldNumber
))
!=
null
)
{
return
fs
.
contains
(
fieldNumber
);
}
}
break
;
default
:
throw
new
IndexOutOfBoundsException
();
}
return
false
;
public
SemanticProperties
getSemanticProperties
()
{
return
((
DualInputOperator
<?,
?,
?,
?>)
getPactContract
()).
getSemanticProperties
();
}
// --------------------------------------------------------------------------------------------
// Miscellaneous
// --------------------------------------------------------------------------------------------
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/UnaryOperatorNode.java
浏览文件 @
78f41e9c
...
...
@@ -21,6 +21,8 @@ package org.apache.flink.compiler.dag;
import
java.util.Arrays
;
import
java.util.List
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.SingleInputSemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.DataStatistics
;
import
org.apache.flink.compiler.operators.OperatorDescriptorSingle
;
...
...
@@ -54,12 +56,12 @@ public class UnaryOperatorNode extends SingleInputNode {
public
String
getName
()
{
return
this
.
name
;
}
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
if
(
input
!=
0
)
{
throw
new
IndexOutOfBoundsException
();
}
return
true
;
@Override
public
SemanticProperties
getSemanticProperties
(
)
{
SingleInputSemanticProperties
sprops
=
new
SingleInputSemanticProperties
();
sprops
.
setAllFieldsConstant
(
true
);
return
sprops
;
}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dag/WorksetIterationNode.java
浏览文件 @
78f41e9c
...
...
@@ -24,6 +24,7 @@ import java.util.Iterator;
import
java.util.List
;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.base.DeltaIterationBase
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.typeinfo.NothingTypeInfo
;
...
...
@@ -218,12 +219,12 @@ public class WorksetIterationNode extends TwoInputNode implements IterationNode
public
String
getName
()
{
return
"Workset Iteration"
;
}
@Override
public
boolean
isFieldConstant
(
int
input
,
int
fieldNumber
)
{
return
false
;
public
SemanticProperties
getSemanticProperties
(
)
{
return
null
;
}
protected
void
readStubAnnotations
()
{}
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/GlobalProperties.java
浏览文件 @
78f41e9c
...
...
@@ -25,10 +25,10 @@ import java.util.Set;
import
org.apache.flink.api.common.functions.Partitioner
;
import
org.apache.flink.api.common.operators.Order
;
import
org.apache.flink.api.common.operators.Ordering
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.dag.OptimizerNode
;
import
org.apache.flink.compiler.plan.Channel
;
import
org.apache.flink.compiler.util.Utils
;
import
org.apache.flink.runtime.operators.shipping.ShipStrategyType
;
...
...
@@ -212,7 +212,19 @@ public class GlobalProperties implements Cloneable {
return
false
;
}
}
public
Ordering
getOrdering
()
{
return
this
.
ordering
;
}
public
void
setOrdering
(
Ordering
ordering
)
{
this
.
ordering
=
ordering
;
}
public
void
setPartitioningFields
(
FieldList
partitioningFields
)
{
this
.
partitioningFields
=
partitioningFields
;
}
public
boolean
isFullyReplicated
()
{
return
this
.
partitioning
==
PartitioningProperty
.
FULL_REPLICATION
;
}
...
...
@@ -234,56 +246,86 @@ public class GlobalProperties implements Cloneable {
}
/**
* Filters these
properties by what can be preserved through the given output contract
.
*
* @param
node The optimizer node
.
* @param input The in
put of the node to filter agains
t.
* @return The
adjusted global properties.
* Filters these
GlobalProperties by the fields that are constant or forwarded to another output field
.
*
* @param
props The node representing the contract
.
* @param input The in
dex of the inpu
t.
* @return The
filtered GlobalProperties
*/
public
GlobalProperties
filterBy
NodesConstantSet
(
OptimizerNode
node
,
int
input
)
{
public
GlobalProperties
filterBy
SemanticProperties
(
SemanticProperties
props
,
int
input
)
{
// check if partitioning survives
FieldList
forwardFields
=
null
;
GlobalProperties
returnProps
=
this
;
if
(
props
==
null
)
{
return
new
GlobalProperties
();
}
if
(
this
.
ordering
!=
null
)
{
for
(
int
col
:
this
.
ordering
.
getInvolvedIndexes
())
{
if
(!
node
.
isFieldConstant
(
input
,
col
))
{
return
new
GlobalProperties
();
Ordering
no
=
new
Ordering
();
for
(
int
index
:
this
.
ordering
.
getInvolvedIndexes
())
{
forwardFields
=
props
.
getForwardFields
(
input
,
index
)
==
null
?
null
:
props
.
getForwardFields
(
input
,
index
).
toFieldList
();
if
(
forwardFields
==
null
)
{
returnProps
=
new
GlobalProperties
();
no
=
null
;
break
;
}
else
{
returnProps
=
returnProps
==
this
?
this
.
clone
()
:
returnProps
;
for
(
int
i
=
0
;
i
<
forwardFields
.
size
();
i
++)
{
no
.
appendOrdering
(
forwardFields
.
get
(
i
),
this
.
ordering
.
getType
(
index
),
this
.
ordering
.
getOrder
(
index
));
}
}
returnProps
.
setOrdering
(
no
);
}
}
if
(
this
.
partitioningFields
!=
null
)
{
for
(
int
colIndex
:
this
.
partitioningFields
)
{
if
(!
node
.
isFieldConstant
(
input
,
colIndex
))
{
return
new
GlobalProperties
();
returnProps
=
returnProps
==
this
?
this
.
clone
()
:
returnProps
;
returnProps
.
setPartitioningFields
(
new
FieldList
());
for
(
int
index
:
this
.
partitioningFields
)
{
forwardFields
=
props
.
getForwardFields
(
input
,
index
)
==
null
?
null
:
props
.
getForwardFields
(
input
,
index
).
toFieldList
();
if
(
forwardFields
==
null
)
{
returnProps
=
new
GlobalProperties
();
break
;
}
else
{
returnProps
.
setPartitioningFields
(
returnProps
.
getPartitioningFields
().
addFields
(
forwardFields
));
}
}
}
if
(
this
.
uniqueFieldCombinations
!=
null
)
{
HashSet
<
FieldSet
>
newSet
=
new
HashSet
<
FieldSet
>();
newSet
.
addAll
(
this
.
uniqueFieldCombinations
);
for
(
Iterator
<
FieldSet
>
combos
=
newSet
.
iterator
();
combos
.
hasNext
();
){
for
(
Iterator
<
FieldSet
>
combos
=
this
.
uniqueFieldCombinations
.
iterator
();
combos
.
hasNext
();
){
FieldSet
current
=
combos
.
next
();
FieldSet
nfs
=
new
FieldSet
();
for
(
Integer
field
:
current
)
{
if
(!
node
.
isFieldConstant
(
input
,
field
))
{
combos
.
remove
();
if
(
props
.
getForwardFields
(
input
,
field
)
==
null
)
{
newSet
.
remove
(
current
);
nfs
=
null
;
break
;
}
else
{
nfs
=
nfs
.
addFields
(
props
.
getForwardFields
(
input
,
field
));
}
}
if
(
nfs
!=
null
)
{
newSet
.
remove
(
current
);
newSet
.
add
(
nfs
);
}
}
if
(
newSet
.
size
()
!=
this
.
uniqueFieldCombinations
.
size
())
{
GlobalProperties
gp
=
clone
();
gp
.
uniqueFieldCombinations
=
newSet
.
isEmpty
()
?
null
:
newSet
;
return
gp
;
}
GlobalProperties
gp
=
returnProps
.
clone
();
gp
.
uniqueFieldCombinations
=
newSet
.
isEmpty
()
?
null
:
newSet
;
return
gp
;
}
if
(
this
.
partitioning
==
PartitioningProperty
.
FULL_REPLICATION
)
{
return
new
GlobalProperties
();
}
return
thi
s
;
return
returnProp
s
;
}
public
void
parameterizeChannel
(
Channel
channel
,
boolean
globalDopChange
)
{
switch
(
this
.
partitioning
)
{
case
RANDOM:
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/InterestingProperties.java
浏览文件 @
78f41e9c
...
...
@@ -23,7 +23,10 @@ import java.util.HashSet;
import
java.util.Iterator
;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.compiler.dag.OptimizerNode
;
import
org.apache.flink.compiler.dag.SingleInputNode
;
import
org.apache.flink.compiler.dag.TwoInputNode
;
/**
* The interesting properties that a node in the optimizer plan hands to its predecessors. It has the
...
...
@@ -87,19 +90,25 @@ public class InterestingProperties implements Cloneable
public
Set
<
RequestedGlobalProperties
>
getGlobalProperties
()
{
return
this
.
globalProps
;
}
public
InterestingProperties
filterByCodeAnnotations
(
OptimizerNode
node
,
int
input
)
{
InterestingProperties
iProps
=
new
InterestingProperties
();
SemanticProperties
props
=
null
;
if
(
node
instanceof
SingleInputNode
)
{
props
=
node
.
getSemanticProperties
();
}
else
if
(
node
instanceof
TwoInputNode
)
{
props
=
node
.
getSemanticProperties
();
}
for
(
RequestedGlobalProperties
rgp
:
this
.
globalProps
)
{
RequestedGlobalProperties
filtered
=
rgp
.
filterBy
NodesConstantSet
(
node
,
input
);
RequestedGlobalProperties
filtered
=
rgp
.
filterBy
SemanticProperties
(
props
,
input
);
if
(
filtered
!=
null
&&
!
filtered
.
isTrivial
())
{
iProps
.
addGlobalProperties
(
filtered
);
}
}
for
(
RequestedLocalProperties
rlp
:
this
.
localProps
)
{
RequestedLocalProperties
filtered
=
rlp
.
filterBy
NodesConstantSet
(
node
,
input
);
RequestedLocalProperties
filtered
=
rlp
.
filterBy
SemanticProperties
(
props
,
input
);
if
(
filtered
!=
null
&&
!
filtered
.
isTrivial
())
{
iProps
.
addLocalProperties
(
filtered
);
}
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/LocalProperties.java
浏览文件 @
78f41e9c
...
...
@@ -20,12 +20,13 @@
package
org.apache.flink.compiler.dataproperties
;
import
java.util.HashSet
;
import
java.util.Iterator
;
import
java.util.Set
;
import
org.apache.flink.api.common.operators.Ordering
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.dag.OptimizerNode
;
/**
* This class represents local properties of the data. A local property is a property that exists
...
...
@@ -126,57 +127,91 @@ public class LocalProperties implements Cloneable {
/**
* Filters these properties by what can be preserved through a user function's constant fields set.
*
* @param
node
The optimizer node that potentially modifies the properties.
*
* @param
props
The optimizer node that potentially modifies the properties.
* @param input The input of the node which is relevant.
*
* @return T
rue, if the resulting properties are non trivial.
*
* @return T
he filtered LocalProperties
*/
public
LocalProperties
filterBy
NodesConstantSet
(
OptimizerNode
node
,
int
input
)
{
public
LocalProperties
filterBy
SemanticProperties
(
SemanticProperties
props
,
int
input
)
{
// check, whether the local order is preserved
Ordering
no
=
this
.
ordering
;
FieldList
ngf
=
this
.
groupedFields
;
Set
<
FieldSet
>
nuf
=
this
.
uniqueFields
;
FieldList
forwardList
=
null
;
if
(
props
==
null
)
{
return
new
LocalProperties
();
}
if
(
this
.
ordering
!=
null
)
{
no
=
new
Ordering
();
final
FieldList
involvedIndexes
=
this
.
ordering
.
getInvolvedIndexes
();
for
(
int
i
=
0
;
i
<
involvedIndexes
.
size
();
i
++)
{
if
(!
node
.
isFieldConstant
(
input
,
involvedIndexes
.
get
(
i
)))
{
if
(
i
==
0
)
{
forwardList
=
props
.
getForwardFields
(
input
,
involvedIndexes
.
get
(
i
))
==
null
?
null
:
props
.
getForwardFields
(
input
,
involvedIndexes
.
get
(
i
)).
toFieldList
();
if
(
forwardList
==
null
)
{
no
=
null
;
ngf
=
null
;
/*if (i == 0) {
no = null;
ngf = null;
} else {
no = this.ordering.createNewOrderingUpToIndex(i);
ngf = no.getInvolvedIndexes();
}
}
*/
break
;
}
else
{
no
.
appendOrdering
(
forwardList
.
get
(
0
),
this
.
ordering
.
getType
(
i
),
this
.
ordering
.
getOrder
(
i
));
ngf
=
no
.
getInvolvedIndexes
();
}
}
}
else
if
(
this
.
groupedFields
!=
null
)
{
// check, whether the local key grouping is preserved
for
(
Integer
index
:
this
.
groupedFields
)
{
if
(!
node
.
isFieldConstant
(
input
,
index
))
{
forwardList
=
props
.
getForwardFields
(
input
,
index
)
==
null
?
null
:
props
.
getForwardFields
(
input
,
index
).
toFieldList
();
if
(
forwardList
==
null
)
{
ngf
=
null
;
break
;
}
else
if
(!
forwardList
.
contains
(
index
))
{
FieldList
grouped
=
new
FieldList
();
for
(
Integer
value
:
ngf
.
toFieldList
())
{
if
(
value
.
intValue
()
==
index
)
{
grouped
=
grouped
.
addFields
(
forwardList
);
}
else
{
grouped
=
grouped
.
addField
(
value
);
}
}
ngf
=
grouped
;
}
}
}
if
(
this
.
uniqueFields
!=
null
&&
this
.
uniqueFields
.
size
()
>
0
)
{
Set
<
FieldSet
>
s
=
new
HashSet
<
FieldSet
>(
this
.
uniqueFields
);
for
(
FieldSet
fields
:
this
.
uniqueFields
)
{
for
(
Integer
index
:
fields
)
{
if
(!
node
.
isFieldConstant
(
input
,
index
))
{
s
.
remove
(
fields
);
if
(
this
.
uniqueFields
!=
null
)
{
HashSet
<
FieldSet
>
newSet
=
new
HashSet
<
FieldSet
>();
newSet
.
addAll
(
this
.
uniqueFields
);
for
(
Iterator
<
FieldSet
>
combos
=
this
.
uniqueFields
.
iterator
();
combos
.
hasNext
();
){
FieldSet
current
=
combos
.
next
();
FieldSet
nfs
=
new
FieldSet
();
for
(
Integer
field
:
current
)
{
if
(
props
.
getForwardFields
(
input
,
field
)
==
null
)
{
newSet
.
remove
(
current
);
nfs
=
null
;
break
;
}
else
{
nfs
=
nfs
.
addFields
(
props
.
getForwardFields
(
input
,
field
));
}
}
if
(
nfs
!=
null
)
{
newSet
.
remove
(
current
);
newSet
.
add
(
nfs
);
}
}
if
(
s
.
size
()
!=
this
.
uniqueFields
.
size
())
{
nuf
=
s
;
}
nuf
=
newSet
.
isEmpty
()
?
null
:
newSet
;
}
if
(
no
==
this
.
ordering
&&
ngf
==
this
.
groupedFields
&&
nuf
==
this
.
uniqueFields
)
{
return
this
;
}
else
{
...
...
@@ -187,7 +222,6 @@ public class LocalProperties implements Cloneable {
return
lp
;
}
}
// --------------------------------------------------------------------------------------------
@Override
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/RequestedGlobalProperties.java
浏览文件 @
78f41e9c
...
...
@@ -21,9 +21,10 @@ package org.apache.flink.compiler.dataproperties;
import
org.apache.flink.api.common.distributions.DataDistribution
;
import
org.apache.flink.api.common.functions.Partitioner
;
import
org.apache.flink.api.common.operators.Ordering
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.CompilerException
;
import
org.apache.flink.compiler.dag.OptimizerNode
;
import
org.apache.flink.compiler.plan.Channel
;
import
org.apache.flink.compiler.util.Utils
;
import
org.apache.flink.runtime.operators.shipping.ShipStrategyType
;
...
...
@@ -188,30 +189,63 @@ public final class RequestedGlobalProperties implements Cloneable {
this
.
customPartitioner
=
null
;
}
public
void
setPartitioningFields
(
FieldSet
partitioned
)
{
this
.
partitioningFields
=
partitioned
;
}
public
void
setPartitioningFields
(
FieldSet
fields
,
PartitioningProperty
partitioning
)
{
this
.
partitioningFields
=
fields
;
this
.
partitioning
=
partitioning
;
}
public
void
setOrdering
(
Ordering
newOrdering
)
{
this
.
ordering
=
newOrdering
;
}
/**
* Filters these properties by what can be preserved by the given node when propagated down
* to the given input.
*
* @param
node
The node representing the contract.
*
* @param
props
The node representing the contract.
* @param input The index of the input.
* @return T
rue, if any non-default value is preserved, false otherwise.
* @return T
he filtered RequestedGlobalProperties
*/
public
RequestedGlobalProperties
filterByNodesConstantSet
(
OptimizerNode
node
,
int
input
)
{
public
RequestedGlobalProperties
filterBySemanticProperties
(
SemanticProperties
props
,
int
input
)
{
FieldList
sourceList
;
RequestedGlobalProperties
returnProps
=
null
;
if
(
props
==
null
)
{
return
null
;
}
// check if partitioning survives
if
(
this
.
ordering
!=
null
)
{
for
(
int
col
:
this
.
ordering
.
getInvolvedIndexes
())
{
if
(!
node
.
isFieldConstant
(
input
,
col
))
{
Ordering
no
=
new
Ordering
();
returnProps
=
new
RequestedGlobalProperties
();
returnProps
.
setPartitioningFields
(
new
FieldSet
(),
this
.
partitioning
);
for
(
int
index
=
0
;
index
<
this
.
ordering
.
getInvolvedIndexes
().
size
();
index
++)
{
int
value
=
this
.
ordering
.
getInvolvedIndexes
().
get
(
index
);
sourceList
=
props
.
getSourceField
(
input
,
value
)
==
null
?
null
:
props
.
getSourceField
(
input
,
value
).
toFieldList
();
if
(
sourceList
!=
null
)
{
no
.
appendOrdering
(
sourceList
.
get
(
0
),
this
.
ordering
.
getType
(
index
),
this
.
ordering
.
getOrder
(
index
));
}
else
{
return
null
;
}
}
returnProps
.
setOrdering
(
no
);
}
else
if
(
this
.
partitioningFields
!=
null
)
{
for
(
int
colIndex
:
this
.
partitioningFields
)
{
if
(!
node
.
isFieldConstant
(
input
,
colIndex
))
{
returnProps
=
new
RequestedGlobalProperties
();
returnProps
.
setPartitioningFields
(
new
FieldSet
(),
this
.
partitioning
);
for
(
Integer
index
:
this
.
partitioningFields
)
{
sourceList
=
props
.
getSourceField
(
input
,
index
)
==
null
?
null
:
props
.
getSourceField
(
input
,
index
).
toFieldList
();
if
(
sourceList
!=
null
)
{
returnProps
.
setPartitioningFields
(
returnProps
.
getPartitionedFields
().
addFields
(
sourceList
),
this
.
partitioning
);
}
else
{
return
null
;
}
}
}
// make sure that certain properties are not pushed down
final
PartitioningProperty
partitioning
=
this
.
partitioning
;
if
(
partitioning
==
PartitioningProperty
.
FULL_REPLICATION
||
...
...
@@ -220,8 +254,8 @@ public final class RequestedGlobalProperties implements Cloneable {
{
return
null
;
}
return
thi
s
;
return
returnProp
s
;
}
/**
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/dataproperties/RequestedLocalProperties.java
浏览文件 @
78f41e9c
...
...
@@ -22,9 +22,9 @@ package org.apache.flink.compiler.dataproperties;
import
java.util.Arrays
;
import
org.apache.flink.api.common.operators.Ordering
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.operators.util.FieldSet
;
import
org.apache.flink.compiler.dag.OptimizerNode
;
import
org.apache.flink.compiler.plan.Channel
;
import
org.apache.flink.compiler.util.Utils
;
import
org.apache.flink.runtime.operators.util.LocalStrategy
;
...
...
@@ -138,29 +138,47 @@ public class RequestedLocalProperties implements Cloneable {
* Filters these properties by what can be preserved through a user function's constant fields set.
* Since interesting properties are filtered top-down, anything that partially destroys the ordering
* makes the properties uninteresting.
*
* @param
node
The optimizer node that potentially modifies the properties.
*
* @param
props
The optimizer node that potentially modifies the properties.
* @param input The input of the node which is relevant.
*
* @return T
rue, if the resulting properties are non trivial.
*
* @return T
he filtered RequestedLocalProperties
*/
public
RequestedLocalProperties
filterByNodesConstantSet
(
OptimizerNode
node
,
int
input
)
{
public
RequestedLocalProperties
filterBySemanticProperties
(
SemanticProperties
props
,
int
input
)
{
FieldList
sourceList
;
RequestedLocalProperties
returnProps
=
this
;
if
(
props
==
null
)
{
return
null
;
}
if
(
this
.
ordering
!=
null
)
{
final
FieldList
involvedIndexes
=
this
.
ordering
.
getInvolvedIndexes
();
for
(
int
i
=
0
;
i
<
involvedIndexes
.
size
();
i
++)
{
if
(!
node
.
isFieldConstant
(
input
,
involvedIndexes
.
get
(
i
)))
{
Ordering
no
=
new
Ordering
();
returnProps
=
this
.
clone
();
for
(
int
index
=
0
;
index
<
this
.
ordering
.
getInvolvedIndexes
().
size
();
index
++)
{
int
value
=
this
.
ordering
.
getInvolvedIndexes
().
get
(
index
);
sourceList
=
props
.
getSourceField
(
input
,
value
)
==
null
?
null
:
props
.
getSourceField
(
input
,
value
).
toFieldList
();
if
(
sourceList
!=
null
)
{
no
.
appendOrdering
(
sourceList
.
get
(
0
),
this
.
ordering
.
getType
(
index
),
this
.
ordering
.
getOrder
(
index
));
}
else
{
return
null
;
}
}
returnProps
.
setOrdering
(
no
);
}
else
if
(
this
.
groupedFields
!=
null
)
{
returnProps
=
this
.
clone
();
returnProps
.
setGroupedFields
(
new
FieldList
());
// check, whether the local key grouping is preserved
for
(
Integer
index
:
this
.
groupedFields
)
{
if
(!
node
.
isFieldConstant
(
input
,
index
))
{
sourceList
=
props
.
getSourceField
(
input
,
index
)
==
null
?
null
:
props
.
getSourceField
(
input
,
index
).
toFieldList
();
if
(
sourceList
!=
null
)
{
returnProps
.
setGroupedFields
(
returnProps
.
getGroupedFields
().
addFields
(
sourceList
));
}
else
{
return
null
;
}
}
}
return
thi
s
;
return
returnProp
s
;
}
/**
...
...
flink-compiler/src/main/java/org/apache/flink/compiler/postpass/GenericFlatTypePostPass.java
浏览文件 @
78f41e9c
...
...
@@ -21,6 +21,7 @@ package org.apache.flink.compiler.postpass;
import
java.util.Map
;
import
org.apache.flink.api.common.operators.SemanticProperties
;
import
org.apache.flink.api.common.operators.util.FieldList
;
import
org.apache.flink.api.common.typeutils.TypeComparatorFactory
;
import
org.apache.flink.api.common.typeutils.TypePairComparatorFactory
;
...
...
@@ -522,7 +523,9 @@ public abstract class GenericFlatTypePostPass<X, T extends AbstractSchema<X>> im
try
{
for
(
Map
.
Entry
<
Integer
,
X
>
entry
:
sourceSchema
)
{
Integer
pos
=
entry
.
getKey
();
if
(
optNode
.
isFieldConstant
(
input
,
pos
))
{
SemanticProperties
sprops
=
optNode
.
getSemanticProperties
();
if
(
sprops
!=
null
&&
sprops
.
getForwardFields
(
input
,
pos
)
!=
null
&&
sprops
.
getForwardFields
(
input
,
pos
).
contains
(
pos
))
{
targetSchema
.
addType
(
pos
,
entry
.
getValue
());
}
}
...
...
flink-compiler/src/test/java/org/apache/flink/compiler/DOPChangeTest.java
浏览文件 @
78f41e9c
...
...
@@ -257,8 +257,11 @@ public class DOPChangeTest extends CompilerTestBase {
// mapper respectively reducer
SinkPlanNode
sinkNode
=
oPlan
.
getDataSinks
().
iterator
().
next
();
SingleInputPlanNode
red2Node
=
(
SingleInputPlanNode
)
sinkNode
.
getPredecessor
();
Assert
.
assertEquals
(
"The Reduce 2 Node has an invalid local strategy."
,
LocalStrategy
.
SORT
,
red2Node
.
getInput
().
getLocalStrategy
());
SingleInputPlanNode
map2Node
=
(
SingleInputPlanNode
)
red2Node
.
getPredecessor
();
Assert
.
assertEquals
(
"The Reduce 2 Node has an invalid local strategy."
,
LocalStrategy
.
NONE
,
red2Node
.
getInput
().
getLocalStrategy
());
Assert
.
assertEquals
(
"The Map 2 Node has an invalid local strategy."
,
LocalStrategy
.
SORT
,
map2Node
.
getInput
().
getLocalStrategy
());
}
/**
...
...
flink-compiler/src/test/java/org/apache/flink/compiler/SemanticPropOptimizerTest.java
0 → 100644
浏览文件 @
78f41e9c
此差异已折叠。
点击以展开。
flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java
浏览文件 @
78f41e9c
...
...
@@ -58,7 +58,40 @@ public class DualInputSemanticProperties extends SemanticProperties {
public
DualInputSemanticProperties
()
{
init
();
}
/**
* Finds the source field where the given field was forwarded from.
* @param dest The destination field in the output data.
* @return FieldSet containing the source input fields.
*/
public
FieldSet
forwardedFrom1
(
int
dest
)
{
FieldSet
fs
=
null
;
for
(
Map
.
Entry
<
Integer
,
FieldSet
>
entry
:
forwardedFields1
.
entrySet
())
{
if
(
entry
.
getValue
().
contains
(
dest
))
{
if
(
fs
==
null
)
{
fs
=
new
FieldSet
();
}
fs
=
fs
.
addField
(
entry
.
getKey
());
}
}
return
fs
;
}
public
FieldSet
forwardedFrom2
(
int
dest
)
{
FieldSet
fs
=
null
;
for
(
Map
.
Entry
<
Integer
,
FieldSet
>
entry
:
forwardedFields2
.
entrySet
())
{
if
(
entry
.
getValue
().
contains
(
dest
))
{
if
(
fs
==
null
)
{
fs
=
new
FieldSet
();
}
fs
=
fs
.
addField
(
entry
.
getKey
());
}
}
return
fs
;
}
/**
* Adds, to the existing information, a field that is forwarded directly
* from the source record(s) in the first input to the destination
...
...
@@ -115,6 +148,10 @@ public class DualInputSemanticProperties extends SemanticProperties {
* @return the destination fields, or null if they do not exist
*/
public
FieldSet
getForwardedField1
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
return
this
.
forwardedFields1
.
get
(
sourceField
);
}
...
...
@@ -174,9 +211,43 @@ public class DualInputSemanticProperties extends SemanticProperties {
* @return the destination fields, or null if they do not exist
*/
public
FieldSet
getForwardedField2
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
return
this
.
forwardedFields2
.
get
(
sourceField
);
}
@Override
public
FieldSet
getSourceField
(
int
input
,
int
field
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
field
);
}
switch
(
input
)
{
case
0
:
return
this
.
forwardedFrom1
(
field
);
case
1
:
return
this
.
forwardedFrom2
(
field
);
default
:
throw
new
IndexOutOfBoundsException
();
}
}
@Override
public
FieldSet
getForwardFields
(
int
input
,
int
field
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
field
);
}
if
(
input
==
0
)
{
return
this
.
getForwardedField1
(
field
);
}
else
if
(
input
==
1
)
{
return
this
.
getForwardedField2
(
field
);
}
return
null
;
}
/**
* Adds, to the existing information, field(s) that are read in
* the source record(s) from the first input.
...
...
@@ -253,7 +324,7 @@ public class DualInputSemanticProperties extends SemanticProperties {
super
.
clearProperties
();
init
();
}
@Override
public
boolean
isEmpty
()
{
return
super
.
isEmpty
()
&&
...
...
@@ -263,7 +334,11 @@ public class DualInputSemanticProperties extends SemanticProperties {
(
readFields2
==
null
||
readFields2
.
size
()
==
0
);
}
@Override
public
String
toString
()
{
return
"DISP("
+
this
.
forwardedFields1
+
"; "
+
this
.
forwardedFields2
+
")"
;
}
private
void
init
()
{
this
.
forwardedFields1
=
new
HashMap
<
Integer
,
FieldSet
>();
this
.
forwardedFields2
=
new
HashMap
<
Integer
,
FieldSet
>();
...
...
flink-core/src/main/java/org/apache/flink/api/common/operators/Ordering.java
浏览文件 @
78f41e9c
...
...
@@ -144,7 +144,7 @@ public class Ordering {
}
for
(
int
i
=
0
;
i
<
this
.
indexes
.
size
();
i
++)
{
if
(
this
.
indexes
.
get
(
i
)
!=
otherOrdering
.
indexes
.
get
(
i
))
{
if
(
this
.
indexes
.
get
(
i
)
.
intValue
()
!=
otherOrdering
.
indexes
.
get
(
i
).
intValue
(
))
{
return
false
;
}
...
...
flink-core/src/main/java/org/apache/flink/api/common/operators/SemanticProperties.java
浏览文件 @
78f41e9c
...
...
@@ -27,7 +27,8 @@ import org.apache.flink.api.common.operators.util.FieldSet;
* Container for the semantic properties associated to an operator.
*/
public
abstract
class
SemanticProperties
implements
Serializable
{
private
boolean
allFieldsConstant
;
private
static
final
long
serialVersionUID
=
1L
;
/** Set of fields that are written in the destination record(s).*/
...
...
@@ -47,7 +48,19 @@ public abstract class SemanticProperties implements Serializable {
this
.
writtenFields
=
this
.
writtenFields
.
addFields
(
writtenFields
);
}
}
public
void
setAllFieldsConstant
(
boolean
constant
)
{
this
.
allFieldsConstant
=
constant
;
}
public
boolean
isAllFieldsConstant
()
{
return
this
.
allFieldsConstant
;
}
public
abstract
FieldSet
getForwardFields
(
int
input
,
int
field
);
public
abstract
FieldSet
getSourceField
(
int
input
,
int
field
);
/**
* Sets the field(s) that are written in the destination record(s).
*
...
...
flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java
浏览文件 @
78f41e9c
...
...
@@ -36,7 +36,41 @@ public class SingleInputSemanticProperties extends SemanticProperties {
/** Set of fields that are read in the source record(s).*/
private
FieldSet
readFields
;
@Override
public
FieldSet
getForwardFields
(
int
input
,
int
field
)
{
if
(
input
!=
0
)
{
throw
new
IndexOutOfBoundsException
();
}
return
this
.
getForwardedField
(
field
);
}
@Override
public
FieldSet
getSourceField
(
int
input
,
int
field
)
{
if
(
input
!=
0
)
{
throw
new
IndexOutOfBoundsException
();
}
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
field
);
}
return
this
.
forwardedFrom
(
field
);
}
public
FieldSet
forwardedFrom
(
int
dest
)
{
FieldSet
fs
=
null
;
for
(
Map
.
Entry
<
Integer
,
FieldSet
>
entry
:
forwardedFields
.
entrySet
())
{
if
(
entry
.
getValue
().
contains
(
dest
))
{
if
(
fs
==
null
)
{
fs
=
new
FieldSet
();
}
fs
=
fs
.
addField
(
entry
.
getKey
());
}
}
return
fs
;
}
public
SingleInputSemanticProperties
()
{
init
();
}
...
...
@@ -95,6 +129,10 @@ public class SingleInputSemanticProperties extends SemanticProperties {
* @return the destination fields, or null if they do not exist
*/
public
FieldSet
getForwardedField
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
return
this
.
forwardedFields
.
get
(
sourceField
);
}
...
...
@@ -145,7 +183,12 @@ public class SingleInputSemanticProperties extends SemanticProperties {
(
forwardedFields
==
null
||
forwardedFields
.
isEmpty
())
&&
(
readFields
==
null
||
readFields
.
size
()
==
0
);
}
@Override
public
String
toString
()
{
return
"SISP("
+
this
.
forwardedFields
+
")"
;
}
private
void
init
()
{
this
.
forwardedFields
=
new
HashMap
<
Integer
,
FieldSet
>();
this
.
readFields
=
null
;
...
...
flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
浏览文件 @
78f41e9c
...
...
@@ -155,8 +155,8 @@ public class ConnectedComponents implements ProgramDescription {
* a vertex is associated with, with a (Source-Vertex-ID, Target-VertexID) edge. The function
* produces a (Target-vertex-ID, Component-ID) pair.
*/
@ConstantFieldsFirst
(
"1 ->
0
"
)
@ConstantFieldsSecond
(
"1 ->
1
"
)
@ConstantFieldsFirst
(
"1 ->
1
"
)
@ConstantFieldsSecond
(
"1 ->
0
"
)
public
static
final
class
NeighborWithComponentIDJoin
implements
JoinFunction
<
Tuple2
<
Long
,
Long
>,
Tuple2
<
Long
,
Long
>,
Tuple2
<
Long
,
Long
>>
{
@Override
...
...
@@ -178,8 +178,6 @@ public class ConnectedComponents implements ProgramDescription {
}
}
@Override
public
String
getDescription
()
{
return
"Parameters: <vertices-path> <edges-path> <result-path> <max-number-of-iterations>"
;
...
...
flink-java/src/main/java/org/apache/flink/api/java/record/functions/FunctionAnnotation.java
浏览文件 @
78f41e9c
...
...
@@ -341,19 +341,19 @@ public class FunctionAnnotation {
return
semanticProperties
;
}
private
static
final
class
ImplicitlyForwardingSingleInputSemanticProperties
extends
SingleInputSemanticProperties
{
private
static
final
long
serialVersionUID
=
1L
;
private
FieldSet
nonForwardedFields
;
private
ImplicitlyForwardingSingleInputSemanticProperties
(
FieldSet
nonForwardedFields
)
{
this
.
nonForwardedFields
=
nonForwardedFields
;
addWrittenFields
(
nonForwardedFields
);
}
/**
* Returns the logical position where the given field is written to.
* In this variant of the semantic properties, all fields are assumed implicitly forwarded,
...
...
@@ -362,52 +362,81 @@ public class FunctionAnnotation {
*/
@Override
public
FieldSet
getForwardedField
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
if
(
this
.
nonForwardedFields
.
contains
(
sourceField
))
{
return
null
;
}
else
{
return
new
FieldSet
(
sourceField
);
}
}
@Override
public
FieldSet
getSourceField
(
int
input
,
int
field
)
{
if
(
input
!=
0
)
{
throw
new
IndexOutOfBoundsException
();
}
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
field
);
}
if
(
this
.
nonForwardedFields
==
null
)
{
return
super
.
getSourceField
(
input
,
field
);
}
if
(
this
.
nonForwardedFields
.
contains
(
field
))
{
return
null
;
}
else
{
return
new
FieldSet
(
field
);
}
}
@Override
public
void
addForwardedField
(
int
sourceField
,
int
destinationField
)
{
throw
new
UnsupportedOperationException
(
"When defining fields as implicitly constant "
+
"(such as through the ConstantFieldsExcept annotation), you cannot manually add forwarded fields."
);
}
@Override
public
void
addForwardedField
(
int
sourceField
,
FieldSet
destinationFields
)
{
throw
new
UnsupportedOperationException
(
"When defining fields as implicitly constant "
+
"(such as through the ConstantFieldsExcept annotation), you cannot manually add forwarded fields."
);
}
@Override
public
void
setForwardedField
(
int
sourceField
,
FieldSet
destinationFields
)
{
throw
new
UnsupportedOperationException
(
"When defining fields as implicitly constant "
+
"(such as through the ConstantFieldsExcept annotation), you cannot manually add forwarded fields."
);
}
}
private
static
final
class
ImplicitlyForwardingTwoInputSemanticProperties
extends
DualInputSemanticProperties
{
private
static
final
long
serialVersionUID
=
1L
;
private
FieldSet
nonForwardedFields1
;
private
FieldSet
nonForwardedFields2
;
private
ImplicitlyForwardingTwoInputSemanticProperties
()
{}
public
void
setImplicitlyForwardingFirstExcept
(
FieldSet
nonForwardedFields
)
{
this
.
nonForwardedFields1
=
nonForwardedFields
;
}
public
void
setImplicitlyForwardingSecondExcept
(
FieldSet
nonForwardedFields
)
{
this
.
nonForwardedFields2
=
nonForwardedFields
;
}
@Override
public
FieldSet
getForwardedField1
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
if
(
this
.
nonForwardedFields1
==
null
)
{
return
super
.
getForwardedField1
(
sourceField
);
}
...
...
@@ -419,9 +448,13 @@ public class FunctionAnnotation {
}
}
}
@Override
public
FieldSet
getForwardedField2
(
int
sourceField
)
{
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
sourceField
);
}
if
(
this
.
nonForwardedFields2
==
null
)
{
return
super
.
getForwardedField2
(
sourceField
);
}
...
...
@@ -433,7 +466,36 @@ public class FunctionAnnotation {
}
}
}
@Override
public
FieldSet
getSourceField
(
int
input
,
int
field
)
{
if
(
input
!=
0
&&
input
!=
1
)
{
throw
new
IndexOutOfBoundsException
();
}
if
(
isAllFieldsConstant
())
{
return
new
FieldSet
(
field
);
}
if
(
this
.
nonForwardedFields1
==
null
&&
this
.
nonForwardedFields2
==
null
)
{
return
super
.
getSourceField
(
input
,
field
);
}
if
(
input
==
0
&&
this
.
nonForwardedFields1
!=
null
&&
this
.
nonForwardedFields1
.
contains
(
field
))
{
return
null
;
}
else
if
(
input
==
0
)
{
return
new
FieldSet
(
field
);
}
if
(
input
==
1
&&
this
.
nonForwardedFields2
!=
null
&&
this
.
nonForwardedFields2
.
contains
(
field
))
{
return
null
;
}
else
if
(
input
==
1
)
{
return
new
FieldSet
(
field
);
}
return
null
;
}
@Override
public
void
addForwardedField1
(
int
sourceField
,
int
destinationField
)
{
if
(
this
.
nonForwardedFields1
==
null
)
{
...
...
@@ -444,7 +506,7 @@ public class FunctionAnnotation {
"(such as through the ConstantFieldsFirstExcept annotation), you cannot manually add forwarded fields."
);
}
}
@Override
public
void
addForwardedField1
(
int
sourceField
,
FieldSet
destinationFields
)
{
if
(
this
.
nonForwardedFields1
==
null
)
{
...
...
@@ -455,7 +517,7 @@ public class FunctionAnnotation {
"(such as through the ConstantFieldsFirstExcept annotation), you cannot manually add forwarded fields."
);
}
}
@Override
public
void
setForwardedField1
(
int
sourceField
,
FieldSet
destinationFields
)
{
if
(
this
.
nonForwardedFields1
==
null
)
{
...
...
@@ -466,7 +528,7 @@ public class FunctionAnnotation {
"(such as through the ConstantFieldsFirstExcept annotation), you cannot manually add forwarded fields."
);
}
}
@Override
public
void
addForwardedField2
(
int
sourceField
,
int
destinationField
)
{
if
(
this
.
nonForwardedFields2
==
null
)
{
...
...
@@ -477,7 +539,7 @@ public class FunctionAnnotation {
"(such as through the ConstantFieldsSecondExcept annotation), you cannot manually add forwarded fields."
);
}
}
@Override
public
void
addForwardedField2
(
int
sourceField
,
FieldSet
destinationFields
)
{
if
(
this
.
nonForwardedFields2
==
null
)
{
...
...
@@ -488,7 +550,7 @@ public class FunctionAnnotation {
"(such as through the ConstantFieldsSecondExcept annotation), you cannot manually add forwarded fields."
);
}
}
@Override
public
void
setForwardedField2
(
int
sourceField
,
FieldSet
destinationFields
)
{
if
(
this
.
nonForwardedFields2
==
null
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录