{"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 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,"milestone":null,"comments":28,"created_at":"2016-03-25T09:02:00Z","updated_at":"2016-04-21T12:10:58Z","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 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. In some cases, data is compressed. This PR avoid conversion from column-oriented storage to row-oriented storage.\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 may have compressed data. If the array is compressed, decompress it and pass it to generated code. If the array is not compressed, it 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 1.9x - 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\n    Java HotSpot(TM) 64-Bit Server VM 1.8.0_66-b17 on Linux 2.6.32-504.el6.x86_64\r\n    Intel(R) Xeon(R) CPU E5-2667 v2 @ 3.30GHz\r\n\r\n    Int Sum with PassThrough cache:     Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       462 /  466         68.1          14.7       1.0X\r\n    ColumnVector codegen                       94 /  100        336.3           3.0       4.9X\r\n\r\n    Int Sum with RunLength cache:       Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       492 /  553         63.9          15.7       1.0X\r\n    ColumnVector codegen                      175 /  180        179.5           5.6       2.8X\r\n\r\n    Int Sum with Dictionary cache:      Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       599 /  610         52.5          19.0       1.0X\r\n    ColumnVector codegen                      320 /  327         98.3          10.2       1.9X\r\n\r\n    Int Sum with Delta cache:           Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       467 /  512         67.4          14.8       1.0X\r\n    ColumnVector codegen                      247 /  250        127.4           7.8       1.9X\r\n\r\n    Long Sum with PassThrough cache:    Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       382 /  420         41.2          24.3       1.0X\r\n    ColumnVector codegen                       89 /  101        176.2           5.7       4.3X\r\n\r\n    Float Sum with PassThrough cache:   Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       476 /  483         66.1          15.1       1.0X\r\n    ColumnVector codegen                       91 /  103        343.8           2.9       5.2X\r\n\r\n    Double Sum with PassThrough cache:  Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n    -------------------------------------------------------------------------------------------\r\n    InternalRow codegen                       290 /  306         54.3          18.4       1.0X\r\n    ColumnVector codegen                       95 /  101        165.7           6.0       3.1X\r\n````\r\n\r\n\r\nMotivating example:\r\n\r\n````\r\n    val df = sc.parallelize(0 to 9, 1).toDF()\r\n    df.cache().filter(\"value <= 5\").show()\r\n````\r\n\r\nGenerated code\r\n````java\r\n/* 005 */ /** Codegened pipeline for:\r\n/* 006 */ * Filter (value#1 <= 5)\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 org.apache.spark.sql.execution.columnar.CachedBatch columnar_batch;\r\n/* 019 */   private scala.collection.Iterator columnar_iterator;\r\n/* 020 */   private int columnar_batchIdx;\r\n/* 021 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inputadapter_col0;\r\n/* 022 */   private UnsafeRow inputadapter_result;\r\n/* 023 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder inputadapter_holder;\r\n/* 024 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter inputadapter_rowWriter;\r\n/* 025 */   private org.apache.spark.sql.execution.metric.LongSQLMetric filter_numOutputRows1;\r\n/* 026 */   private org.apache.spark.sql.execution.metric.LongSQLMetricValue filter_metricValue1;\r\n/* 027 */   private UnsafeRow filter_result1;\r\n/* 028 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder filter_holder1;\r\n/* 029 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter filter_rowWriter1;\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_batch = null;\r\n/* 045 */     columnar_iterator = null;\r\n/* 046 */     columnar_batchIdx = 0;\r\n/* 047 */     inputadapter_col0 = null;\r\n/* 048 */     inputadapter_result = new UnsafeRow(1);\r\n/* 049 */     this.inputadapter_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(inputadapter_result, 0);\r\n/* 050 */     this.inputadapter_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(inputadapter_holder, 1);\r\n/* 051 */     this.filter_numOutputRows1 = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];\r\n/* 052 */     filter_metricValue1 = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) filter_numOutputRows1.localValue();\r\n/* 053 */     filter_result1 = new UnsafeRow(1);\r\n/* 054 */     this.filter_holder1 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(filter_result1, 0);\r\n/* 055 */     this.filter_rowWriter1 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(filter_holder1, 1);\r\n/* 056 */   }\r\n/* 057 */\r\n/* 058 */   private void processBatch() throws java.io.IOException {\r\n/* 059 */     /*** PRODUCE: Filter (value#1 <= 5) */\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 */         if (columnar_iterator.hasNext()) {\r\n/* 066 */           columnar_batch = (org.apache.spark.sql.execution.columnar.CachedBatch)(columnar_iterator.next());\r\n/* 067 */         } else {\r\n/* 068 */           cleanup();\r\n/* 069 */           break;\r\n/* 070 */         }\r\n/* 071 */         inputadapter_col0 = columnar_batch.column(((org.apache.spark.sql.execution.columnar.ColumnarIterator)inputadapter_input1).getColumnIndexes()[0\r\n]);\r\n/* 072 */       }\r\n/* 073 */\r\n/* 074 */       int inputadapter_numRows = columnar_batch.numRows();\r\n/* 075 */       while (columnar_batchIdx < inputadapter_numRows) {\r\n/* 076 */         int inputadapter_rowIdx = columnar_batchIdx++;\r\n/* 077 */         /*** CONSUME: Filter (value#1 <= 5) */\r\n/* 078 */\r\n/* 079 */         /* columnVector[inputadapter_col0, inputadapter_rowIdx, int] */\r\n/* 080 */         int inputadapter_value1 = inputadapter_col0.getInt(inputadapter_rowIdx);\r\n/* 081 */\r\n/* 082 */         /* (input[0, int] <= 5) */\r\n/* 083 */         boolean filter_value4 = false;\r\n/* 084 */         filter_value4 = inputadapter_value1 <= 5;\r\n/* 085 */         if (!filter_value4) continue;\r\n/* 086 */\r\n/* 087 */         filter_metricValue1.add(1);\r\n/* 088 */\r\n/* 089 */         /*** CONSUME: WholeStageCodegen */\r\n/* 090 */\r\n/* 091 */         filter_rowWriter1.write(0, inputadapter_value1);\r\n/* 092 */         append(filter_result1);\r\n/* 093 */         if (shouldStop()) return;\r\n/* 094 */       }\r\n/* 095 */       columnar_batchIdx = 0;\r\n/* 096 */     }\r\n/* 097 */   }\r\n/* 098 */\r\n/* 099 */   private void processRow() throws java.io.IOException {\r\n/* 100 */     /*** PRODUCE: Filter (value#1 <= 5) */\r\n/* 101 */\r\n/* 102 */     /*** PRODUCE: INPUT */\r\n/* 103 */\r\n/* 104 */     while (inputadapter_input.hasNext()) {\r\n/* 105 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();\r\n/* 106 */       /*** CONSUME: Filter (value#1 <= 5) */\r\n/* 107 */\r\n/* 108 */       /* input[0, int] */\r\n/* 109 */       int inputadapter_value = inputadapter_row.getInt(0);\r\n/* 110 */\r\n/* 111 */       /* (input[0, int] <= 5) */\r\n/* 112 */       boolean filter_value = false;\r\n/* 113 */       filter_value = inputadapter_value <= 5;\r\n/* 114 */       if (!filter_value) continue;\r\n/* 115 */\r\n/* 116 */       filter_metricValue.add(1);\r\n/* 117 */\r\n/* 118 */       /*** CONSUME: WholeStageCodegen */\r\n/* 119 */\r\n/* 120 */       filter_rowWriter.write(0, inputadapter_value);\r\n/* 121 */       append(filter_result);\r\n/* 122 */       if (shouldStop()) return;\r\n/* 123 */     }\r\n/* 124 */   }\r\n/* 125 */\r\n/* 126 */   private void cleanup() {\r\n/* 127 */     columnar_batch = null;\r\n/* 128 */     columnar_iterator = null;\r\n/* 129 */\r\n/* 130 */     inputadapter_col0 = null;\r\n/* 131 */\r\n/* 132 */   }\r\n/* 133 */\r\n/* 134 */   protected void processNext() throws java.io.IOException {\r\n/* 135 */     org.apache.spark.sql.execution.columnar.ColumnarIterator wholestagecodegen_columnItr = null;\r\n/* 136 */     if (columnar_batch != null) {\r\n/* 137 */       wholestagecodegen_columnItr = (org.apache.spark.sql.execution.columnar.ColumnarIterator)inputadapter_input1;\r\n/* 138 */       columnar_iterator = wholestagecodegen_columnItr.getInput();\r\n/* 139 */       processBatch();\r\n/* 140 */     } else if (inputadapter_input1 instanceof org.apache.spark.sql.execution.columnar.ColumnarIterator &&\r\n/* 141 */       ((wholestagecodegen_columnItr = (org.apache.spark.sql.execution.columnar.ColumnarIterator)inputadapter_input1).isSupportColumnarCodeGen())) {\r\n/* 142 */       columnar_batchIdx = 0;\r\n/* 143 */       columnar_iterator = wholestagecodegen_columnItr.getInput();\r\n/* 144 */       processBatch();\r\n/* 145 */     } else {\r\n/* 146 */       processRow();\r\n/* 147 */     }\r\n/* 148 */   }\r\n/* 149 */ }\r\n````\r\n\r\n## How was this patch tested?\r\n\r\nTested existing test suites\r\nwill add test suites for operations to dataframe generated by df.cache().\r\n\r\n\r\n(If this patch involves UI changes, please attach a screenshot; otherwise, remove this)\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/212886240","html_url":"https://github.com/apache/spark/pull/11956#issuecomment-212886240","issue_url":"https://api.github.com/repos/apache/spark/issues/11956","id":212886240,"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-04-21T12:10:58Z","updated_at":"2016-04-21T12:10:58Z","body":"Test FAILed.\nRefer to this link for build results (access rights to CI server needed): \nhttps://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/56526/\nTest FAILed."}}