{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/11956","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/11956/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/11956/comments","events_url":"https://api.github.com/repos/apache/spark/issues/11956/events","html_url":"https://github.com/apache/spark/pull/11956","id":143460199,"number":11956,"title":"[SPARK-14098][SQL] Generate Java code that gets a float/double value in each column of CachedBatch when DataFrame.cache() is called","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":59,"created_at":"2016-03-25T09:02:00Z","updated_at":"2016-06-29T11:45:02Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/11956","html_url":"https://github.com/apache/spark/pull/11956","diff_url":"https://github.com/apache/spark/pull/11956.diff","patch_url":"https://github.com/apache/spark/pull/11956.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR generates Java code to get a float/double value of each column from CachedBatch when DataFrame.cache() is called. This is done in whole stage code generation.\r\n\r\nWhen DataFrame.cache() is called, data is stored as column-oriented storage (columnar cache) in CachedBatch. This PR avoid conversion from column-oriented storage to row-oriented storage.\r\n\r\nThis PR handles only float and double that are stored in a column without compression. Another PR will handle other primitive types that may be stored in a column in a compressed format. This is for ease of review by reducing the size of PR\r\n\r\nThis PR consists of two parts.\r\n\r\n1. Pass data in CachedBatch to generated code by using ```decompress()``` method. CachedBatch consists of multiple ByteBuffer arrays. A ByteBuffer is just passed to generated code.\r\n\r\n2. Generate code both for row-oriented storage and column-oriented storage only if\r\n - InMemoryColumnarTableScan exists in a plan sub-tree. A decision is performed by checking an given iterator is ColumnaIterator at runtime\r\n - Sort or join does not exist in a plan sub-tree. \r\n\r\nThis PR generates Java code for columnar cache only if types in all columns, which are accessed in operations, are primitive\r\n\r\nThis PR improves performance of aggregate sum by 3.8x - 5.2x. This benchmark is available at [here](https://github.com/kiszk/spark/blob/SPARK-14098/sql/core/src/test/scala/org/apache/spark/sql/DataFrameCacheBenchmark.scala)\r\n\r\nPerformance results:\r\n````\r\nJava HotSpot(TM) 64-Bit Server VM 1.8.0_66-b17 on Linux 2.6.32-504.el6.x86_64\r\nIntel(R) Xeon(R) CPU E5-2667 v2 @ 3.30GHz\r\n\r\nRunning benchmark: Double Sum with PassThrough cache\r\n  Running case: InternalRow codegen\r\n  Running case: ColumnVector codegen\r\n\r\nDouble Sum with PassThrough cache:  Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n-------------------------------------------------------------------------------------------\r\nInternalRow codegen                       334 /  342         47.1          21.2       1.0X\r\nColumnVector codegen                       88 /   99        179.1           5.6       3.8X\r\n\r\nRunning benchmark: Float Sum with PassThrough cache\r\n  Running case: InternalRow codegen\r\n  Running case: ColumnVector codegen\r\n\r\nFloat Sum with PassThrough cache:   Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n-------------------------------------------------------------------------------------------\r\nInternalRow codegen                       473 /  485         66.5          15.0       1.0X\r\nColumnVector codegen                       91 /   95        345.6           2.9       5.2X\r\n````\r\n\r\n\r\nMotivating example:\r\n\r\n````\r\nval df = sc.parallelize(0 to 9, 1).map(i => i.toFloat).toDF()\r\ndf.cache().filter(\"value <= 5.1\").show()\r\n````\r\n\r\nGenerated code\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 */ /** Codegened pipeline for:\r\n/* 006 */ * Filter (value#1 <= 5.1)\r\n/* 007 */ +- INPUT\r\n/* 008 */ */\r\n/* 009 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 010 */   private Object[] references;\r\n/* 011 */   private scala.collection.Iterator inputadapter_input;\r\n/* 012 */   private org.apache.spark.sql.execution.metric.LongSQLMetric filter_numOutputRows;\r\n/* 013 */   private org.apache.spark.sql.execution.metric.LongSQLMetricValue filter_metricValue;\r\n/* 014 */   private UnsafeRow filter_result;\r\n/* 015 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder filter_holder;\r\n/* 016 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter filter_rowWriter;\r\n/* 017 */   private scala.collection.Iterator inputadapter_input1;\r\n/* 018 */   private int columnar_batchIdx;\r\n/* 019 */   private int columnar_numRows;\r\n/* 020 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inputadapter_col0;\r\n/* 021 */   private UnsafeRow inputadapter_result;\r\n/* 022 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder inputadapter_holder;\r\n/* 023 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter inputadapter_rowWriter;\r\n/* 024 */   private org.apache.spark.sql.execution.metric.LongSQLMetric filter_numOutputRows1;\r\n/* 025 */   private org.apache.spark.sql.execution.metric.LongSQLMetricValue filter_metricValue1;\r\n/* 026 */   private UnsafeRow filter_result1;\r\n/* 027 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder filter_holder1;\r\n/* 028 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter filter_rowWriter1;\r\n/* 029 */   private org.apache.spark.sql.execution.columnar.ColumnarIterator columnar_itr;\r\n/* 030 */\r\n/* 031 */   public GeneratedIterator(Object[] references) {\r\n/* 032 */     this.references = references;\r\n/* 033 */   }\r\n/* 034 */\r\n/* 035 */   public void init(int index, scala.collection.Iterator inputs[]) {\r\n/* 036 */     partitionIndex = index;\r\n/* 037 */     inputadapter_input = inputs[0];\r\n/* 038 */     this.filter_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[0];\r\n/* 039 */     filter_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) filter_numOutputRows.localValue();\r\n/* 040 */     filter_result = new UnsafeRow(1);\r\n/* 041 */     this.filter_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(filter_result, 0);\r\n/* 042 */     this.filter_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(filter_holder, 1);\r\n/* 043 */     inputadapter_input1 = inputs[0];\r\n/* 044 */     columnar_batchIdx = 0;\r\n/* 045 */     columnar_numRows = 0;\r\n/* 046 */     inputadapter_col0 = null;\r\n/* 047 */     inputadapter_result = new UnsafeRow(1);\r\n/* 048 */     this.inputadapter_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(inputadapter_result, 0);\r\n/* 049 */     this.inputadapter_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(inputadapter_holder, 1);\r\n/* 050 */     this.filter_numOutputRows1 = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];\r\n/* 051 */     filter_metricValue1 = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) filter_numOutputRows1.localValue();\r\n/* 052 */     filter_result1 = new UnsafeRow(1);\r\n/* 053 */     this.filter_holder1 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(filter_result1, 0);\r\n/* 054 */     this.filter_rowWriter1 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(filter_holder1, 1);\r\n/* 055 */     columnar_itr = null;\r\n/* 056 */   }\r\n/* 057 */\r\n/* 058 */   private void processBatch() throws java.io.IOException {\r\n/* 059 */     /*** PRODUCE: Filter (value#1 <= 5.1) */\r\n/* 060 */\r\n/* 061 */     /*** PRODUCE: INPUT */\r\n/* 062 */\r\n/* 063 */     while (true) {\r\n/* 064 */       if (columnar_batchIdx == 0) {\r\n/* 065 */         columnar_numRows = columnar_itr.initForColumnar();\r\n/* 066 */         if (columnar_numRows < 0) {\r\n/* 067 */           cleanup();\r\n/* 068 */           break;\r\n/* 069 */         }\r\n/* 070 */         inputadapter_col0 = columnar_itr.getColumn(0);\r\n/* 071 */       }\r\n/* 072 */\r\n/* 073 */       while (columnar_batchIdx < columnar_numRows) {\r\n/* 074 */         int inputadapter_rowIdx = columnar_batchIdx++;\r\n/* 075 */         /*** CONSUME: Filter (value#1 <= 5.1) */\r\n/* 076 */\r\n/* 077 */         /* columnVector[inputadapter_col0, inputadapter_rowIdx, float] */\r\n/* 078 */         float inputadapter_value1 = inputadapter_col0.getFloat(inputadapter_rowIdx);\r\n/* 079 */\r\n/* 080 */         /* (input[0, float] <= 5.1) */\r\n/* 081 */         boolean filter_value4 = false;\r\n/* 082 */         filter_value4 = inputadapter_value1 <= 5.1;\r\n/* 083 */         if (!filter_value4) continue;\r\n/* 084 */\r\n/* 085 */         filter_metricValue1.add(1);\r\n/* 086 */\r\n/* 087 */         /*** CONSUME: WholeStageCodegen */\r\n/* 088 */\r\n/* 089 */         filter_rowWriter1.write(0, inputadapter_value1);\r\n/* 090 */         append(filter_result1);\r\n/* 091 */         if (shouldStop()) return;\r\n/* 092 */       }\r\n/* 093 */       columnar_batchIdx = 0;\r\n/* 094 */     }\r\n/* 095 */   }\r\n/* 096 */\r\n/* 097 */   private void processRow() throws java.io.IOException {\r\n/* 098 */     /*** PRODUCE: Filter (value#1 <= 5.1) */\r\n/* 099 */\r\n/* 100 */     /*** PRODUCE: INPUT */\r\n/* 101 */\r\n/* 102 */     while (inputadapter_input.hasNext()) {\r\n/* 103 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 104 */       /*** CONSUME: Filter (value#1 <= 5.1) */\r\n/* 105 */\r\n/* 106 */       /* input[0, float] */\r\n/* 107 */       float inputadapter_value = inputadapter_row.getFloat(0);\r\n/* 108 */\r\n/* 109 */       /* (input[0, float] <= 5.1) */\r\n/* 110 */       boolean filter_value = false;\r\n/* 111 */       filter_value = inputadapter_value <= 5.1;\r\n/* 112 */       if (!filter_value) continue;\r\n/* 112 */       if (!filter_value) continue;\r\n/* 113 */\r\n/* 114 */       filter_metricValue.add(1);\r\n/* 115 */\r\n/* 116 */       /*** CONSUME: WholeStageCodegen */\r\n/* 117 */\r\n/* 118 */       filter_rowWriter.write(0, inputadapter_value);\r\n/* 119 */       append(filter_result);\r\n/* 120 */       if (shouldStop()) return;\r\n/* 121 */     }\r\n/* 122 */   }\r\n/* 123 */\r\n/* 124 */   private void cleanup() {\r\n/* 125 */     inputadapter_col0 = null;\r\n/* 126 */\r\n/* 127 */     columnar_itr = null;\r\n/* 128 */   }\r\n/* 129 */\r\n/* 130 */   protected void processNext() throws java.io.IOException {\r\n/* 131 */     if ((columnar_batchIdx != 0) ||\r\n/* 132 */       (inputadapter_input1 instanceof org.apache.spark.sql.execution.columnar.ColumnarIterator &&\r\n/* 133 */         (columnar_itr = (org.apache.spark.sql.execution.columnar.ColumnarIterator)inputadapter_input1).isSupportColumnarCodeGen())) {\r\n/* 134 */       processBatch();\r\n/* 135 */     } else {\r\n/* 136 */       processRow();\r\n/* 137 */     }\r\n/* 138 */   }\r\n/* 139 */ }\r\n````\r\n\r\n## How was this patch tested?\r\n\r\nTested existing test suites\r\nadded test suites for operations to dataframe generated by df.cache().\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/229332663","html_url":"https://github.com/apache/spark/pull/11956#issuecomment-229332663","issue_url":"https://api.github.com/repos/apache/spark/issues/11956","id":229332663,"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-06-29T11:45:02Z","updated_at":"2016-06-29T11:45:02Z","body":"Test PASSed.\nRefer to this link for build results (access rights to CI server needed): \nhttps://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/61460/\nTest PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}