[SPARK-13996][SQL] Add more not null attributes for Filter codegen

[viirya]

What changes were proposed in this pull request?

JIRA: https://issues.apache.org/jira/browse/SPARK-13996

Filter codegen finds the attributes not null by checking IsNotNull(a) expression with a condition if child.output.contains(a). However, the current approach to checking it is not comprehensive. We can improve it.

E.g., for this plan:

val rdd = sqlContext.sparkContext.makeRDD(Seq(Row(1, "1"), Row(null, "1"), Row(2, "2")))
val schema = new StructType().add("k", IntegerType).add("v", StringType)
val smallDF = sqlContext.createDataFrame(rdd, schema)
val df = smallDF.filter("isnotnull(k + 1)")

The code snippet generated without this patch:

/* 031 */   protected void processNext() throws java.io.IOException {
/* 032 */     /*** PRODUCE: Filter isnotnull((k#0 + 1)) */
/* 033 */
/* 034 */     /*** PRODUCE: INPUT */
/* 035 */
/* 036 */     while (!shouldStop() && inputadapter_input.hasNext()) {
/* 037 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 038 */       /*** CONSUME: Filter isnotnull((k#0 + 1)) */
/* 039 */       /* input[0, int] */
/* 040 */       boolean filter_isNull = inputadapter_row.isNullAt(0);
/* 041 */       int filter_value = filter_isNull ? -1 : (inputadapter_row.getInt(0));
/* 042 */
/* 043 */       /* isnotnull((input[0, int] + 1)) */
/* 044 */       /* (input[0, int] + 1) */
/* 045 */       boolean filter_isNull3 = true;
/* 046 */       int filter_value3 = -1;
/* 047 */   
/* 048 */       if (!filter_isNull) {
/* 049 */         filter_isNull3 = false; // resultCode could change nullability.
/* 050 */         filter_value3 = filter_value + 1;
/* 051 */     
/* 052 */       }
/* 053 */       if (!(!(filter_isNull3))) continue;
/* 054 */   
/* 055 */       filter_metricValue.add(1);

With this patch:

/* 031 */   protected void processNext() throws java.io.IOException {
/* 032 */     /*** PRODUCE: Filter isnotnull((k#0 + 1)) */
/* 033 */
/* 034 */     /*** PRODUCE: INPUT */
/* 035 */
/* 036 */     while (!shouldStop() && inputadapter_input.hasNext()) {
/* 037 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 038 */       /*** CONSUME: Filter isnotnull((k#0 + 1)) */
/* 039 */       /* input[0, int] */
/* 040 */       boolean filter_isNull = inputadapter_row.isNullAt(0);
/* 041 */       int filter_value = filter_isNull ? -1 : (inputadapter_row.getInt(0));
/* 042 */
/* 043 */       if (filter_isNull) continue;
/* 044 */
/* 045 */       filter_metricValue.add(1);

How was this patch tested?

Existing tests.

[SparkQA]

Test build #53498 has finished for PR 11810 at commit 8e59a74.

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

[SparkQA]

Test build #53511 has finished for PR 11810 at commit 376c6f0.

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

[SparkQA]

Test build #53523 has finished for PR 11810 at commit 3d73249.

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

[viirya]

The attributes from conditions can have different qualifiers than child.output, which makes later indexOf call failed. So here use exprId instead.

[SparkQA]

Test build #53592 has finished for PR 11810 at commit 662c4e3.

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

[viirya]

cc @davies

[viirya]

ping @davies Please take a look at this to see if it is good for you. Thanks.

[davies]

This is not correct, for example, IsNotNull(IsNull(a)) does not means a can not be null.

[viirya]

ah, you are right. With the NullIntolerant trait introduced in #11809, we can solve this easily. Wait for that PR to be ready then we can re-visit this. Thanks!

[SparkQA]

Test build #54767 has finished for PR 11810 at commit 0c7afe7.

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

[viirya]

retest this please.

[SparkQA]

Test build #54771 has finished for PR 11810 at commit 0c7afe7.

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

[viirya]

retest this please.

[viirya]

Looks like StateStoreSuite a bit flaky...

[SparkQA]

Test build #54774 has finished for PR 11810 at commit 0c7afe7.

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

[viirya]

retest this please.

[SparkQA]

Test build #54786 has finished for PR 11810 at commit 0c7afe7.

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

[davies]

LGTM

[davies]

Merging this into master, thanks!

[viirya]

Thanks for reviewing this.