{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/17302","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/17302/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/17302/comments","events_url":"https://api.github.com/repos/apache/spark/issues/17302/events","html_url":"https://github.com/apache/spark/pull/17302","id":214347136,"number":17302,"title":"[SPARK-19959][SQL] Fix to throw NullPointerException in  df[java.lang.Long].collect","user":{"login":"kiszk","id":1315079,"avatar_url":"https://avatars1.githubusercontent.com/u/1315079?v=3","gravatar_id":"","url":"https://api.github.com/users/kiszk","html_url":"https://github.com/kiszk","followers_url":"https://api.github.com/users/kiszk/followers","following_url":"https://api.github.com/users/kiszk/following{/other_user}","gists_url":"https://api.github.com/users/kiszk/gists{/gist_id}","starred_url":"https://api.github.com/users/kiszk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kiszk/subscriptions","organizations_url":"https://api.github.com/users/kiszk/orgs","repos_url":"https://api.github.com/users/kiszk/repos","events_url":"https://api.github.com/users/kiszk/events{/privacy}","received_events_url":"https://api.github.com/users/kiszk/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":3,"created_at":"2017-03-15T10:33:51Z","updated_at":"2017-03-15T19:32:38Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/17302","html_url":"https://github.com/apache/spark/pull/17302","diff_url":"https://github.com/apache/spark/pull/17302.diff","patch_url":"https://github.com/apache/spark/pull/17302.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR fixes `NullPointerException` in the generated code by Catalyst. When we run the following code, we get the following `NullPointerException`. This is because there is no null checks for `inputadapter_value`  while `java.lang.Long inputadapter_value` at Line 30 may have `null`.\r\n\r\nThis happen when a type of DataFrame is nullable primitive type such as `java.lang.Long` and the wholestage codegen is used. While the physical plan keeps `nullable=true` in `input[0, java.lang.Long, true].longValue`, `BoundReference.doGenCode` ignores `nullable=true`. Thus, nullcheck code will not be generated and `NullPointerException` will occur.\r\n\r\nThis PR checks the nullability and correctly generates nullcheck if needed.\r\n```java\r\nsparkContext.parallelize(Seq[java.lang.Long](0L, null, 2L), 1).toDF.collect\r\n```\r\n\r\n```java\r\nCaused by: java.lang.NullPointerException\r\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(generated.java:37)\r\n\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\r\n\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:393)\r\n...\r\n```\r\n\r\nGenerated code without this PR\r\n```java\r\n/* 005 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 006 */   private Object[] references;\r\n/* 007 */   private scala.collection.Iterator[] inputs;\r\n/* 008 */   private scala.collection.Iterator inputadapter_input;\r\n/* 009 */   private UnsafeRow serializefromobject_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\r\n/* 012 */\r\n/* 013 */   public GeneratedIterator(Object[] references) {\r\n/* 014 */     this.references = references;\r\n/* 015 */   }\r\n/* 016 */\r\n/* 017 */   public void init(int index, scala.collection.Iterator[] inputs) {\r\n/* 018 */     partitionIndex = index;\r\n/* 019 */     this.inputs = inputs;\r\n/* 020 */     inputadapter_input = inputs[0];\r\n/* 021 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 022 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 023 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 024 */\r\n/* 025 */   }\r\n/* 026 */\r\n/* 027 */   protected void processNext() throws java.io.IOException {\r\n/* 028 */     while (inputadapter_input.hasNext() && !stopEarly()) {\r\n/* 029 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 030 */       java.lang.Long inputadapter_value = (java.lang.Long)inputadapter_row.get(0, null);\r\n/* 031 */\r\n/* 032 */       boolean serializefromobject_isNull = true;\r\n/* 033 */       long serializefromobject_value = -1L;\r\n/* 034 */       if (!false) {\r\n/* 035 */         serializefromobject_isNull = false;\r\n/* 036 */         if (!serializefromobject_isNull) {\r\n/* 037 */           serializefromobject_value = inputadapter_value.longValue();\r\n/* 038 */         }\r\n/* 039 */\r\n/* 040 */       }\r\n/* 041 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 042 */\r\n/* 043 */       if (serializefromobject_isNull) {\r\n/* 044 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 045 */       } else {\r\n/* 046 */         serializefromobject_rowWriter.write(0, serializefromobject_value);\r\n/* 047 */       }\r\n/* 048 */       append(serializefromobject_result);\r\n/* 049 */       if (shouldStop()) return;\r\n/* 050 */     }\r\n/* 051 */   }\r\n/* 052 */ }\r\n```\r\n\r\nGenerated code with this PR (Only line 34 is changed)\r\n\r\n```java\r\n/* 005 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 006 */   private Object[] references;\r\n/* 007 */   private scala.collection.Iterator[] inputs;\r\n/* 008 */   private scala.collection.Iterator inputadapter_input;\r\n/* 009 */   private UnsafeRow serializefromobject_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\r\n/* 012 */\r\n/* 013 */   public GeneratedIterator(Object[] references) {\r\n/* 014 */     this.references = references;\r\n/* 015 */   }\r\n/* 016 */\r\n/* 017 */   public void init(int index, scala.collection.Iterator[] inputs) {\r\n/* 018 */     partitionIndex = index;\r\n/* 019 */     this.inputs = inputs;\r\n/* 020 */     inputadapter_input = inputs[0];\r\n/* 021 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 022 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 023 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 024 */\r\n/* 025 */   }\r\n/* 026 */\r\n/* 028 */     while (inputadapter_input.hasNext() && !stopEarly()) {\r\n/* 029 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 030 */       java.lang.Long inputadapter_value = (java.lang.Long)inputadapter_row.get(0, null);\r\n/* 031 */\r\n/* 032 */       boolean serializefromobject_isNull = true;\r\n/* 033 */       long serializefromobject_value = -1L;\r\n/* 034 */       if (!((false) || inputadapter_row.isNullAt(0))) {\r\n/* 035 */         serializefromobject_isNull = false;\r\n/* 036 */         if (!serializefromobject_isNull) {\r\n/* 037 */           serializefromobject_value = inputadapter_value.longValue();\r\n/* 038 */         }\r\n/* 039 */\r\n/* 040 */       }\r\n/* 041 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 042 */\r\n/* 043 */       if (serializefromobject_isNull) {\r\n/* 044 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 045 */       } else {\r\n/* 046 */         serializefromobject_rowWriter.write(0, serializefromobject_value);\r\n/* 047 */       }\r\n/* 048 */       append(serializefromobject_result);\r\n/* 049 */       if (shouldStop()) return;\r\n/* 050 */     }\r\n/* 051 */   }\r\n/* 052 */ }\r\n```\r\n\r\n## How was this patch tested?\r\n\r\nAdded new test suites in `DataFrameSuites`"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/286854118","html_url":"https://github.com/apache/spark/pull/17302#issuecomment-286854118","issue_url":"https://api.github.com/repos/apache/spark/issues/17302","id":286854118,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars2.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2017-03-15T19:32:38Z","updated_at":"2017-03-15T19:32:38Z","body":"**[Test build #74614 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74614/testReport)** for PR 17302 at commit [`2deeba8`](https://github.com/apache/spark/commit/2deeba8ce246ffad26097f74525c1f0fbe4237ef)."}}
{"actor":{"display_login":"SparkQA"}}