{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/13909","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/13909/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/13909/comments","events_url":"https://api.github.com/repos/apache/spark/issues/13909/events","html_url":"https://github.com/apache/spark/pull/13909","id":162311500,"number":13909,"title":"[SPARK-16213][SQL] Reduce runtime overhead of a program that creates an primitive array in DataFrame","user":{"login":"kiszk","id":1315079,"avatar_url":"https://avatars.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":90,"created_at":"2016-06-26T04:07:29Z","updated_at":"2016-12-26T07:49:29Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/13909","html_url":"https://github.com/apache/spark/pull/13909","diff_url":"https://github.com/apache/spark/pull/13909.diff","patch_url":"https://github.com/apache/spark/pull/13909.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR reduces runtime overhead of a program the creates an primitive array in DataFrame by using the similar approach to #15044. Generated code performs boxing operation in an assignment from InternalRow to an `Object[]` temporary array (at Lines 051 and 061 in the generated code before without this PR). If we know that type of array elements is primitive, we apply the following optimizations:\r\n1. Eliminate a pair of `isNullAt()` and a null assignment\r\n2. Allocate an primitive array instead of `Object[]` (eliminate boxing operations)\r\n3. Create `UnsafeArrayData` by using `UnsafeArrayWriter` to keep a primitive array in a row format instead of doing non-lightweight operations in constructor of `GenericArrayData` \r\nThe PR also performs the same things for `CreateMap`.\r\n\r\nHere are performance results of [DataFrame programs](https://github.com/kiszk/spark/blob/6bf54ec5e227689d69f6db991e9ecbc54e153d0a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/PrimitiveArrayBenchmark.scala#L83-L112) by up to 18.8x over without this PR.\r\n\r\n```\r\nWithout SPARK-16043\r\nOpenJDK 64-Bit Server VM 1.8.0_91-b14 on Linux 4.4.11-200.fc22.x86_64\r\nIntel Xeon E3-12xx v2 (Ivy Bridge)\r\nRead a primitive array in DataFrame:     Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nInt                                           3805 / 4150          0.0      507308.9       1.0X\r\nDouble                                        3593 / 3852          0.0      479056.9       1.1X\r\n\r\nWith SPARK-16043\r\nRead a primitive array in DataFrame:     Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nInt                                            202 /  3050          0.0       26998.2       1.0X\r\nDouble                                         198 /  204          0.0       26449.4       1.0X\r\n```\r\nNote : #15780 is enabled for these measurements\r\n\r\nAn motivating example\r\n\r\n``` java\r\nval df = sparkContext.parallelize(Seq(0.0d, 1.0d), 1).toDF\r\ndf.selectExpr(\"Array(value + 1.1d, value + 2.2d)\").show\r\n```\r\n\r\nGenerated code without this PR\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 */   private Object[] project_values;\r\n/* 013 */   private UnsafeRow project_result;\r\n/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder project_holder;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter project_rowWriter;\r\n/* 016 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter project_arrayWriter;\r\n/* 017 */\r\n/* 018 */   public GeneratedIterator(Object[] references) {\r\n/* 019 */     this.references = references;\r\n/* 020 */   }\r\n/* 021 */\r\n/* 022 */   public void init(int index, scala.collection.Iterator[] inputs) {\r\n/* 023 */     partitionIndex = index;\r\n/* 024 */     this.inputs = inputs;\r\n/* 025 */     inputadapter_input = inputs[0];\r\n/* 026 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 027 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 028 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 029 */     this.project_values = null;\r\n/* 030 */     project_result = new UnsafeRow(1);\r\n/* 031 */     this.project_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(project_result, 32);\r\n/* 032 */     this.project_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(project_holder, 1);\r\n/* 033 */     this.project_arrayWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter();\r\n/* 034 */\r\n/* 035 */   }\r\n/* 036 */\r\n/* 037 */   protected void processNext() throws java.io.IOException {\r\n/* 038 */     while (inputadapter_input.hasNext()) {\r\n/* 039 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 040 */       double inputadapter_value = inputadapter_row.getDouble(0);\r\n/* 041 */\r\n/* 042 */       final boolean project_isNull = false;\r\n/* 043 */       this.project_values = new Object[2];\r\n/* 044 */       boolean project_isNull1 = false;\r\n/* 045 */\r\n/* 046 */       double project_value1 = -1.0;\r\n/* 047 */       project_value1 = inputadapter_value + 1.1D;\r\n/* 048 */       if (false) {\r\n/* 049 */         project_values[0] = null;\r\n/* 050 */       } else {\r\n/* 051 */         project_values[0] = project_value1;\r\n/* 052 */       }\r\n/* 053 */\r\n/* 054 */       boolean project_isNull4 = false;\r\n/* 055 */\r\n/* 056 */       double project_value4 = -1.0;\r\n/* 057 */       project_value4 = inputadapter_value + 2.2D;\r\n/* 058 */       if (false) {\r\n/* 059 */         project_values[1] = null;\r\n/* 060 */       } else {\r\n/* 061 */         project_values[1] = project_value4;\r\n/* 062 */       }\r\n/* 063 */\r\n/* 064 */       final ArrayData project_value = new org.apache.spark.sql.catalyst.util.GenericArrayData(project_values);\r\n/* 065 */       this.project_values = null;\r\n/* 066 */       project_holder.reset();\r\n/* 067 */\r\n/* 068 */       project_rowWriter.zeroOutNullBytes();\r\n/* 069 */\r\n/* 070 */       if (project_isNull) {\r\n/* 071 */         project_rowWriter.setNullAt(0);\r\n/* 072 */       } else {\r\n/* 073 */         // Remember the current cursor so that we can calculate how many bytes are\r\n/* 074 */         // written later.\r\n/* 075 */         final int project_tmpCursor = project_holder.cursor;\r\n/* 076 */\r\n/* 077 */         if (project_value instanceof UnsafeArrayData) {\r\n/* 078 */           final int project_sizeInBytes = ((UnsafeArrayData) project_value).getSizeInBytes();\r\n/* 079 */           // grow the global buffer before writing data.\r\n/* 080 */           project_holder.grow(project_sizeInBytes);\r\n/* 081 */           ((UnsafeArrayData) project_value).writeToMemory(project_holder.buffer, project_holder.cursor);\r\n/* 082 */           project_holder.cursor += project_sizeInBytes;\r\n/* 083 */\r\n/* 084 */         } else {\r\n/* 085 */           final int project_numElements = project_value.numElements();\r\n/* 086 */           project_arrayWriter.initialize(project_holder, project_numElements, 8);\r\n/* 087 */\r\n/* 088 */           for (int project_index = 0; project_index < project_numElements; project_index++) {\r\n/* 089 */             if (project_value.isNullAt(project_index)) {\r\n/* 090 */               project_arrayWriter.setNullDouble(project_index);\r\n/* 091 */             } else {\r\n/* 092 */               final double project_element = project_value.getDouble(project_index);\r\n/* 093 */               project_arrayWriter.write(project_index, project_element);\r\n/* 094 */             }\r\n/* 095 */           }\r\n/* 096 */         }\r\n/* 097 */\r\n/* 098 */         project_rowWriter.setOffsetAndSize(0, project_tmpCursor, project_holder.cursor - project_tmpCursor);\r\n/* 099 */       }\r\n/* 100 */       project_result.setTotalSize(project_holder.totalSize());\r\n/* 101 */       append(project_result);\r\n/* 102 */       if (shouldStop()) return;\r\n/* 103 */     }\r\n/* 104 */   }\r\n/* 105 */ }\r\n```\r\n\r\nGenerated code with this PR\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 */   private org.apache.spark.sql.catalyst.expressions.UnsafeArrayData project_value;\r\n/* 013 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder project_holder;\r\n/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter project_createArrayWriter;\r\n/* 015 */   private UnsafeRow project_result;\r\n/* 016 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder project_holder1;\r\n/* 017 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter project_rowWriter;\r\n/* 018 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter project_arrayWriter;\r\n/* 019 */\r\n/* 020 */   public GeneratedIterator(Object[] references) {\r\n/* 021 */     this.references = references;\r\n/* 022 */   }\r\n/* 023 */\r\n/* 024 */   public void init(int index, scala.collection.Iterator[] inputs) {\r\n/* 025 */     partitionIndex = index;\r\n/* 026 */     this.inputs = inputs;\r\n/* 027 */     inputadapter_input = inputs[0];\r\n/* 028 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 029 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 030 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 031 */     project_value = new org.apache.spark.sql.catalyst.expressions.UnsafeArrayData();\r\n/* 032 */     project_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(project_value, 2);\r\n/* 033 */     project_createArrayWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter();\r\n/* 034 */     project_result = new UnsafeRow(1);\r\n/* 035 */     this.project_holder1 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(project_result, 32);\r\n/* 036 */     this.project_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(project_holder1, 1);\r\n/* 037 */     this.project_arrayWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter();\r\n/* 038 */\r\n/* 039 */   }\r\n/* 040 */\r\n/* 041 */   protected void processNext() throws java.io.IOException {\r\n/* 042 */     while (inputadapter_input.hasNext()) {\r\n/* 043 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 044 */       double inputadapter_value = inputadapter_row.getDouble(0);\r\n/* 045 */\r\n/* 046 */       project_holder.reset();\r\n/* 047 */       project_createArrayWriter.initialize(project_holder, 2, 8);\r\n/* 048 */\r\n/* 049 */       boolean project_isNull1 = false;\r\n/* 050 */\r\n/* 051 */       double project_value1 = -1.0;\r\n/* 052 */       project_value1 = inputadapter_value + 1.1D;\r\n/* 053 */       project_createArrayWriter.write(0, project_value1);\r\n/* 054 */\r\n/* 055 */       boolean project_isNull4 = false;\r\n/* 056 */\r\n/* 057 */       double project_value4 = -1.0;\r\n/* 058 */       project_value4 = inputadapter_value + 2.2D;\r\n/* 059 */       project_createArrayWriter.write(1, project_value4);\r\n/* 060 */       project_holder1.reset();\r\n/* 061 */\r\n/* 062 */       // Remember the current cursor so that we can calculate how many bytes are\r\n/* 063 */       // written later.\r\n/* 064 */       final int project_tmpCursor = project_holder1.cursor;\r\n/* 065 */\r\n/* 066 */       if (project_value instanceof UnsafeArrayData) {\r\n/* 067 */         final int project_sizeInBytes = ((UnsafeArrayData) project_value).getSizeInBytes();\r\n/* 068 */         // grow the global buffer before writing data.\r\n/* 069 */         project_holder1.grow(project_sizeInBytes);\r\n/* 070 */         ((UnsafeArrayData) project_value).writeToMemory(project_holder1.buffer, project_holder1.cursor);\r\n/* 071 */         project_holder1.cursor += project_sizeInBytes;\r\n/* 072 */\r\n/* 073 */       } else {\r\n/* 074 */         final int project_numElements = project_value.numElements();\r\n/* 075 */         project_arrayWriter.initialize(project_holder1, project_numElements, 8);\r\n/* 076 */\r\n/* 077 */         for (int project_index = 0; project_index < project_numElements; project_index++) {\r\n/* 078 */           if (project_value.isNullAt(project_index)) {\r\n/* 079 */             project_arrayWriter.setNullDouble(project_index);\r\n/* 080 */           } else {\r\n/* 081 */             final double project_element = project_value.getDouble(project_index);\r\n/* 082 */             project_arrayWriter.write(project_index, project_element);\r\n/* 083 */           }\r\n/* 084 */         }\r\n/* 085 */       }\r\n/* 086 */\r\n/* 087 */       project_rowWriter.setOffsetAndSize(0, project_tmpCursor, project_holder1.cursor - project_tmpCursor);\r\n/* 088 */       project_result.setTotalSize(project_holder1.totalSize());\r\n/* 089 */       append(project_result);\r\n/* 090 */       if (shouldStop()) return;\r\n/* 091 */     }\r\n/* 092 */   }\r\n/* 093 */ }\r\n```\r\n## How was this patch tested?\r\n\r\nAdded unit tests into `DataFrameComplexTypeSuite`\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/269178806","html_url":"https://github.com/apache/spark/pull/13909#issuecomment-269178806","issue_url":"https://api.github.com/repos/apache/spark/issues/13909","id":269178806,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.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":"2016-12-26T07:49:29Z","updated_at":"2016-12-26T07:49:29Z","body":"**[Test build #70584 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70584/testReport)** for PR 13909 at commit [`69d5e33`](https://github.com/apache/spark/commit/69d5e33d2035fc5f6f4dfec65bde60c7dfc39548).\n * This patch **fails Spark unit tests**.\n * This patch merges cleanly.\n * This patch adds no public classes."}}
{"actor":{"display_login":"SparkQA"}}