[SPARK-13751] [SQL] generate better code for Filter

[davies]

What changes were proposed in this pull request?

This PR improve the codegen of Filter by:

1. filter out the rows early if it have null value in it that will cause the condition result in null or false. After this, we could simplify the condition, because the input are not nullable anymore.
2. Split the condition as conjunctive predicates, then check them one by one.

Here is a piece of generated code for Filter in TPCDS Q55:

/* 109 *//*** CONSUME: Filter ((((isnotnull(d_moy#149) && isnotnull(d_year#147)) && (d_moy#149 = 11)) && (d_year#147 = 1999)) && isnotnull(d_date_sk#141)) *//* 110 *//* input[0, int] *//* 111 */booleanproject_isNull2 = rdd_row.isNullAt(0);
/* 112 */intproject_value2 = project_isNull2 ? -1 : (rdd_row.getInt(0));
/* 113 *//* input[1, int] *//* 114 */booleanproject_isNull3 = rdd_row.isNullAt(1);
/* 115 */intproject_value3 = project_isNull3 ? -1 : (rdd_row.getInt(1));
/* 116 *//* input[2, int] *//* 117 */booleanproject_isNull4 = rdd_row.isNullAt(2);
/* 118 */intproject_value4 = project_isNull4 ? -1 : (rdd_row.getInt(2));
/* 119 *//* 120 */if (project_isNull3) continue;
/* 121 */if (project_isNull4) continue;
/* 122 */if (project_isNull2) continue;
/* 123 *//* 124 *//* (input[1, int] = 11) *//* 125 */booleanfilter_value6 = false;
/* 126 */filter_value6 = project_value3 == 11;
/* 127 */if (!filter_value6) continue;
/* 128 *//* 129 *//* (input[2, int] = 1999) *//* 130 */booleanfilter_value9 = false;
/* 131 */filter_value9 = project_value4 == 1999;
/* 132 */if (!filter_value9) continue;
/* 133 *//* 134 */filter_metricValue1.add(1);
/* 135 *//* 136 *//*** CONSUME: Project [d_date_sk#141] *//* 137 *//* 138 */project_rowWriter1.write(0, project_value2);
/* 139 */append(project_result1.copy());

How was this patch tested?

Existing tests.

[SparkQA]

Test build #52686 has finished for PR 11585 at commit edc934f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class Filter(condition: Expression, child: SparkPlan)

[nongli]

@sameeragarwal how does this relate to what you are working on?

[davies]

@nongli It seems that once we have this, we don't need #11511 anymore, right?

[sameeragarwal]

@nongli@davies I think part of the rationale for having #11511 was to avoid having to extract the operator's (and its child's) constraints during the physical planning/ codegen phase. Could we not just rewrite the code generated by filter such that all isNotNull attributes are converted to ${input(i).isNull? Along with #11372, #11511 would help us achieve the same goal.

[sameeragarwal]

Also @davies, if you prefer your current approach, we then need to propagate the constraints in the physical plan as well (currently they are only propagated in the logical plan).

[davies]

After offline discussion with @sameeragarwal , we agreed that 1) the constraints should only used in logical plan 2) we should use the IsNotNull() inserted by #11372, 3) we do not need to re-order the conditions anymore, just do that in Filter. 4) The IsNotNull from join ares pushed down into it's children, join requires nullability propagated from it's children, that's another story.

I will update this PR based on these, and revert #11511

cc @yhuai@nongli

[SparkQA]

Test build #52757 has finished for PR 11585 at commit 4317853.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #52780 has finished for PR 11585 at commit 2ec7f5e.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[nongli]

comment why

[SparkQA]

Test build #52858 has finished for PR 11585 at commit b2906e6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[nongli]

LGTM

[davies]

Merged into master