{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15044","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15044/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15044/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15044/events","html_url":"https://github.com/apache/spark/pull/15044","id":176167878,"number":15044,"title":"[SQL][SPARK-17490] Optimize SerializeFromObject() for a primitive array","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":13,"created_at":"2016-09-10T08:20:38Z","updated_at":"2016-10-06T18:39:05Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15044","html_url":"https://github.com/apache/spark/pull/15044","diff_url":"https://github.com/apache/spark/pull/15044.diff","patch_url":"https://github.com/apache/spark/pull/15044.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nWaiting for merging #13680\r\n\r\nThis PR optimizes `SerializeFromObject()` for an primitive array. This is derived from #13758 to address one of problems by using a simple way in #13758.\r\n\r\nThe current implementation always generates `GenericArrayData` from `SerializeFromObject()` for any type of an array in a logical plan. This involves a boxing at a constructor of `GenericArrayData` when `SerializedFromObject()` has an primitive array.\r\n\r\nThis PR enables to generate `UnsafeArrayData` from `SerializeFromObject()` for a primitive array. It can avoid boxing to create an instance of `ArrayData` in the generated code by Catalyst.\r\n\r\nPerformance improvement is up to 2.2x\r\n```\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\n\r\nWithout this PR\r\nWrite an array in Dataset:               Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nInt                                            469 /  544         17.9          55.9       1.0X\r\nDouble                                        1113 / 1137          7.5         132.7       0.4X\r\n\r\nwith this PR\r\nWrite an array in Dataset:               Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nInt                                            280 /  302         30.0          33.4       1.0X\r\nDouble                                         503 /  519         16.7          60.0       0.6X\r\n```\r\n\r\n\r\nHere is an example program that will happen in mllib as described in [SPARK-16070](https://issues.apache.org/jira/browse/SPARK-16070).\r\n\r\n```\r\nsparkContext.parallelize(Seq(Array(1, 2)), 1).toDS.map(e => e).show\r\n```\r\n\r\nGenerated code before applying this PR\r\n````java\r\n/* 039 */   protected void processNext() throws java.io.IOException {\r\n/* 040 */     while (inputadapter_input.hasNext()) {\r\n/* 041 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 042 */       int[] inputadapter_value = (int[])inputadapter_row.get(0, null);\r\n/* 043 */\r\n/* 044 */       Object mapelements_obj = ((Expression) references[0]).eval(null);\r\n/* 045 */       scala.Function1 mapelements_value1 = (scala.Function1) mapelements_obj;\r\n/* 046 */\r\n/* 047 */       boolean mapelements_isNull = false || false;\r\n/* 048 */       int[] mapelements_value = null;\r\n/* 049 */       if (!mapelements_isNull) {\r\n/* 050 */         Object mapelements_funcResult = null;\r\n/* 051 */         mapelements_funcResult = mapelements_value1.apply(inputadapter_value);\r\n/* 052 */         if (mapelements_funcResult == null) {\r\n/* 053 */           mapelements_isNull = true;\r\n/* 054 */         } else {\r\n/* 055 */           mapelements_value = (int[]) mapelements_funcResult;\r\n/* 056 */         }\r\n/* 057 */\r\n/* 058 */       }\r\n/* 059 */       mapelements_isNull = mapelements_value == null;\r\n/* 060 */\r\n/* 061 */       serializefromobject_argIsNulls[0] = mapelements_isNull;\r\n/* 062 */       serializefromobject_argValue = mapelements_value;\r\n/* 063 */\r\n/* 064 */       boolean serializefromobject_isNull = false;\r\n/* 065 */       for (int idx = 0; idx < 1; idx++) {\r\n/* 066 */         if (serializefromobject_argIsNulls[idx]) { serializefromobject_isNull = true; break; }\r\n/* 067 */       }\r\n/* 068 */\r\n/* 069 */       final ArrayData serializefromobject_value = serializefromobject_isNull ? null : new org.apache.spark.sql.catalyst.util.GenericArrayData(serializefromobject_argValue);\r\n/* 070 */       serializefromobject_holder.reset();\r\n/* 071 */\r\n/* 072 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 073 */\r\n/* 074 */       if (serializefromobject_isNull) {\r\n/* 075 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 076 */       } else {\r\n/* 077 */         // Remember the current cursor so that we can calculate how many bytes are\r\n/* 078 */         // written later.\r\n/* 079 */         final int serializefromobject_tmpCursor = serializefromobject_holder.cursor;\r\n/* 080 */\r\n/* 081 */         if (serializefromobject_value instanceof UnsafeArrayData) {\r\n/* 082 */           final int serializefromobject_sizeInBytes = ((UnsafeArrayData) serializefromobject_value).getSizeInBytes();\r\n/* 083 */           // grow the global buffer before writing data.\r\n/* 084 */           serializefromobject_holder.grow(serializefromobject_sizeInBytes);\r\n/* 085 */           ((UnsafeArrayData) serializefromobject_value).writeToMemory(serializefromobject_holder.buffer, serializefromobject_holder.cursor);\r\n/* 086 */           serializefromobject_holder.cursor += serializefromobject_sizeInBytes;\r\n/* 087 */\r\n/* 088 */         } else {\r\n/* 089 */           final int serializefromobject_numElements = serializefromobject_value.numElements();\r\n/* 090 */           serializefromobject_arrayWriter.initialize(serializefromobject_holder, serializefromobject_numElements, 4);\r\n/* 091 */\r\n/* 092 */           for (int serializefromobject_index = 0; serializefromobject_index < serializefromobject_numElements; serializefromobject_index++) {\r\n/* 093 */             if (serializefromobject_value.isNullAt(serializefromobject_index)) {\r\n/* 094 */               serializefromobject_arrayWriter.setNullInt(serializefromobject_index);\r\n/* 095 */             } else {\r\n/* 096 */               final int serializefromobject_element = serializefromobject_value.getInt(serializefromobject_index);\r\n/* 097 */               serializefromobject_arrayWriter.write(serializefromobject_index, serializefromobject_element);\r\n/* 098 */             }\r\n/* 099 */           }\r\n/* 100 */         }\r\n/* 101 */\r\n/* 102 */         serializefromobject_rowWriter.setOffsetAndSize(0, serializefromobject_tmpCursor, serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 103 */       }\r\n/* 104 */       serializefromobject_result.setTotalSize(serializefromobject_holder.totalSize());\r\n/* 105 */       append(serializefromobject_result);\r\n/* 106 */       if (shouldStop()) return;\r\n/* 107 */     }\r\n/* 108 */   }\r\n/* 109 */ }\r\n````\r\n\r\n\r\nGenerated code after applying this PR\r\n````java\r\n/* 035 */   protected void processNext() throws java.io.IOException {\r\n/* 036 */     while (inputadapter_input.hasNext()) {\r\n/* 037 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 038 */       int[] inputadapter_value = (int[])inputadapter_row.get(0, null);\r\n/* 039 */\r\n/* 040 */       Object mapelements_obj = ((Expression) references[0]).eval(null);\r\n/* 041 */       scala.Function1 mapelements_value1 = (scala.Function1) mapelements_obj;\r\n/* 042 */\r\n/* 043 */       boolean mapelements_isNull = false || false;\r\n/* 044 */       int[] mapelements_value = null;\r\n/* 045 */       if (!mapelements_isNull) {\r\n/* 046 */         Object mapelements_funcResult = null;\r\n/* 047 */         mapelements_funcResult = mapelements_value1.apply(inputadapter_value);\r\n/* 048 */         if (mapelements_funcResult == null) {\r\n/* 049 */           mapelements_isNull = true;\r\n/* 050 */         } else {\r\n/* 051 */           mapelements_value = (int[]) mapelements_funcResult;\r\n/* 052 */         }\r\n/* 053 */\r\n/* 054 */       }\r\n/* 055 */       mapelements_isNull = mapelements_value == null;\r\n/* 056 */\r\n/* 057 */       boolean serializefromobject_isNull = mapelements_isNull;\r\n/* 058 */       final ArrayData serializefromobject_value = serializefromobject_isNull ? null : org.apache.spark.sql.catalyst.expressions.UnsafeArrayData.fromPrimitiveArray(mapelements_value);\r\n/* 059 */       serializefromobject_isNull = serializefromobject_value == null;\r\n/* 060 */       serializefromobject_holder.reset();\r\n/* 061 */\r\n/* 062 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 063 */\r\n/* 064 */       if (serializefromobject_isNull) {\r\n/* 065 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 066 */       } else {\r\n/* 067 */         // Remember the current cursor so that we can calculate how many bytes are\r\n/* 068 */         // written later.\r\n/* 069 */         final int serializefromobject_tmpCursor = serializefromobject_holder.cursor;\r\n/* 070 */\r\n/* 071 */         if (serializefromobject_value instanceof UnsafeArrayData) {\r\n/* 072 */           final int serializefromobject_sizeInBytes = ((UnsafeArrayData) serializefromobject_value).getSizeInBytes();\r\n/* 073 */           // grow the global buffer before writing data.\r\n/* 074 */           serializefromobject_holder.grow(serializefromobject_sizeInBytes);\r\n/* 075 */           ((UnsafeArrayData) serializefromobject_value).writeToMemory(serializefromobject_holder.buffer, serializefromobject_holder.cursor);\r\n/* 076 */           serializefromobject_holder.cursor += serializefromobject_sizeInBytes;\r\n/* 077 */\r\n/* 078 */         } else {\r\n/* 079 */           final int serializefromobject_numElements = serializefromobject_value.numElements();\r\n/* 080 */           serializefromobject_arrayWriter.initialize(serializefromobject_holder, serializefromobject_numElements, 4);\r\n/* 081 */\r\n/* 082 */           for (int serializefromobject_index = 0; serializefromobject_index < serializefromobject_numElements; serializefromobject_index++) {\r\n/* 083 */             if (serializefromobject_value.isNullAt(serializefromobject_index)) {\r\n/* 084 */               serializefromobject_arrayWriter.setNullInt(serializefromobject_index);\r\n/* 085 */             } else {\r\n/* 086 */               final int serializefromobject_element = serializefromobject_value.getInt(serializefromobject_index);\r\n/* 087 */               serializefromobject_arrayWriter.write(serializefromobject_index, serializefromobject_element);\r\n/* 088 */             }\r\n/* 089 */           }\r\n/* 090 */         }\r\n/* 091 */\r\n/* 092 */         serializefromobject_rowWriter.setOffsetAndSize(0, serializefromobject_tmpCursor, serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 093 */       }\r\n/* 094 */       serializefromobject_result.setTotalSize(serializefromobject_holder.totalSize());\r\n/* 095 */       append(serializefromobject_result);\r\n/* 096 */       if (shouldStop()) return;\r\n/* 097 */     }\r\n/* 098 */   }\r\n/* 099 */ }\r\n````\r\n\r\n## How was this patch tested?\r\n\r\nAdded a test in `DatasetSuite`\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/252051573","html_url":"https://github.com/apache/spark/pull/15044#issuecomment-252051573","issue_url":"https://api.github.com/repos/apache/spark/issues/15044","id":252051573,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2016-10-06T18:39:05Z","updated_at":"2016-10-06T18:39:05Z","body":"Test FAILed.\nRefer to this link for build results (access rights to CI server needed): \nhttps://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66453/\nTest FAILed."}}
{"actor":{"display_login":"AmplabJenkins"}}