{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/18066","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/18066/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/18066/comments","events_url":"https://api.github.com/repos/apache/spark/issues/18066/events","html_url":"https://github.com/apache/spark/pull/18066","id":230589881,"number":18066,"title":"[SPARK-20822][SQL] Generate code to build table cache using ColumnarBatch and to get value from ColumnVector","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":2,"created_at":"2017-05-23T05:18:48Z","updated_at":"2017-05-23T09:45:27Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/18066","html_url":"https://github.com/apache/spark/pull/18066","diff_url":"https://github.com/apache/spark/pull/18066.diff","patch_url":"https://github.com/apache/spark/pull/18066.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR generates the following Java code\r\n1. Build each in-memory table cache using `ColumnarBatch` with `ColumnVector` instead of using CachedBatch with `Array[Byte]`.\r\n2. Get a value for a column in `ColumnVector without using an iterator\r\n\r\nAs the first step, for ease of review, I supported only integer and double data types with whole-stage codegen. Another PR will address an execution path without whole-stage codegen\r\n\r\nThis PR implements the follings:\r\n1. Keep a in-memory table cache using `ColumnarBatch` with `ColumnVector`. For supporting the new and coventional cache data structure, this PR declares `CachedBatch` as trait, and declares `CachedColumnarBatch` and `CachedBatchBytes` as actual implementations.\r\n2. Generate Java code to build a in-memory table cache.\r\n3. Generate Java code to directly get value from `ColumnVector`.\r\n\r\nThis PR improves runtime performance by\r\n1. build in-memory table cache by eliminating lots of virtual calls and complicated data path.\r\n2. eliminating data copy from column-oriented storage to `InternalRow` in a `SpecificColumnarIterator` iterator.\r\n\r\n\r\n**Options**\r\nA ColumnVector for all primitive data types in ColumnarBatch can be compressed. Currently, there are two ways to enable compression:\r\n\r\n1. Set true into a property `spark.sql.inMemoryColumnarStorage.compressed (default is true)`, or\r\n2. Call `DataFrame.persist(st)`, where st is `MEMORY_ONLY_SER`, `MEMORY_ONLY_SER_2`, `MEMORY_AND_DISK_SER`, or `MEMORY_AND_DISK_SER_2`.\r\n\r\n\r\n**an example program**\r\n```java\r\nval df = sparkContext.parallelize((1 to 10), 1).map(i => (i, i.toDouble)).toDF(\"i\", \"d\").cache\r\ndf.filter(\"i < 8 and 4.0 < d\").show\r\n```\r\n\r\n**Generated code for building a in-memory table cache**\r\n```\r\n/* 001 */ import scala.collection.Iterator;\r\n/* 002 */ import org.apache.spark.sql.types.DataType;\r\n/* 003 */ import org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder;\r\n/* 004 */ import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter;\r\n/* 005 */ import org.apache.spark.sql.execution.columnar.MutableUnsafeRow;\r\n/* 006 */ import org.apache.spark.sql.execution.vectorized.ColumnVector;\r\n/* 007 */\r\n/* 008 */ public SpecificColumnarIterator generate(Object[] references) {\r\n/* 009 */   return new SpecificColumnarIterator(references);\r\n/* 010 */ }\r\n/* 011 */\r\n/* 012 */ class SpecificColumnarIterator extends org.apache.spark.sql.execution.columnar.ColumnarIterator {\r\n/* 013 */   private ColumnVector[] colInstances;\r\n/* 014 */   private UnsafeRow unsafeRow = new UnsafeRow(0);\r\n/* 015 */   private BufferHolder bufferHolder = new BufferHolder(unsafeRow);\r\n/* 016 */   private UnsafeRowWriter rowWriter = new UnsafeRowWriter(bufferHolder, 0);\r\n/* 017 */   private MutableUnsafeRow mutableRow = null;\r\n/* 018 */\r\n/* 019 */   private int rowIdx = 0;\r\n/* 020 */   private int numRowsInBatch = 0;\r\n/* 021 */\r\n/* 022 */   private scala.collection.Iterator input = null;\r\n/* 023 */   private DataType[] columnTypes = null;\r\n/* 024 */   private int[] columnIndexes = null;\r\n/* 025 */\r\n/* 026 */\r\n/* 027 */\r\n/* 028 */   public SpecificColumnarIterator(Object[] references) {\r\n/* 029 */\r\n/* 030 */     this.mutableRow = new MutableUnsafeRow(rowWriter);\r\n/* 031 */   }\r\n/* 032 */\r\n/* 033 */   public void initialize(Iterator input, DataType[] columnTypes, int[] columnIndexes) {\r\n/* 034 */     this.input = input;\r\n/* 035 */     this.columnTypes = columnTypes;\r\n/* 036 */     this.columnIndexes = columnIndexes;\r\n/* 037 */   }\r\n/* 038 */\r\n/* 039 */\r\n/* 040 */\r\n/* 041 */   public boolean hasNext() {\r\n/* 042 */     if (rowIdx < numRowsInBatch) {\r\n/* 043 */       return true;\r\n/* 044 */     }\r\n/* 045 */     if (!input.hasNext()) {\r\n/* 046 */       return false;\r\n/* 047 */     }\r\n/* 048 */\r\n/* 049 */     org.apache.spark.sql.execution.columnar.CachedColumnarBatch cachedBatch =\r\n/* 050 */     (org.apache.spark.sql.execution.columnar.CachedColumnarBatch) input.next();\r\n/* 051 */     org.apache.spark.sql.execution.vectorized.ColumnarBatch batch = cachedBatch.columnarBatch();\r\n/* 052 */     rowIdx = 0;\r\n/* 053 */     numRowsInBatch = cachedBatch.getNumRows();\r\n/* 054 */     colInstances = new ColumnVector[columnIndexes.length];\r\n/* 055 */     for (int i = 0; i < columnIndexes.length; i ++) {\r\n/* 056 */       colInstances[i] = batch.column(columnIndexes[i]);\r\n/* 057 */     }\r\n/* 058 */\r\n/* 059 */     return hasNext();\r\n/* 060 */   }\r\n/* 061 */\r\n/* 062 */   public InternalRow next() {\r\n/* 063 */     bufferHolder.reset();\r\n/* 064 */     rowWriter.zeroOutNullBytes();\r\n/* 065 */\r\n/* 066 */     unsafeRow.setTotalSize(bufferHolder.totalSize());\r\n/* 067 */     rowIdx += 1;\r\n/* 068 */     return unsafeRow;\r\n/* 069 */   }\r\n/* 070 */ }\r\n```\r\n\r\n**Generated code by whole-stage codegen (lines 75-78 are major changes)**\r\n```\r\n/* 001 */ public Object generate(Object[] references) {\r\n/* 002 */   return new GeneratedIterator(references);\r\n/* 003 */ }\r\n/* 004 */\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 boolean agg_initAgg;\r\n/* 009 */   private boolean agg_bufIsNull;\r\n/* 010 */   private long agg_bufValue;\r\n/* 011 */   private scala.collection.Iterator inmemorytablescan_input;\r\n/* 012 */   private org.apache.spark.sql.execution.metric.SQLMetric inmemorytablescan_numOutputRows;\r\n/* 013 */   private org.apache.spark.sql.execution.metric.SQLMetric inmemorytablescan_scanTime;\r\n/* 014 */   private long inmemorytablescan_scanTime1;\r\n/* 015 */   private org.apache.spark.sql.execution.vectorized.ColumnarBatch inmemorytablescan_batch;\r\n/* 016 */   private int inmemorytablescan_batchIdx;\r\n/* 017 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inmemorytablescan_colInstance0;\r\n/* 018 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inmemorytablescan_colInstance1;\r\n/* 019 */   private UnsafeRow inmemorytablescan_result;\r\n/* 020 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder inmemorytablescan_holder;\r\n/* 021 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter inmemorytablescan_rowWriter;\r\n/* 022 */   private org.apache.spark.sql.execution.metric.SQLMetric filter_numOutputRows;\r\n/* 023 */   private UnsafeRow filter_result;\r\n/* 024 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder filter_holder;\r\n/* 025 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter filter_rowWriter;\r\n/* 026 */   private org.apache.spark.sql.execution.metric.SQLMetric agg_numOutputRows;\r\n/* 027 */   private org.apache.spark.sql.execution.metric.SQLMetric agg_aggTime;\r\n/* 028 */   private UnsafeRow agg_result;\r\n/* 029 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder;\r\n/* 030 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter;\r\n/* 031 */\r\n/* 032 */   public GeneratedIterator(Object[] references) {\r\n/* 033 */     this.references = references;\r\n/* 034 */   }\r\n/* 035 */\r\n/* 036 */   public void init(int index, scala.collection.Iterator[] inputs) {\r\n/* 037 */     partitionIndex = index;\r\n/* 038 */     this.inputs = inputs;\r\n/* 039 */     wholestagecodegen_init_0();\r\n/* 040 */     wholestagecodegen_init_1();\r\n/* 041 */\r\n/* 042 */   }\r\n/* 043 */\r\n/* 044 */   private void wholestagecodegen_init_0() {\r\n/* 045 */     agg_initAgg = false;\r\n/* 046 */\r\n/* 047 */     inmemorytablescan_input = inputs[0];\r\n/* 048 */     this.inmemorytablescan_numOutputRows = (org.apache.spark.sql.execution.metric.SQLMetric) references[0];\r\n/* 049 */     this.inmemorytablescan_scanTime = (org.apache.spark.sql.execution.metric.SQLMetric) references[1];\r\n/* 050 */     inmemorytablescan_scanTime1 = 0;\r\n/* 051 */     inmemorytablescan_batch = null;\r\n/* 052 */     inmemorytablescan_batchIdx = 0;\r\n/* 053 */     inmemorytablescan_colInstance0 = null;\r\n/* 054 */     inmemorytablescan_colInstance1 = null;\r\n/* 055 */     inmemorytablescan_result = new UnsafeRow(2);\r\n/* 056 */     this.inmemorytablescan_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(inmemorytablescan_result, 0);\r\n/* 057 */     this.inmemorytablescan_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(inmemorytablescan_holder, 2);\r\n/* 058 */     this.filter_numOutputRows = (org.apache.spark.sql.execution.metric.SQLMetric) references[2];\r\n/* 059 */     filter_result = new UnsafeRow(2);\r\n/* 060 */     this.filter_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(filter_result, 0);\r\n/* 061 */     this.filter_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(filter_holder, 2);\r\n/* 062 */\r\n/* 063 */   }\r\n/* 064 */\r\n/* 065 */   private void agg_doAggregateWithoutKey() throws java.io.IOException {\r\n/* 066 */     // initialize aggregation buffer\r\n/* 067 */     agg_bufIsNull = false;\r\n/* 068 */     agg_bufValue = 0L;\r\n/* 069 */\r\n/* 070 */     if (inmemorytablescan_batch == null) {\r\n/* 071 */       inmemorytablescan_nextBatch();\r\n/* 072 */     }\r\n/* 073 */     while (inmemorytablescan_batch != null) {\r\n/* 074 */       int inmemorytablescan_numRows = inmemorytablescan_batch.numRows();\r\n/* 075 */       int inmemorytablescan_localEnd = inmemorytablescan_numRows - inmemorytablescan_batchIdx;\r\n/* 076 */       for (int inmemorytablescan_localIdx = 0; inmemorytablescan_localIdx < inmemorytablescan_localEnd; inmemorytablescan_localIdx++) {\r\n/* 077 */         int inmemorytablescan_rowIdx = inmemorytablescan_batchIdx + inmemorytablescan_localIdx;\r\n/* 078 */         int inmemorytablescan_value = inmemorytablescan_colInstance0.getInt(inmemorytablescan_rowIdx);\r\n/* 079 */\r\n/* 080 */         boolean filter_isNull = false;\r\n/* 081 */\r\n/* 082 */         boolean filter_value = false;\r\n/* 083 */         filter_value = inmemorytablescan_value > 4;\r\n/* 084 */         if (!filter_value) continue;\r\n/* 085 */         double inmemorytablescan_value1 = inmemorytablescan_colInstance1.getDouble(inmemorytablescan_rowIdx);\r\n/* 086 */\r\n/* 087 */         boolean filter_isNull3 = false;\r\n/* 088 */\r\n/* 089 */         boolean filter_value3 = false;\r\n/* 090 */         filter_value3 = org.apache.spark.util.Utils.nanSafeCompareDoubles(inmemorytablescan_value1, 10.0D) > 0;\r\n/* 091 */         if (!filter_value3) continue;\r\n/* 092 */\r\n/* 093 */         filter_numOutputRows.add(1);\r\n/* 094 */\r\n/* 095 */         // do aggregate\r\n/* 096 */         // common sub-expressions\r\n/* 097 */\r\n/* 098 */         // evaluate aggregate function\r\n/* 099 */         boolean agg_isNull1 = false;\r\n/* 100 */\r\n/* 101 */         long agg_value1 = -1L;\r\n/* 102 */         agg_value1 = agg_bufValue + 1L;\r\n/* 103 */         // update aggregation buffer\r\n/* 104 */         agg_bufIsNull = false;\r\n/* 105 */         agg_bufValue = agg_value1;\r\n/* 106 */         // shouldStop check is eliminated\r\n/* 107 */       }\r\n/* 108 */       inmemorytablescan_batchIdx = inmemorytablescan_numRows;\r\n/* 109 */       inmemorytablescan_batch = null;\r\n/* 110 */       inmemorytablescan_nextBatch();\r\n/* 111 */     }\r\n/* 112 */     inmemorytablescan_scanTime.add(inmemorytablescan_scanTime1 / (1000 * 1000));\r\n/* 113 */     inmemorytablescan_scanTime1 = 0;\r\n/* 114 */\r\n/* 115 */   }\r\n/* 116 */\r\n/* 117 */   private void inmemorytablescan_nextBatch() throws java.io.IOException {\r\n/* 118 */     long getBatchStart = System.nanoTime();\r\n/* 119 */     if (inmemorytablescan_input.hasNext()) {\r\n/* 120 */       inmemorytablescan_batch = (org.apache.spark.sql.execution.vectorized.ColumnarBatch)inmemorytablescan_input.next();\r\n/* 121 */       inmemorytablescan_numOutputRows.add(inmemorytablescan_batch.numRows());\r\n/* 122 */       inmemorytablescan_batchIdx = 0;\r\n/* 123 */       inmemorytablescan_colInstance0 = inmemorytablescan_batch.column(0);\r\n/* 124 */       inmemorytablescan_colInstance1 = inmemorytablescan_batch.column(1);\r\n/* 125 */\r\n/* 126 */     }\r\n/* 127 */     inmemorytablescan_scanTime1 += System.nanoTime() - getBatchStart;\r\n/* 128 */   }\r\n/* 129 */\r\n/* 130 */   private void wholestagecodegen_init_1() {\r\n/* 131 */     this.agg_numOutputRows = (org.apache.spark.sql.execution.metric.SQLMetric) references[3];\r\n/* 132 */     this.agg_aggTime = (org.apache.spark.sql.execution.metric.SQLMetric) references[4];\r\n/* 133 */     agg_result = new UnsafeRow(1);\r\n/* 134 */     this.agg_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result, 0);\r\n/* 135 */     this.agg_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder, 1);\r\n/* 136 */\r\n/* 137 */   }\r\n/* 138 */\r\n/* 139 */   protected void processNext() throws java.io.IOException {\r\n/* 140 */     while (!agg_initAgg) {\r\n/* 141 */       agg_initAgg = true;\r\n/* 142 */       long agg_beforeAgg = System.nanoTime();\r\n/* 143 */       agg_doAggregateWithoutKey();\r\n/* 144 */       agg_aggTime.add((System.nanoTime() - agg_beforeAgg) / 1000000);\r\n/* 145 */\r\n/* 146 */       // output the result\r\n/* 147 */\r\n/* 148 */       agg_numOutputRows.add(1);\r\n/* 149 */       agg_rowWriter.zeroOutNullBytes();\r\n/* 150 */\r\n/* 151 */       if (agg_bufIsNull) {\r\n/* 152 */         agg_rowWriter.setNullAt(0);\r\n/* 153 */       } else {\r\n/* 154 */         agg_rowWriter.write(0, agg_bufValue);\r\n/* 155 */       }\r\n/* 156 */       append(agg_result);\r\n/* 157 */     }\r\n/* 158 */   }\r\n/* 159 */ }\r\n```\r\n\r\n## How was this patch tested?\r\n\r\nAdd test suites for wider columns"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/303347515","html_url":"https://github.com/apache/spark/pull/18066#issuecomment-303347515","issue_url":"https://api.github.com/repos/apache/spark/issues/18066","id":303347515,"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-05-23T09:45:27Z","updated_at":"2017-05-23T09:45:27Z","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}