arXiv:1301.4200v1 [cs.DB] 17 Jan 2013

Enabling Operator Reordering in Data Flow Programs
Through Static Code Analysis
Fabian Hueske

Aljoscha Krettek

Kostas Tzoumas

Technische Universität Berlin, Germany
fabian.hueske@tu-berlin.de

Technische Universität Berlin, Germany
aljoscha.krettek@campus.tu-berlin.de

Technische Universität Berlin, Germany
kostas.tzoumas@tu-berlin.de

Abstract
In many massively parallel data management platforms, programs
are represented as small imperative pieces of code connected in a
data flow. This popular abstraction makes it hard to apply algebraic
reordering techniques employed by relational DBMSs and other
systems that use an algebraic programming abstraction. We present
a code analysis technique based on reverse data and control flow
analysis that discovers a set of properties from user code, which
can be used to emulate algebraic optimizations in this setting.

1. Introduction
Motivated by the recent “Big Data” trend, a new breed of massively
parallel data processing systems has emerged. Examples of these
systems include MapReduce [8] and its open-source implementation Hadoop [1], Dryad [11], Hyracks [6], and our own Stratosphere system [5]. These systems typically expose to the programmer a data flow programming model. Programs are composed as
directed acyclic graphs (DAGs) of operators, some of the latter typically being written in a general-purpose imperative programming
language. This model restricts control flow only within the limits
of operators, and permits only dataflow-based communication between operators. Since operators can only communicate with each
other by passing sets of records in a pre-defined hardwired manner,
set-oriented execution and data parallelism can be achieved.
Contrary to these systems, relational DBMSs, the traditional
workhorses for managing data at scale, are able to optimize queries because they adopt an algebraic programming model based
on relational algebra. For example, a query optimizer is able to
⋉ (S ⋊
⋉ T )) to the expression
transform the expression σR.X <3 (R ⋊
⋉ S) ⋊
⋉ T , exploiting the associativity and commuta(σR.X <3 (R) ⋊
tivity properties of selections and joins.
While algebraic reordering can lead to orders of magnitude
faster execution, it is not fully supported by modern parallel processing systems, due to their non-algebraic programming models.
Operators are typically written in a general-purpose imperative language, and their semantics are therefore hidden from the system. In
our previous work [10], we bridged this gap by showing that exposure of a handful of operator properties to the system can enable
reorderings that can simulate most algebraic reorderings used by
modern query optimizers. We discovered these properties using a

Permission to make digital or hard copies of all or part of this work for personal or
classroom use is granted without fee provided that copies are not made or distributed
for profit or commercial advantage and that copies bear this notice and the full citation
on the first page. To copy otherwise, to republish, to post on servers or to redistribute
to lists, requires prior specific permission and/or a fee.
XLDI 2012 September 9, 2012, Copenhagen, Denmark.
Copyright c 2012 ACM [to be supplied]. . . $10.00

custom, shallow code analysis pass over the operators’ code. Here
we describe this code analysis in detail, which we believe is of interest by itself as an non-traditional use case of code analysis techniques. We note that our techniques are applicable in the context
of many data processing systems which support MapReduce-style
UDFs such as parallel programming models [5, 8], higher-level languages [2, 7], and database systems [3, 9].
Related work: In our previous work [10] we describe and formally prove the conditions to reorder user-defined operators. That
paper also contains a more complete treatment of related work.
Here, we focus on more directly related research. Manimal [12]
uses static code analysis of MapReduce programs for the purpose
of recommending possible indexes. Our code analysis can be seen
as an example of peephole optimization [4], and some of the concepts may bear similarity to techniques for loop optimization. However, we are not aware of code analysis being used before for the
purpose of swapping imperative blocks of code to improve performance of data-intensive programs.
The rest of this paper is organized as follows. Section 2 describes the programming model of our system, and introduces the
reordering technology. Section 3 discusses our code analysis algorithm in detail. Finally, Section 4 concludes and offers research
directions.

2. Data Flow Operator Reordering
In our PACT programming model [5], a program P is a DAG of
sources, sinks, and operators which are connected by data channels.
A source generates records and passes them to connected operators.
A sink receives records from operators and serializes them into an
output format. Records consist of fields of arbitrary types. To define an operator O, the programmer must specify (i) a second-order
function (SOF) signature, picked from a pre-defined set of system
second-order functions (currently Map, Reduce, Match, Cross, and
CoGroup), and (ii) a first-order function (called user-defined function, UDF) that is used as the parameter of the SOF. The model is
strictly second-order, in that a UDF is not allowed to call SOFs. The
intuition of this model is that the SOF defines a logical mapping of
the operator’s input records into groups, and the UDF is invoked
once for each group. These UDF invocations are independent, and
can be thus scheduled on different nodes of a computing cluster.
Figure 1(a) shows an example PACT program. The data flow
starts with two data sources Src1 and Src2 that provide records
which have the fields [0, 1] and [3, 4] set respectively (the numbering is arbitrary). Src1 feeds its data into a Map operator with
a UDF f1 . The Map SOF creates an independent group for each
input record, and f1 is itself written in Java. UDF f1 reads both
fields of its input record (0 and 1), appends the sum of both fields
as field 2, and emits the record. Similarly, the records of Src2 are
forwarded to a Map operator with UDF f2 which sums the fields 3

(a)
Snk1

(b)
Snk1

(c)
Snk1

Match( f3 ,[0],[3])

Map( f1 )

Map( f2 )

Match( f3 ,[0],[3])

Match( f3 ,[0],[3])

Map( f1 )

Map( f2 )

[0,1]
Src1

[3,4]
Src2

[0,1]
Src1

Map( f2 )

Map( f1 )

[3,4]
Src2

[0,1]
Src1

[3,4]
Src2

Figure 1. Example Data Flows: (a) original order, (b) first reordered alternative, (c) second reordered alternative
and 4, appends the sum as field 5 and emits the record. The outputs
of both Map operators are forwarded as inputs to a Match operator
with a UDF f3 and the key field [0] for the first and [3] for the second input. The Match SOF creates a group for each pair of records
from both inputs that match on their key fields. f3 merges the fields
of both input records and emits the result. We give the pseudo-code
of all three user functions in the form of 3-address code [4] below.
10:
11:
12:
13:
14:
15:
16:

f1(InRec $ir)
$a:=getField($ir,0)
$b:=getField($ir,1)
$c:=$a + $b
$or:=copy($ir)
setField($or,2,$c)
emit($or)

20:
21:
22:
23:
24:
25:
26:
27:
28:

f2(InRec $ir)
$x:=getField($ir,3)
$y:=getField($ir,4)
$z:=$x + $y
$or:=create()
setField($or,3,$x)
setField($or,4,$y)
setField($or,5,$z)
emit($or)

30: f3(InRec $ir1, InRec $ir2)
31: $or:=copy($ir1)
32: union($or,$ir2)
33: emit($or)
The pseudo-code shows the UDF API to process PACT records.
The user-functions f1, f2, and f3 receive as input one or two
input records of type InRec. The only way that a user function can emit an output record of type OutRec is by calling the
emit(OutRec) function. Output records can be either initialized
as empty (OutRec create()), or by copying an input record
(OutRec copy(InRec)). Records can be combined via the function void union(OutRec,InRec). Fields can be read to a variable via Object getField(InRec,int) addressed by their position in the input record. The value of a field can be set via void
setField(OutRec,int,Object). Note that our record API is
based on basic operations and similar to other systems’ APIs such
as Apache Pig [2].
Figures 1 (b) and (c) show potential reorderings of the original
data flow (a) where either Map( f1 ) or Map( f2 ) has been reordered
with Match( f3 , [0], [3]). While data flow (b) is a valid reordering,
alternative (c) does not produce the same result as (a). In previous
work, we presented conditions for valid reorderings of data flow
operators centered around conflicts of operators on fields [10]. For
example, since we know that f1 reads fields 0 and 1, and writes field
2, while f3 reads fields 0 and 3, we can conclude that f1 and f3 only
have a read conflict on field 0, and can thus be safely reordered.
UDFs that have write conflicts cannot be reordered. This would be
true if f1 did not append the sum as field 2, but overwrote field 0
with the sum. Additional complications arise from the way output
records are formed. Although on the first sight, f1 and f2 perform a
very similar operation, i. e., summing two fields and appending the
result, there is a fundamental difference. While f1 creates its output
record by copying the input record (line 14), f2 creates an empty
output record (line 24) and explicitly copies the fields of the input
record (lines 25,26). The side effect of creating an empty output

record is that all fields of an input record are implicitly removed
from the output. By reordering Map( f2 ) with Match( f3 , [0], [3]), the
fields 0, 1, and 2 will get lost since Map( f2 ) does not explicitly copy
them into the newly created output record.
The information that needs to be extracted from the user code
in order to reason about reordering of operators is as follows. The
read set R f of a UDF f is the set of fields from its input data sets
that might influence the UDF’s output, i. e., fields that are read and
evaluated by f . The write set W f is the set of fields of the output data set that have different values from the corresponding input
field. The emit cardinality bounds ⌊EC f ⌋ and ⌈EC f ⌉ are lower and
upper bounds for the number of records emitted per invocation of
f . Reference [10] defines these properties more formally, and provides conditions for reordering operators with various SOFs given
knowledge of these properties. In addition to change the order of
operators, the optimizer can leverage these properties to avoid expensive data processing operations, e. g., a previously partitioned
data set is still partitioned after a UDF was applied, if the partitioning fields were not modified by the UDF. Moreover, field projections can be pushed down based on read set information.
While it is very difficult to statically derive the exact properties
by UDF code analysis in the general case, it is possible to conservatively approximate them. In reference [10] we discussed this
static code analysis pass for the simple case of unary operators. In
the next section, we provide the full algorithm that deals with the
additional complexity due to binary operators, and provide detailed
pseudo-code.

3. Code Analysis Algorithm
Our algorithm relies on a static code analysis (SCA) framework
to get the bytecode of the analyzed UDF, for example as typed
three-address code [4]. The framework must provide a control flow
graph (CFG) abstraction, in which each code statement is represented by one node along with a function P REDS(s) that returns
the statements in the CFG that are “true” predecessors of statement s, i.e., they are not both predecessors and descendants. Finally, the framework must provide two methods D EF -U SE(s, $v)
and U SE -D EF(s, $v) that represent the Definition-Use chain of the
variable $v at statement s, and the Use-Definition chain of variable
$v at statement s respectively. Any SCA framework that provides
these abstraction can be used.
The algorithm visits each UDF in a topological order implied
by the program DAG starting from the data sources. For each UDF
f , the function V ISIT-U DF of Algorithm 1 is invoked. First, we
compute the read set R f of the UDF (lines 7-10). For each statement
of the form $t := getField($ir, n) that results in a valid use
of variable $t (D EF -U SE (g, $t)6= 0)
/ we add field n to R f
Approximating the write set W f is more involved. We compute
four sets of integers that we eventually use to compute an approximation of W f . The origin set O f of UDF f is a set of input ids.
An integer o ∈ O f means that all fields of the o-th input record of
f are copied verbatim to the output. The explicit modification set
E f contains fields that are modified and then included in the output. We generally assume that fields are uniquely numbered within
the program (as in Figure 1). The copy set C f contains fields that
are copied verbatim from one input record to the output. Finally,
the projection set Pf contains fields that are projected from the output, by explicitly being set to null. The write set is computed from
these sets using the function C OMPUTE -W RITE -S ET (lines 1-5).
All fields in E f and Pf are explicitly modified or set to null and
therefore in W f . For inputs that are not in the origin set O f , we
add all fields of that input which are not in C f , i. e., not explicitly
copied.
To derive the four sets, function V ISIT-U DF finds all statements
of the form e: emit($or), which include the output record $or in

Algorithm 1 Code analysis algorithm
1:
2:
3:
4:
5:
6:
7:
8:
9:
10:
11:
12:
13:
14:
15:
16:
17:
18:
19:
20:
21:
22:
23:
24:
25:
26:
27:
28:
29:
30:
31:
32:
33:
34:
35:
36:
37:
38:
39:
40:
41:
42:

function C OMPUTE -W RITE -S ET( f ,O f ,E f ,C f ,Pf )
W f = E f ∪ Pf
for i ∈ I NPUTS( f ) do
if i ∈
/ O f then W f = W f ∪ (I NPUT-F IELDS( f ,i) \C f )
return W f
function V ISIT-UDF( f )
R f = 0/
G = all statements of the form g:$t=getField($ir,n)
for g in G do
if D EF -U SE(g, $t)6= 0/ then R f = R f ∪ {n}
E = all statements of the form e:emit($or)
(O f ,E f ,C f ,Pf ) = V ISIT-S TMT(A NY(E), $or)
for e in E do
(Oe ,Ee ,Ce ,Pe ) = V ISIT-S TMT(e, $or)
(O f ,E f ,C f ,Pf ) = M ERGE((O f ,E f ,C f ,Pf ),(Oe ,Ee ,Ce ,Pe ))
return (R f ,O f ,E f ,C f ,Pf )
function V ISIT-S TMT(s, $or)
if VISITED(s,$or) then
return M EMO -S ETS (s, $or)
V ISITED(s,$or) = true
if s of the form $or = create() then return (0,
/ 0,
/ 0,
/ 0)
/
if s of the form $or = copy($ir) then
return (I NPUT-I D($ir), 0,
/ 0,
/ 0)
/
Ps = P REDS (s)
(Os ,Es ,Cs ,Ps ) = V ISIT-S TMT(A NY(Ps), $or)
for p in Ps do
(O p ,E p ,Cp ,Pp ) = V ISIT-S TMT(p, $or)
(Os ,Es ,Cs ,Ps ) = M ERGE((Os ,Es ,Cs ,Ps ), (O p ,E p ,Cp ,Pp ))
if s of the form union($or, $ir) then
return (Os ∪ I NPUT-I D($ir),Es ,Cs ,Ps )
if s of the form setField($or, n, $t) then
T =U SE -D EF (s, $t)
if all t ∈ T of the form $t=getField($ir,n) then
return (Os ,Es ,Cs ∪ {n},Ps )
else
return (Os ,Es ∪ {n},Cs ,Ps )
if s of the form setField($or, n, null) then
return (Os ,Es ,Cs ,Ps ∪ {n})
function M ERGE((O1 ,E1 ,C1 ,P1 ), (O2 ,E2 ,C2 ,P2 ))
C = (C1 ∩C2 ) ∪ {x|x ∈ C1 , I NPUT-I D(x) ∈ O2 }
∪ {x|x ∈ C2 , I NPUT-I D(x) ∈ O1 }
return (O1 ∩ O2 ,E1 ∪ E2 ,C,P1 ∪ P2 )

ation point of the output record, where it is initialized to the empty
record. The recursion then ends. Another base case is reaching a
statement $or = copy($ir) (line 22) where the output record is
created by copying all fields of the input record $ir. This adds the
input id of record $ir to the origin set O. A union statement (line
29) results in an inclusion of the input id of the input record $ir
in the origin set O, and a further recursion for the output record
$or. The algorithm maintains a memo table M EMO -S ETS to support early exit of the recursion in the presence of loops (line 18).
The memo table is implicitly updated at every return statement of
V ISIT-S TMT .
Function V ISIT-S TMT always terminates in the presence of
loops in the UDF code, since it will eventually find the statement
that creates the output record, or visit a previously seen statement.
This is due to P REDS always exiting a loop after visiting its first
statement. Thus, loop bodies are only visited once by the algorithm.
The complexity of the algorithm is O(en), where n is the size of the
UDF code, and e the number of emit statements. This assumes that
the Use-Def and Def-Use chains have been precomputed.
The lower and upper bound on the emit cardinality of the UDF
can be derived by another pass over the UDF code. We determine
the bounds for each emit statement e and combine those to derive
the bounds of the UDF. For the lower bound ⌊EC f ⌋, we check
whether there is a statement before statement e that jumps to a
statement after e. If there is none, the emit statement will always
be executed and we set ⌊EC f ⌋ = 1. If such a statement exists,
statement e could potentially be skipped during execution, so we
set ⌊EC f ⌋ = 0. For the upper bound ⌈EC f ⌉, we determine whether
there is a statement after e that can jump to a statement before
e. If yes, the statement could be executed several times during
the UDF’s execution, so we set ⌈EC f ⌉ = +∞. If such a statement
does not exist, statement e can be executed at most once so we set
⌈EC f ⌉ = 1. To combine the bounds we choose for the lower bound
of the UDF the highest lower bound over all emit statements and for
the upper bound the highest upper bound over all emit statements.
Our previous work [10] compares read and write sets which
are automatically derived by our static code analysis technique and
from manually attached annotations. We show that our technique
yields very precise estimations with only little loss of optimization
potential. However, we note that the estimation quality depends on
the programming style.

4. Conclusions and Future Work
the output (line 11). It then calls for each statement e the recursive
function V ISIT-S TMT that recurses from statement e backwards
in the control flow graph (lines 12-15). The function performs a
combination of reverse data flow and control flow analysis but does
not change the values computed for statements once they have been
determined. The function A NY returns an arbitrary element of a set.
The useful work is done in lines 24-38 of the algorithm. First,
the algorithm finds all predecessor statements of the current statement, and recursively calls V ISIT-S TMT . The sets are merged using
the M ERGE function (lines 39-42). M ERGE provides a conservative
approximation of these sets, by creating maximal E, P sets, and
minimal O,C sets. This guarantees that the data conflicts that will
arise are a superset of the true conflicts in the program. When a
statement of the form setField($or, n, null) is found (line
37), field n of the output record is explicitly projected, and is
thus added to the projection set P. When a statement of the form
setField($or, n, $t) is found (line 31), the U SE -D EF chain
of $t is checked. If the temporary variable $t came directly from
field n of the input, it is added to the copy set C, otherwise it is
added to the explicit write set E. When we encounter a statement
of the form $or = create() (line 21), we have reached the cre-

We presented a shallow code analysis technique that operates on
data flow programs composed of imperative building blocks (“operators”). The analysis is a hybrid of reverse data flow and control flow analysis, and determines sets of record fields that express
the data conflicts of operators. These sets can be used to “emulate” algebraic reorderings in the dataflow program. Our techniques
guarantee safety through conservatism and are applicable to many
data processing systems that support UDFs. Future work includes
research on intrusive user-code optimizations, i. e., modifying the
code of UDFs, and on the effects that the use of functional programming languages to specify UDFs has on our approach and possible
optimizations.

Acknowledgments
This research was funded by the German Research Foundation
under grant FOR 1036. We would like to thank Volker Markl,
our coauthors from previous work [10], and the members of the
Stratosphere team.

References
[1] http://hadoop.apache.org/.

[2] http://pig.apache.org/.
[3] http://www.greenplum.com/technology/mapreduce.

[8] J. Dean and S. Ghemawat. Mapreduce: Simplified data processing on
large clusters. In OSDI, pages 137–150, 2004.

Compilers:

[9] E. Friedman, P. M. Pawlowski, and J. Cieslewicz. Sql/mapreduce: A
practical approach to self-describing, polymorphic, and parallelizable
user-defined functions. PVLDB, 2(2):1402–1413, 2009.

[5] D. Battré, S. Ewen, F. Hueske, O. Kao, V. Markl, and D. Warneke.
Nephele/pacts: a programming model and execution framework for
web-scale analytical processing. In SoCC, pages 119–130, 2010.

[10] F. Hueske, M. Peters, M. Sax, A. Rheinländer, R. Bergmann, A. Krettek, and K. Tzoumas. Opening the black boxes in data flow optimization. PVLDB, 5(11):1256–1267, 2012.

[6] V. R. Borkar, M. J. Carey, R. Grover, N. Onose, and R. Vernica.
Hyracks: A flexible and extensible foundation for data-intensive computing. In ICDE, pages 1151–1162, 2011.

[11] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: distributed data-parallel programs from sequential building blocks. pages
59–72, 2007.

[7] R. Chaiken, B. Jenkins, P.-Å. Larson, B. Ramsey, D. Shakib,
S. Weaver, and J. Zhou. Scope: easy and efficient parallel processing of massive data sets. PVLDB, 1(2):1265–1276, 2008.

[12] E. Jahani, M. J. Cafarella, and C. Ré. Automatic optimization for
mapreduce programs. PVLDB, 4(6):385–396, 2011.

[4] A. V. Aho, M. S. Lam, R. Sethi, and J. D. Ullman.
Principles, Techniques and Tools. Pearson, 2006.

