{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/17172","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/17172/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/17172/comments","events_url":"https://api.github.com/repos/apache/spark/issues/17172/events","html_url":"https://github.com/apache/spark/pull/17172","id":212037326,"number":17172,"title":"[SPARK-19008][SQL] Improve performance of Dataset.map by eliminating boxing/unboxing","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":8,"created_at":"2017-03-06T06:46:17Z","updated_at":"2017-03-08T12:15:45Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/17172","html_url":"https://github.com/apache/spark/pull/17172","diff_url":"https://github.com/apache/spark/pull/17172.diff","patch_url":"https://github.com/apache/spark/pull/17172.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR improve performance of Dataset.map() for primitive types by removing boxing/unbox operations. This is based on [the discussion](https://github.com/apache/spark/pull/16391#discussion_r93788919) with @cloud-fan.\r\n\r\nCurrent Catalyst generates a method call to a `apply()` method of an anonymous function written in Scala. The types of an argument and return value are `java.lang.Object`. As a result, each method call for a primitive value involves a pair of unboxing and boxing for calling this `apply()` method and a pair of boxing and unboxing for returning from this `apply()` method.\r\n\r\nThis PR directly calls a specialized version of a `apply()` method without boxing and unboxing. For example, if types of an arguments ant return value is `int`, this PR generates a method call to `apply$mcII$sp`. This PR supports any combination of `Int`, `Long`, `Float`, and `Double`.\r\n\r\n\r\nThe following is a benchmark result using [this program](https://github.com/apache/spark/pull/16391/files) with 4.7x. Here is a Dataset part of this program.\r\n\r\nWithout this PR\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_111-8u111-b14-2ubuntu0.16.04.2-b14 on Linux 4.4.0-47-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nback-to-back map:                        Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nRDD                                           1923 / 1952         52.0          19.2       1.0X\r\nDataFrame                                      526 /  548        190.2           5.3       3.7X\r\nDataset                                       3094 / 3154         32.3          30.9       0.6X\r\n```\r\n\r\nWith this PR\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_111-8u111-b14-2ubuntu0.16.04.2-b14 on Linux 4.4.0-47-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nback-to-back map:                        Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nRDD                                           1883 / 1892         53.1          18.8       1.0X\r\nDataFrame                                      502 /  642        199.1           5.0       3.7X\r\nDataset                                        657 /  784        152.2           6.6       2.9X\r\n```\r\n\r\n```java\r\n  def backToBackMap(spark: SparkSession, numRows: Long, numChains: Int): Benchmark = {\r\n    import spark.implicits._\r\n    val rdd = spark.sparkContext.range(0, numRows)\r\n    val ds = spark.range(0, numRows)\r\n    val func = (l: Long) => l + 1\r\n    val benchmark = new Benchmark(\"back-to-back map\", numRows)\r\n...\r\n    benchmark.addCase(\"Dataset\") { iter =>\r\n      var res = ds.as[Long]\r\n      var i = 0\r\n      while (i < numChains) {\r\n        res = res.map(func)\r\n        i += 1\r\n      }\r\n      res.queryExecution.toRdd.foreach(_ => Unit)\r\n    }\r\n    benchmark\r\n  }\r\n```\r\n\r\n\r\nA motivating example\r\n```java\r\nSeq(1, 2, 3).toDS.map(i => i * 7).show\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 deserializetoobject_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder deserializetoobject_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter deserializetoobject_rowWriter;\r\n/* 012 */   private int mapelements_argValue;\r\n/* 013 */   private UnsafeRow mapelements_result;\r\n/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder mapelements_holder;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter mapelements_rowWriter;\r\n/* 016 */   private UnsafeRow serializefromobject_result;\r\n/* 017 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 018 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\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 */     deserializetoobject_result = new UnsafeRow(1);\r\n/* 029 */     this.deserializetoobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(deserializetoobject_result, 0);\r\n/* 030 */     this.deserializetoobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(deserializetoobject_holder, 1);\r\n/* 031 */\r\n/* 032 */     mapelements_result = new UnsafeRow(1);\r\n/* 033 */     this.mapelements_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(mapelements_result, 0);\r\n/* 034 */     this.mapelements_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(mapelements_holder, 1);\r\n/* 035 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 036 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 037 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\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() && !stopEarly()) {\r\n/* 043 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 044 */       int inputadapter_value = inputadapter_row.getInt(0);\r\n/* 045 */\r\n/* 046 */       boolean mapelements_isNull = true;\r\n/* 047 */       int mapelements_value = -1;\r\n/* 048 */       if (!false) {\r\n/* 049 */         mapelements_argValue = inputadapter_value;\r\n/* 050 */\r\n/* 051 */         mapelements_isNull = false;\r\n/* 052 */         if (!mapelements_isNull) {\r\n/* 053 */           Object mapelements_funcResult = null;\r\n/* 054 */           mapelements_funcResult = ((scala.Function1) references[0]).apply(mapelements_argValue);\r\n/* 055 */           if (mapelements_funcResult == null) {\r\n/* 056 */             mapelements_isNull = true;\r\n/* 057 */           } else {\r\n/* 058 */             mapelements_value = (Integer) mapelements_funcResult;\r\n/* 059 */           }\r\n/* 060 */\r\n/* 061 */         }\r\n/* 062 */\r\n/* 063 */       }\r\n/* 064 */\r\n/* 065 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 066 */\r\n/* 067 */       if (mapelements_isNull) {\r\n/* 068 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 069 */       } else {\r\n/* 070 */         serializefromobject_rowWriter.write(0, mapelements_value);\r\n/* 071 */       }\r\n/* 072 */       append(serializefromobject_result);\r\n/* 073 */       if (shouldStop()) return;\r\n/* 074 */     }\r\n/* 075 */   }\r\n/* 076 */ }\r\n```\r\n\r\nGenerated code with this PR (lines 48-56 are changed)\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 deserializetoobject_result;\r\n/* 010 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder deserializetoobject_holder;\r\n/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter deserializetoobject_rowWriter;\r\n/* 012 */   private int mapelements_argValue;\r\n/* 013 */   private UnsafeRow mapelements_result;\r\n/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder mapelements_holder;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter mapelements_rowWriter;\r\n/* 016 */   private UnsafeRow serializefromobject_result;\r\n/* 017 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder serializefromobject_holder;\r\n/* 018 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter serializefromobject_rowWriter;\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 */     deserializetoobject_result = new UnsafeRow(1);\r\n/* 029 */     this.deserializetoobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(deserializetoobject_result, 0);\r\n/* 030 */     this.deserializetoobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(deserializetoobject_holder, 1);\r\n/* 031 */\r\n/* 032 */     mapelements_result = new UnsafeRow(1);\r\n/* 033 */     this.mapelements_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(mapelements_result, 0);\r\n/* 034 */     this.mapelements_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(mapelements_holder, 1);\r\n/* 035 */     serializefromobject_result = new UnsafeRow(1);\r\n/* 036 */     this.serializefromobject_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(serializefromobject_result, 0);\r\n/* 037 */     this.serializefromobject_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(serializefromobject_holder, 1);\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() && !stopEarly()) {\r\n/* 043 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 044 */       int inputadapter_value = inputadapter_row.getInt(0);\r\n/* 045 */\r\n/* 046 */       boolean mapelements_isNull = true;\r\n/* 047 */       int mapelements_value = -1;\r\n/* 048 */       if (!false) {\r\n/* 049 */         mapelements_argValue = inputadapter_value;\r\n/* 050 */\r\n/* 051 */         mapelements_isNull = false;\r\n/* 052 */         if (!mapelements_isNull) {\r\n/* 053 */           mapelements_value = ((scala.Function1) references[0]).apply$mcII$sp(mapelements_argValue);\r\n/* 054 */         }\r\n/* 055 */\r\n/* 056 */       }\r\n/* 057 */\r\n/* 058 */       serializefromobject_rowWriter.zeroOutNullBytes();\r\n/* 059 */\r\n/* 060 */       if (mapelements_isNull) {\r\n/* 061 */         serializefromobject_rowWriter.setNullAt(0);\r\n/* 062 */       } else {\r\n/* 063 */         serializefromobject_rowWriter.write(0, mapelements_value);\r\n/* 064 */       }\r\n/* 065 */       append(serializefromobject_result);\r\n/* 066 */       if (shouldStop()) return;\r\n/* 067 */     }\r\n/* 068 */   }\r\n/* 069 */ }\r\n```\r\n\r\nJava bytecode for methods for `i => i * 7`\r\n```java\r\n$ javap -c Test\\$\\$anonfun\\$5\\$\\$anonfun\\$apply\\$mcV\\$sp\\$1.class\r\nCompiled from \"Test.scala\"\r\npublic final class org.apache.spark.sql.Test$$anonfun$5$$anonfun$apply$mcV$sp$1 extends scala.runtime.AbstractFunction1$mcII$sp implements scala.Serializable {\r\n  public static final long serialVersionUID;\r\n\r\n  public final int apply(int);\r\n    Code:\r\n       0: aload_0\r\n       1: iload_1\r\n       2: invokevirtual #18                 // Method apply$mcII$sp:(I)I\r\n       5: ireturn\r\n\r\n  public int apply$mcII$sp(int);\r\n    Code:\r\n       0: iload_1\r\n       1: bipush        7\r\n       3: imul\r\n       4: ireturn\r\n\r\n  public final java.lang.Object apply(java.lang.Object);\r\n    Code:\r\n       0: aload_0\r\n       1: aload_1\r\n       2: invokestatic  #29                 // Method scala/runtime/BoxesRunTime.unboxToInt:(Ljava/lang/Object;)I\r\n       5: invokevirtual #31                 // Method apply:(I)I\r\n       8: invokestatic  #35                 // Method scala/runtime/BoxesRunTime.boxToInteger:(I)Ljava/lang/Integer;\r\n      11: areturn\r\n\r\n  public org.apache.spark.sql.Test$$anonfun$5$$anonfun$apply$mcV$sp$1(org.apache.spark.sql.Test$$anonfun$5);\r\n    Code:\r\n       0: aload_0\r\n       1: invokespecial #42                 // Method scala/runtime/AbstractFunction1$mcII$sp.\"<init>\":()V\r\n       4: return\r\n}\r\n```\r\n## How was this patch tested?\r\n\r\nAdded new test suites to `DatasetPrimitiveSuite`.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/285025894","html_url":"https://github.com/apache/spark/pull/17172#issuecomment-285025894","issue_url":"https://api.github.com/repos/apache/spark/issues/17172","id":285025894,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars0.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":"2017-03-08T12:15:45Z","updated_at":"2017-03-08T12:15:45Z","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}