Fabian Hueske created FLINK-10100:
-------------------------------------

             Summary: Optimizer pushes partitioning past Null-Filter
                 Key: FLINK-10100
                 URL: https://issues.apache.org/jira/browse/FLINK-10100
             Project: Flink
          Issue Type: Bug
          Components: DataSet API, Optimizer
    Affects Versions: 1.5.2, 1.4.2, 1.3.3, 1.6.0, 1.7.0
            Reporter: Fabian Hueske


The DataSet optimizer pushes certain operations like partitioning or sorting 
past Filter operators.
It does that because it knows that a {{FilterFunction}} cannot modify the 
records but only indicate whether a record should be forwarded or not. 

However, this causes problems if the filter should remove records with null 
keys. In that case, the partitioning can be pushed past the filter such that 
the partitioner has to deal with null keys. This can fail with a 
{{NullPointerException}}.

The following code produces an affected plan.

{code}
List<Row> rowList = new ArrayList<>();
rowList.add(Row.of(null, 1L));
rowList.add(Row.of(2L, 2L));
rowList.add(Row.of(2L, 2L));
rowList.add(Row.of(3L, 3L));
rowList.add(Row.of(null, 3L));

DataSet<Row> rows = env.fromCollection(rowList, Types.ROW(Types.LONG, 
Types.LONG));

DataSet<Long> result = rows
  .filter(r -> r.getField(0) != null)
    .setParallelism(4)
  .groupBy(0)
  .reduceGroup((Iterable<Row> vals, Collector<Long> out) -> {
      long cnt = 0L;
      for(Row v : vals) { cnt++; }
        out.collect(cnt);
      }).returns(Types.LONG)
        .setParallelism(4);

result.output(new DiscardingOutputFormat());
System.out.println(env.getExecutionPlan());
{code}

To resolve the problem, we could remove the field-forward property of 
{{FilterFunction}}. In general, it is typically more efficient to filter before 
shipping or sorting data. So this might also improve the performance of certain 
plans.

As a *workaround* until this bug is fix, users can implement the filter with a 
{{FlatMapFunction}}. {{FlatMapFunction}} is a more generic interface and the 
optimizer cannot automatically infer how the function behaves and won't push 
partitionings or sorts past the function.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Reply via email to