[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Filter-Join Ordering Issue

Hi Dylan,

I roughly looked at your job program and the DAG of the job. It seems that the optimizer chose the wrong optimization execution plan.

cc Till.

Thanks, vino.

Dylan Adams <dylan.adams@xxxxxxxxx> 于2018年8月8日周三 上午2:26写道:
I'm trying to use the Flink DataSet API to validate some records and have run into an issue. My program uses joins to validate inputs against reference data. One of the attributes I'm validating is optional, and only needs to be validated when non-NULL. So I added a filter to prevent the null-keyed records from being used in the validation join, and was surprised to receive this exception:

java.lang.RuntimeException: A NullPointerException occured while accessing a key field in a POJO. Most likely, the value grouped/joined on is null. Field name: optionalKey
at org.apache.flink.api.java.typeutils.runtime.PojoComparator.hash(PojoComparator.java:199)

It looks like the problem is that Flink has pushed the hash partitioning aspect of the join before the filter for the null-keyed records and is trying to hash the null keys. The issue can be seen in the plan visualization: https://raw.githubusercontent.com/dkadams/flink-plan-issue/master/plan-visualization.png

I was able to reproduce the problem in v1.4.2 and 1.5.2, with this small project: https://github.com/dkadams/flink-plan-issue/

Is this expected behavior or a bug? FLINK-1915 seems to have the same root problem, but with a negative performance impact instead of a RuntimeException.