{"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":8,"created_at":"2016-09-10T08:20:38Z","updated_at":"2016-10-05T19:11:20Z","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\n\r\nPerformance results of micro benchmarks will be avaiable.\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/* 001 */ public Object generate(Object[] references) {\r\n/* 002 */   return new GeneratedIterator(references);\r\n/* 003 */ }\r\n/* 004 */\r\n/* 005 */ /**\r\n * Codegend pipeline for\r\n * SerializeFromObject [newInstance(class org.apache.spark.sql.catalyst.util.GenericArrayData) AS value#7]\r\n * +- MapElements <function1>, obj#6: [I\r\n *    +- Scan ExternalRDDScan[obj#1]\r\n */\r\n/* 006 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 007 */   private Object[] references;\r\n/* 008 */   private scala.collection.Iterator inputadapter_input;\r\n/* 009 */   private UnsafeRow mapelements_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder mapelements_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter mapelements_rowWriter;\r\n/* 012 */   private boolean[] serializefromobject_argIsNulls;\r\n/* 013 */   private int[] serializefromobject_argValue;\r\n/* 014 */   private UnsafeRow serializefromobject_result;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 016 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\r\n/* 017 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter serializefromobject_arrayWriter;\r\n/* 018 */\r\n/* 019 */   public GeneratedIterator(Object[] references) {\r\n/* 020 */     this.references = references;\r\n/* 021 */   }\r\n/* 022 */\r\n/* 023 */   public void init(int index, scala.collection.Iterator inputs[]) {\r\n/* 024 */     partitionIndex = index;\r\n/* 025 */     inputadapter_input = inputs[0];\r\n/* 026 */     mapelements_result = new UnsafeRow(1);\r\n/* 027 */     this.mapelements_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(mapelements_result, 32);\r\n/* 028 */     this.mapelements_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(mapelements_holder, 1);\r\n/* 029 */     serializefromobject_argIsNulls = new boolean[1];\r\n/* 030 */\r\n/* 031 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 032 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 32);\r\n/* 033 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 034 */     this.serializefromobject_arrayWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter();\r\n/* 035 */   }\r\n/* 036 */\r\n/* 037 */   protected void processNext() throws java.io.IOException {\r\n/* 038 */     // PRODUCE: SerializeFromObject [newInstance(class org.apache.spark.sql.catalyst.util.GenericArrayData) AS value#7]\r\n/* 039 */     // PRODUCE: MapElements <function1>, obj#6: [I\r\n/* 040 */     // PRODUCE: InputAdapter\r\n/* 041 */     while (inputadapter_input.hasNext()) {\r\n/* 042 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 043 */       // CONSUME: MapElements <function1>, obj#6: [I\r\n/* 044 */       // input[0, [I, false]\r\n/* 045 */       int[] inputadapter_value = (int[])inputadapter_row.get(0, null);\r\n/* 046 */\r\n/* 047 */       // CONSUME: SerializeFromObject [newInstance(class org.apache.spark.sql.catalyst.util.GenericArrayData) AS value#7]\r\n/* 048 */       // <function1>.apply\r\n/* 049 */       Object mapelements_obj = ((Expression) references[0]).eval(null);\r\n/* 050 */       scala.Function1 mapelements_value1 = (scala.Function1) mapelements_obj;\r\n/* 051 */\r\n/* 052 */       boolean mapelements_isNull = false || false;\r\n/* 053 */       int[] mapelements_value = null;\r\n/* 054 */       if (!mapelements_isNull) {\r\n/* 055 */         Object mapelements_funcResult = null;\r\n/* 056 */         mapelements_funcResult = mapelements_value1.apply(inputadapter_value);\r\n/* 057 */         if (mapelements_funcResult == null) {\r\n/* 058 */           mapelements_isNull = true;\r\n/* 059 */         } else {\r\n/* 060 */           mapelements_value = (int[]) mapelements_funcResult;\r\n/* 061 */         }\r\n/* 062 */\r\n/* 063 */       }\r\n/* 064 */       mapelements_isNull = mapelements_value == null;\r\n/* 065 */\r\n/* 066 */       // CONSUME: WholeStageCodegen\r\n/* 067 */       // newInstance(class org.apache.spark.sql.catalyst.util.GenericArrayData)\r\n/* 068 */       serializefromobject_argIsNulls[0] = mapelements_isNull;\r\n/* 069 */       serializefromobject_argValue = mapelements_value;\r\n/* 070 */\r\n/* 071 */       boolean serializefromobject_isNull = false;\r\n/* 072 */       for (int idx = 0; idx < 1; idx++) {\r\n/* 073 */         if (serializefromobject_argIsNulls[idx]) { serializefromobject_isNull = true; break; }\r\n/* 074 */       }\r\n/* 075 */\r\n/* 076 */       final ArrayData serializefromobject_value = serializefromobject_isNull ? null : new org.apache.spark.sql.catalyst.util.GenericArrayData(serializefromobject_argValue);\r\n/* 077 */       serializefromobject_holder.reset();\r\n/* 078 */\r\n/* 079 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 080 */\r\n/* 081 */       if (serializefromobject_isNull) {\r\n/* 082 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 083 */       } else {\r\n/* 084 */         // Remember the current cursor so that we can calculate how many bytes are\r\n/* 085 */         // written later.\r\n/* 086 */         final int serializefromobject_tmpCursor = serializefromobject_holder.cursor;\r\n/* 087 */\r\n/* 088 */         if (serializefromobject_value instanceof UnsafeArrayData) {\r\n/* 089 */           final int serializefromobject_sizeInBytes = ((UnsafeArrayData) serializefromobject_value).getSizeInBytes();\r\n/* 090 */           // grow the global buffer before writing data.\r\n/* 091 */           serializefromobject_holder.grow(serializefromobject_sizeInBytes);\r\n/* 092 */           ((UnsafeArrayData) serializefromobject_value).writeToMemory(serializefromobject_holder.buffer, serializefromobject_holder.cursor);\r\n/* 093 */           serializefromobject_holder.cursor += serializefromobject_sizeInBytes;\r\n/* 094 */\r\n/* 095 */         } else {\r\n/* 096 */           final int serializefromobject_numElements = serializefromobject_value.numElements();\r\n/* 097 */           serializefromobject_arrayWriter.initialize(serializefromobject_holder, serializefromobject_numElements, 4);\r\n/* 098 */\r\n/* 099 */           for (int serializefromobject_index = 0; serializefromobject_index < serializefromobject_numElements; serializefromobject_index++) {\r\n/* 100 */             if (serializefromobject_value.isNullAt(serializefromobject_index)) {\r\n/* 101 */               serializefromobject_arrayWriter.setNullAt(serializefromobject_index);\r\n/* 102 */             } else {\r\n/* 103 */               final int serializefromobject_element = serializefromobject_value.getInt(serializefromobject_index);\r\n/* 104 */               serializefromobject_arrayWriter.write(serializefromobject_index, serializefromobject_element);\r\n/* 105 */             }\r\n/* 106 */           }\r\n/* 107 */         }\r\n/* 108 */\r\n/* 109 */         serializefromobject_rowWriter.setOffsetAndSize(0, serializefromobject_tmpCursor, serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 110 */         serializefromobject_rowWriter.alignToWords(serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 111 */       }\r\n/* 112 */       serializefromobject_result.setTotalSize(serializefromobject_holder.totalSize());\r\n/* 113 */       append(serializefromobject_result);\r\n/* 114 */       if (shouldStop()) return;\r\n/* 115 */     }\r\n/* 116 */   }\r\n/* 117 */ }\r\n````\r\n\r\n\r\nGenerated code after applying this PR\r\n````java\r\n/* 001 */ public Object generate(Object[] references) {\r\n/* 002 */   return new GeneratedIterator(references);\r\n/* 003 */ }\r\n/* 004 */\r\n/* 005 */ /**\r\n * Codegend pipeline for\r\n * SerializeFromObject [staticinvoke(class org.apache.spark.sql.catalyst.expressions.UnsafeArrayData, ArrayType(IntegerType,false), fromPrimitiveArray, input[0, [I, true], true) AS value#7]\r\n * +- MapElements <function1>, obj#6: [I\r\n *    +- Scan ExternalRDDScan[obj#1]\r\n */\r\n/* 006 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 007 */   private Object[] references;\r\n/* 008 */   private scala.collection.Iterator inputadapter_input;\r\n/* 009 */   private UnsafeRow mapelements_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder mapelements_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter mapelements_rowWriter;\r\n/* 012 */   private UnsafeRow serializefromobject_result;\r\n/* 013 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter serializefromobject_arrayWriter;\r\n/* 016 */\r\n/* 017 */   public GeneratedIterator(Object[] references) {\r\n/* 018 */     this.references = references;\r\n/* 019 */   }\r\n/* 020 */\r\n/* 021 */   public void init(int index, scala.collection.Iterator inputs[]) {\r\n/* 022 */     partitionIndex = index;\r\n/* 023 */     inputadapter_input = inputs[0];\r\n/* 024 */     mapelements_result = new UnsafeRow(1);\r\n/* 025 */     this.mapelements_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(mapelements_result, 32);\r\n/* 026 */     this.mapelements_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(mapelements_holder, 1);\r\n/* 027 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 028 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 32);\r\n/* 029 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\r\n/* 030 */     this.serializefromobject_arrayWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeArrayWriter();\r\n/* 031 */   }\r\n/* 032 */\r\n/* 035 */     // PRODUCE: MapElements <function1>, obj#6: [I\r\n/* 036 */     // PRODUCE: InputAdapter\r\n/* 037 */     while (inputadapter_input.hasNext()) {\r\n/* 038 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 039 */       // CONSUME: MapElements <function1>, obj#6: [I\r\n/* 040 */       // input[0, [I, false]\r\n/* 041 */       int[] inputadapter_value = (int[])inputadapter_row.get(0, null);\r\n/* 042 */\r\n/* 043 */       // CONSUME: SerializeFromObject [staticinvoke(class org.apache.spark.sql.catalyst.expressions.UnsafeArrayData, ArrayType(IntegerType,false), fromPrimitiveArray, input[0, [I, true], true) AS value#7]\r\n/* 044 */       // <function1>.apply\r\n/* 045 */       Object mapelements_obj = ((Expression) references[0]).eval(null);\r\n/* 046 */       scala.Function1 mapelements_value1 = (scala.Function1) mapelements_obj;\r\n/* 047 */\r\n/* 048 */       boolean mapelements_isNull = false || false;\r\n/* 049 */       int[] mapelements_value = null;\r\n/* 050 */       if (!mapelements_isNull) {\r\n/* 051 */         Object mapelements_funcResult = null;\r\n/* 052 */         mapelements_funcResult = mapelements_value1.apply(inputadapter_value);\r\n/* 053 */         if (mapelements_funcResult == null) {\r\n/* 054 */           mapelements_isNull = true;\r\n/* 055 */         } else {\r\n/* 056 */           mapelements_value = (int[]) mapelements_funcResult;\r\n/* 057 */         }\r\n/* 058 */\r\n/* 059 */       }\r\n/* 060 */       mapelements_isNull = mapelements_value == null;\r\n/* 061 */\r\n/* 062 */       // CONSUME: WholeStageCodegen\r\n/* 063 */       // staticinvoke(class org.apache.spark.sql.catalyst.expressions.UnsafeArrayData, ArrayType(IntegerType,false), fromPrimitiveArray, input[0, [I, true], true)\r\n/* 064 */       boolean serializefromobject_isNull = mapelements_isNull;\r\n/* 065 */       final ArrayData serializefromobject_value = serializefromobject_isNull ? null : org.apache.spark.sql.catalyst.expressions.UnsafeArrayData.fromPrimitiveArray(mapelements_value);\r\n/* 066 */       serializefromobject_isNull = serializefromobject_value == null;\r\n/* 067 */       serializefromobject_holder.reset();\r\n/* 068 */\r\n/* 069 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 070 */\r\n/* 071 */       if (serializefromobject_isNull) {\r\n/* 072 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 073 */       } else {\r\n/* 074 */         // Remember the current cursor so that we can calculate how many bytes are\r\n/* 075 */         // written later.\r\n/* 076 */         final int serializefromobject_tmpCursor = serializefromobject_holder.cursor;\r\n/* 077 */\r\n/* 078 */         if (serializefromobject_value instanceof UnsafeArrayData) {\r\n/* 079 */           final int serializefromobject_sizeInBytes = ((UnsafeArrayData) serializefromobject_value).getSizeInBytes();\r\n/* 080 */           // grow the global buffer before writing data.\r\n/* 081 */           serializefromobject_holder.grow(serializefromobject_sizeInBytes);\r\n/* 082 */           ((UnsafeArrayData) serializefromobject_value).writeToMemory(serializefromobject_holder.buffer, serializefromobject_holder.cursor);\r\n/* 083 */           serializefromobject_holder.cursor += serializefromobject_sizeInBytes;\r\n/* 084 */\r\n/* 085 */         } else {\r\n/* 086 */           final int serializefromobject_numElements = serializefromobject_value.numElements();\r\n/* 087 */           serializefromobject_arrayWriter.initialize(serializefromobject_holder, serializefromobject_numElements, 4);\r\n/* 088 */\r\n/* 089 */           for (int serializefromobject_index = 0; serializefromobject_index < serializefromobject_numElements; serializefromobject_index++) {\r\n/* 090 */             if (serializefromobject_value.isNullAt(serializefromobject_index)) {\r\n/* 091 */               serializefromobject_arrayWriter.setNullAt(serializefromobject_index);\r\n/* 092 */             } else {\r\n/* 093 */               final int serializefromobject_element = serializefromobject_value.getInt(serializefromobject_index);\r\n/* 094 */               serializefromobject_arrayWriter.write(serializefromobject_index, serializefromobject_element);\r\n/* 095 */             }\r\n/* 096 */           }\r\n/* 097 */         }\r\n/* 098 */\r\n/* 099 */         serializefromobject_rowWriter.setOffsetAndSize(0, serializefromobject_tmpCursor, serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 100 */         serializefromobject_rowWriter.alignToWords(serializefromobject_holder.cursor - serializefromobject_tmpCursor);\r\n/* 101 */       }\r\n/* 102 */       serializefromobject_result.setTotalSize(serializefromobject_holder.totalSize());\r\n/* 103 */       append(serializefromobject_result);\r\n/* 104 */       if (shouldStop()) return;\r\n/* 105 */     }\r\n/* 106 */   }\r\n/* 107 */ }\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/251770045","html_url":"https://github.com/apache/spark/pull/15044#issuecomment-251770045","issue_url":"https://api.github.com/repos/apache/spark/issues/15044","id":251770045,"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},"created_at":"2016-10-05T19:11:20Z","updated_at":"2016-10-05T19:11:20Z","body":"@hvanhovell this PR is ready for review. Which PR is preferable, #15044 or #13758?"}}
{"actor":{"display_login":"kiszk"}}