{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15219","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15219/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15219/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15219/events","html_url":"https://github.com/apache/spark/pull/15219","id":178944688,"number":15219,"title":"[SPARK-14098][SQL] Generate Java code to build CachedColumnarBatch and get values from CachedColumnarBatch 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":39,"created_at":"2016-09-23T18:35:05Z","updated_at":"2016-12-01T17:56:00Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15219","html_url":"https://github.com/apache/spark/pull/15219","diff_url":"https://github.com/apache/spark/pull/15219.diff","patch_url":"https://github.com/apache/spark/pull/15219.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\n**[Here](https://docs.google.com/document/d/1-2BnW5ibuHIeQzmHEGIGkEcuMUCTk87pmPis2DKRg-Q/edit?usp=sharing) is a design document for this change.** This PR is derived from https://github.com/apache/spark/pull/11956 and https://github.com/apache/spark/pull/13899.\r\n\r\n**I am splitting this PR into multiple smaller child PRs (https://github.com/apache/spark/pull/15462, https://github.com/apache/spark/pull/15467, https://github.com/apache/spark/pull/15468)  for ease of review.**\r\n\r\n\r\nThis PR implements a new in-memory cache feature used by `DataFrame.cache` and `Dataset.cache`. The followings are basic design of this PR with suggestions from @davies.  \r\n\r\n1. Use `ColumnarBatch` with `ColumnVector` that are common data representations for columnar storage\r\n2. Use common compression scheme (e.g. lz4) for all of the data types in a `ColumnVector`\r\n3. Generate code that is simple and specialized for each in-memory cache to build an in-memory cache\r\n4. Generate code that directly reads data from `ColumnVector` for the in-memory cache by whole-stage codegen.\r\n\r\nBased on this design, this PR generates two kinds of Java code for `DataFrame.cache()`/`Dataset.cache()`\r\n1. Generate Java code to build `CachedColumnarBatch`, which keeps data in `ColumnarBatch`\r\n2. Generate Java code to get a value of each column from `ColumnarBatch`\r\n-a Get a value directly from from `ColumnarBatch` in code generated by whole stage code gen (primary path)\r\n-b Get a value thru an iterator if whole stage code gen is disabled (e.g. # of columns is more than 100, as backup path)\r\n\r\n\r\nAdvantage of this PR improves runtime performance and refactors to use common components.\r\nFor performance, this PR eliminates lots of virtual calls and data copy in old code paths. In particular, for 2-a., this PR avoids data copy from a columnar storage `CachedBatch` to a row-based iterator.\r\nFor refactoring, the following PR may remove unused components for `CachedBatchBytes`.\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\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\nThe compression scheme is specified by a property `spark.sql.inMemoryColumnarStorage.compression.codec` (default is `lz4`).\r\n\r\n**Performance results**\r\n3.4x for building a `CachedColumnarBatch` and getting values (in 1 and 2-a) \r\n\r\n**ToDo: the following PR performs compression for complex data type such as array**\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_91-b14 on Linux 4.4.11-200.fc22.x86_64\r\nIntel Xeon E3-12xx v2 (Ivy Bridge)\r\nCache random keys:                       Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\ncache = T columnarBatch = F compress = T      7211 / 7366          2.3         429.8       1.0X\r\ncache = T columnarBatch = F compress = F      2381 / 2460          7.0         141.9       3.0X\r\ncache = F                                      137 /  140        122.7           8.1      52.7X\r\ncache = T columnarBatch = T compress = T      2109 / 2252          8.0         125.7       3.4X\r\ncache = T columnarBatch = T compress = F      1126 / 1184         14.9          67.1       6.4X\r\n```\r\n\r\n1.2x for getting values (only in 2-a)\r\n\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_91-b14 on Linux 4.4.11-200.fc22.x86_64\r\nIntel Xeon E3-12xx v2 (Ivy Bridge)\r\nCache random keys:                       Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\ncache = T columnarBatch = F compress = T      1615 / 1655         41.5          24.1       1.0X\r\ncache = T columnarBatch = F compress = F      1603 / 1690         41.9          23.9       1.0X\r\ncache = F                                      444 /  449        151.3           6.6       3.6X\r\ncache = T columnarBatch = T compress = T      1404 / 1526         47.8          20.9       1.2X\r\ncache = T columnarBatch = T compress = F       116 /  125        579.0           1.7      13.9X\r\n```\r\n\r\n**Here is an example program**\r\n\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\n1. Generated code for building `CachedColumnarBatch`\r\n\r\n``` java\r\n/* 001 */ import org.apache.spark.memory.MemoryMode;\r\n/* 002 */ import org.apache.spark.sql.catalyst.InternalRow;\r\n/* 003 */ import org.apache.spark.sql.execution.columnar.CachedColumnarBatch;\r\n/* 004 */ import org.apache.spark.sql.execution.columnar.GenerateColumnarBatch;\r\n/* 005 */ import org.apache.spark.sql.execution.vectorized.ColumnarBatch;\r\n/* 006 */ import org.apache.spark.sql.execution.vectorized.ColumnVector;\r\n/* 007 */ import org.apache.spark.sql.execution.vectorized.OnHeapUnsafeColumnVector;\r\n/* 008 */\r\n/* 009 */ public GeneratedColumnarBatchIterator generate(Object[] references) {\r\n/* 010 */   return new GeneratedColumnarBatchIterator(references);\r\n/* 011 */ }\r\n/* 012 */\r\n/* 013 */ class GeneratedColumnarBatchIterator extends org.apache.spark.sql.execution.columnar.ColumnarBatchIterator {\r\n/* 014 */   private long bytesInBatch;\r\n/* 015 */   private java.util.Iterator rowIterator;\r\n/* 016 */   private org.apache.spark.sql.types.StructType schema;\r\n/* 017 */   private org.apache.spark.sql.execution.columnar.IntColumnStats colStat0;\r\n/* 018 */   private org.apache.spark.sql.execution.columnar.DoubleColumnStats colStat1;\r\n/* 019 */   private org.apache.spark.SparkConf conf;\r\n/* 020 */\r\n/* 021 */   public GeneratedColumnarBatchIterator(Object[] references) {\r\n/* 022 */     bytesInBatch = 0;\r\n/* 023 */     this.rowIterator = (java.util.Iterator) references[0];\r\n/* 024 */     this.schema = (org.apache.spark.sql.types.StructType) references[1];\r\n/* 025 */\r\n/* 026 */\r\n/* 027 */     this.conf = (org.apache.spark.SparkConf) references[2];\r\n/* 028 */   }\r\n/* 029 */\r\n/* 030 */\r\n/* 031 */\r\n/* 032 */   org.apache.spark.sql.execution.columnar.ColumnStats[] statsArray = new org.apache.spark.sql.execution.columnar.ColumnStats[2];\r\n/* 033 */   private void allocateColumnStats() {\r\n/* 034 */     InternalRow row = null;\r\n/* 035 */     colStat0 = new org.apache.spark.sql.execution.columnar.IntColumnStats(); statsArray[0] = colStat0;\r\n/* 036 */     colStat1 = new org.apache.spark.sql.execution.columnar.DoubleColumnStats(); statsArray[1] = colStat1;\r\n/* 037 */   }\r\n/* 038 */\r\n/* 039 */   @Override\r\n/* 040 */   public boolean hasNext() {\r\n/* 041 */     return rowIterator.hasNext();\r\n/* 042 */   }\r\n/* 043 */\r\n/* 044 */   @Override\r\n/* 045 */   public CachedColumnarBatch next() {\r\n/* 046 */     ColumnarBatch columnarBatch =\r\n/* 047 */     ColumnarBatch.allocate(schema, MemoryMode.ON_HEAP_UNSAFE, 10000);\r\n/* 048 */     allocateColumnStats();\r\n/* 049 */     int rowNum = 0;\r\n/* 050 */     bytesInBatch = 0;\r\n/* 051 */     while (rowIterator.hasNext() && rowNum < 10000 && bytesInBatch < 4194304) {\r\n/* 052 */       InternalRow row = (InternalRow) rowIterator.next();\r\n/* 053 */       if (row.isNullAt(0)) {\r\n/* 054 */         columnarBatch.column(0).putNull(rowNum);\r\n/* 055 */         colStat0.gatherNullStats();\r\n/* 056 */       } else {\r\n/* 057 */         int val = row.getInt(0);\r\n/* 058 */         columnarBatch.column(0).putInt(rowNum, val);\r\n/* 059 */         bytesInBatch += 4;\r\n/* 060 */         colStat0.gatherValueStats(val);\r\n/* 061 */       }\r\n/* 062 */       if (row.isNullAt(1)) {\r\n/* 063 */         columnarBatch.column(1).putNull(rowNum);\r\n/* 064 */         colStat1.gatherNullStats();\r\n/* 065 */       } else {\r\n/* 066 */         double val = row.getDouble(1);\r\n/* 067 */         columnarBatch.column(1).putDouble(rowNum, val);\r\n/* 068 */         bytesInBatch += 8;\r\n/* 069 */         colStat1.gatherValueStats(val);\r\n/* 070 */       }\r\n/* 071 */\r\n/* 072 */       rowNum += 1;\r\n/* 073 */     }\r\n/* 074 */     columnarBatch.setNumRows(rowNum);\r\n/* 075 */     for (int i = 0; i < 2; i++) {\r\n/* 076 */       ((OnHeapUnsafeColumnVector)columnarBatch.column(i)).compress(conf);\r\n/* 077 */     }\r\n/* 078 */     return CachedColumnarBatch.apply(\r\n/* 079 */       columnarBatch, GenerateColumnarBatch.generateStats(statsArray));\r\n/* 080 */   }\r\n/* 081 */ }\r\n```\r\n\r\n2.a Generated code for getting values from `CachedColumnarBatch` in code generated by whole stage codegen (primarypath)\r\n\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 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {\r\n/* 006 */   private Object[] references;\r\n/* 007 */   private scala.collection.Iterator inmemorytablescan_input;\r\n/* 008 */   private org.apache.spark.sql.execution.metric.SQLMetric inmemorytablescan_numOutputRows;\r\n/* 009 */   private org.apache.spark.sql.execution.metric.SQLMetric inmemorytablescan_scanTime;\r\n/* 010 */   private long inmemorytablescan_scanTime1;\r\n/* 011 */   private org.apache.spark.sql.execution.vectorized.ColumnarBatch inmemorytablescan_batch;\r\n/* 012 */   private org.apache.spark.SparkConf inmemorytablescan_conf;\r\n/* 013 */   private int inmemorytablescan_batchIdx;\r\n/* 014 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inmemorytablescan_colInstance0;\r\n/* 015 */   private org.apache.spark.sql.execution.vectorized.ColumnVector inmemorytablescan_colInstance1;\r\n/* 016 */   private UnsafeRow inmemorytablescan_result;\r\n/* 017 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder inmemorytablescan_holder;\r\n/* 018 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter inmemorytablescan_rowWriter;\r\n/* 019 */   private org.apache.spark.sql.execution.metric.SQLMetric filter_numOutputRows;\r\n/* 020 */   private UnsafeRow filter_result;\r\n/* 021 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder filter_holder;\r\n/* 022 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter filter_rowWriter;\r\n/* 023 */\r\n/* 024 */   public GeneratedIterator(Object[] references) {\r\n/* 025 */     this.references = references;\r\n/* 026 */   }\r\n/* 027 */\r\n/* 028 */   public void init(int index, scala.collection.Iterator inputs[]) {\r\n/* 029 */     partitionIndex = index;\r\n/* 030 */     inmemorytablescan_input = inputs[0];\r\n/* 031 */     this.inmemorytablescan_numOutputRows = (org.apache.spark.sql.execution.metric.SQLMetric) references[0];\r\n/* 032 */     this.inmemorytablescan_scanTime = (org.apache.spark.sql.execution.metric.SQLMetric) references[1];\r\n/* 033 */     inmemorytablescan_scanTime1 = 0;\r\n/* 034 */     inmemorytablescan_batch = null;\r\n/* 035 */     this.inmemorytablescan_conf = (org.apache.spark.SparkConf) references[2];\r\n/* 036 */     inmemorytablescan_batchIdx = 0;\r\n/* 037 */     inmemorytablescan_colInstance0 = null;\r\n/* 038 */     inmemorytablescan_colInstance1 = null;\r\n/* 039 */     inmemorytablescan_result = new UnsafeRow(2);\r\n/* 040 */     this.inmemorytablescan_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(inmemorytablescan_result, 0);\r\n/* 041 */     this.inmemorytablescan_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(inmemorytablescan_holder, 2);\r\n/* 042 */     this.filter_numOutputRows = (org.apache.spark.sql.execution.metric.SQLMetric) references[3];\r\n/* 043 */     filter_result = new UnsafeRow(2);\r\n/* 044 */     this.filter_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(filter_result, 0);\r\n/* 045 */     this.filter_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(filter_holder, 2);\r\n/* 046 */   }\r\n/* 047 */\r\n/* 048 */   private void inmemorytablescan_nextBatch() throws java.io.IOException {\r\n/* 049 */     long getBatchStart = System.nanoTime();\r\n/* 050 */     if (inmemorytablescan_input.hasNext()) {\r\n/* 051 */       inmemorytablescan_batch = (org.apache.spark.sql.execution.vectorized.ColumnarBatch)inmemorytablescan_input.next();\r\n/* 052 */\r\n/* 053 */       inmemorytablescan_numOutputRows.add(inmemorytablescan_batch.numRows());\r\n/* 054 */       inmemorytablescan_batchIdx = 0;\r\n/* 055 */       inmemorytablescan_colInstance0 = inmemorytablescan_batch.column(0); ((org.apache.spark.sql.execution.vectorized.OnHeapUnsafeColumnVector)inmemorytablescan_colInstance0).decompress(inmemorytablescan_conf);\r\n/* 056 */       inmemorytablescan_colInstance1 = inmemorytablescan_batch.column(1); ((org.apache.spark.sql.execution.vectorized.OnHeapUnsafeColumnVector)inmemorytablescan_colInstance1).decompress(inmemorytablescan_conf);\r\n/* 057 */\r\n/* 058 */     }\r\n/* 059 */     inmemorytablescan_scanTime1 += System.nanoTime() - getBatchStart;\r\n/* 060 */   }\r\n/* 061 */\r\n/* 062 */   protected void processNext() throws java.io.IOException {\r\n/* 063 */     if (inmemorytablescan_batch == null) {\r\n/* 064 */       inmemorytablescan_nextBatch();\r\n/* 065 */     }\r\n/* 066 */     while (inmemorytablescan_batch != null) {\r\n/* 067 */       int numRows = inmemorytablescan_batch.numRows();\r\n/* 068 */       while (inmemorytablescan_batchIdx < numRows) {\r\n/* 069 */         int inmemorytablescan_rowIdx = inmemorytablescan_batchIdx++;\r\n/* 070 */         boolean inmemorytablescan_isNull = inmemorytablescan_colInstance0.isNullAt(inmemorytablescan_rowIdx);\r\n/* 071 */         int inmemorytablescan_value = inmemorytablescan_isNull ? -1 : (inmemorytablescan_colInstance0.getInt(inmemorytablescan_rowIdx));\r\n/* 072 */\r\n/* 073 */         if (!(!(inmemorytablescan_isNull))) continue;\r\n/* 074 */\r\n/* 075 */         boolean filter_isNull2 = false;\r\n/* 076 */\r\n/* 077 */         boolean filter_value2 = false;\r\n/* 078 */         filter_value2 = inmemorytablescan_value < 8;\r\n/* 079 */         if (!filter_value2) continue;\r\n/* 080 */         boolean inmemorytablescan_isNull1 = inmemorytablescan_colInstance1.isNullAt(inmemorytablescan_rowIdx);\r\n/* 081 */         double inmemorytablescan_value1 = inmemorytablescan_isNull1 ? -1.0 : (inmemorytablescan_colInstance1.getDouble(inmemorytablescan_rowIdx));\r\n/* 082 */\r\n/* 083 */         if (!(!(inmemorytablescan_isNull1))) continue;\r\n/* 084 */\r\n/* 085 */         boolean filter_isNull7 = false;\r\n/* 086 */\r\n/* 087 */         boolean filter_value7 = false;\r\n/* 088 */         filter_value7 = org.apache.spark.util.Utils.nanSafeCompareDoubles(4.0D, inmemorytablescan_value1) < 0;\r\n/* 089 */         if (!filter_value7) continue;\r\n/* 090 */\r\n/* 091 */         filter_numOutputRows.add(1);\r\n/* 092 */\r\n/* 093 */         filter_rowWriter.write(0, inmemorytablescan_value);\r\n/* 094 */\r\n/* 095 */         filter_rowWriter.write(1, inmemorytablescan_value1);\r\n/* 096 */         append(filter_result);\r\n/* 097 */         if (shouldStop()) return;\r\n/* 098 */       }\r\n/* 099 */       inmemorytablescan_batch = null;\r\n/* 100 */       inmemorytablescan_nextBatch();\r\n/* 101 */     }\r\n/* 102 */     inmemorytablescan_scanTime.add(inmemorytablescan_scanTime1 / (1000 * 1000));\r\n/* 103 */     inmemorytablescan_scanTime1 = 0;\r\n/* 104 */   }\r\n/* 105 */ }\r\n```\r\n\r\n2.b Generated code for copying values from `CachedColumnarBatch` in a generated iterator\r\n\r\n``` java\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 */ import org.apache.spark.sql.execution.vectorized.OnHeapUnsafeColumnVector;\r\n/* 008 */\r\n/* 009 */ public SpecificColumnarIterator generate(Object[] references) {\r\n/* 010 */   return new SpecificColumnarIterator(references);\r\n/* 011 */ }\r\n/* 012 */\r\n/* 013 */ class SpecificColumnarIterator extends org.apache.spark.sql.execution.columnar.ColumnarIterator {\r\n/* 014 */   private ColumnVector[] colInstances;\r\n/* 015 */   private UnsafeRow unsafeRow = new UnsafeRow(2);\r\n/* 016 */   private BufferHolder bufferHolder = new BufferHolder(unsafeRow);\r\n/* 017 */   private UnsafeRowWriter rowWriter = new UnsafeRowWriter(bufferHolder, 2);\r\n/* 018 */   private MutableUnsafeRow mutableRow = null;\r\n/* 019 */\r\n/* 020 */   private int rowIdx = 0;\r\n/* 021 */   private int numRowsInBatch = 0;\r\n/* 022 */\r\n/* 023 */   private scala.collection.Iterator input = null;\r\n/* 024 */   private DataType[] columnTypes = null;\r\n/* 025 */   private int[] columnIndexes = null;\r\n/* 026 */\r\n/* 027 */   private org.apache.spark.SparkConf conf;\r\n/* 028 */\r\n/* 029 */   public SpecificColumnarIterator(Object[] references) {\r\n/* 030 */     this.conf = (org.apache.spark.SparkConf) references[0];\r\n/* 031 */     this.mutableRow = new MutableUnsafeRow(rowWriter);\r\n/* 032 */   }\r\n/* 033 */\r\n/* 034 */   public void initialize(Iterator input, DataType[] columnTypes, int[] columnIndexes) {\r\n/* 035 */     this.input = input;\r\n/* 036 */     this.columnTypes = columnTypes;\r\n/* 037 */     this.columnIndexes = columnIndexes;\r\n/* 038 */   }\r\n/* 039 */\r\n/* 040 */\r\n/* 041 */\r\n/* 042 */   public boolean hasNext() {\r\n/* 043 */     if (rowIdx < numRowsInBatch) {\r\n/* 044 */       return true;\r\n/* 045 */     }\r\n/* 046 */     if (!input.hasNext()) {\r\n/* 047 */       return false;\r\n/* 048 */     }\r\n/* 049 */\r\n/* 050 */     org.apache.spark.sql.execution.columnar.CachedColumnarBatch cachedBatch =\r\n/* 051 */     (org.apache.spark.sql.execution.columnar.CachedColumnarBatch) input.next();\r\n/* 052 */     org.apache.spark.sql.execution.vectorized.ColumnarBatch batch = cachedBatch.columnarBatch();\r\n/* 053 */     rowIdx = 0;\r\n/* 054 */     numRowsInBatch = cachedBatch.getNumRows();\r\n/* 055 */     colInstances = new ColumnVector[columnIndexes.length];\r\n/* 056 */     for (int i = 0; i < columnIndexes.length; i ++) {\r\n/* 057 */       colInstances[i] = batch.column(columnIndexes[i]);\r\n/* 058 */       ((OnHeapUnsafeColumnVector)colInstances[i]).decompress(conf);\r\n/* 059 */     }\r\n/* 060 */\r\n/* 061 */     return hasNext();\r\n/* 062 */   }\r\n/* 063 */\r\n/* 064 */   public InternalRow next() {\r\n/* 065 */     bufferHolder.reset();\r\n/* 066 */     rowWriter.zeroOutNullBytes();\r\n/* 067 */     InternalRow row = null;\r\n/* 068 */\r\n/* 069 */     if (colInstances[0].isNullAt(rowIdx)) {\r\n/* 070 */       mutableRow.setNullAt(0);\r\n/* 071 */     } else {\r\n/* 072 */       mutableRow.setInt(0, colInstances[0].getInt(rowIdx));\r\n/* 073 */     }\r\n/* 074 */\r\n/* 075 */     if (colInstances[1].isNullAt(rowIdx)) {\r\n/* 076 */       mutableRow.setNullAt(1);\r\n/* 077 */     } else {\r\n/* 078 */       mutableRow.setDouble(1, colInstances[1].getDouble(rowIdx));\r\n/* 079 */     }\r\n/* 080 */\r\n/* 081 */     unsafeRow.setTotalSize(bufferHolder.totalSize());\r\n/* 082 */     rowIdx += 1;\r\n/* 083 */     return unsafeRow;\r\n/* 084 */   }\r\n/* 085 */ }\r\n\r\n\r\n```\r\n## How was this patch tested?\r\n\r\nAdded new tests\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/264244751","html_url":"https://github.com/apache/spark/pull/15219#issuecomment-264244751","issue_url":"https://api.github.com/repos/apache/spark/issues/15219","id":264244751,"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-12-01T17:56:00Z","updated_at":"2016-12-01T17:56:00Z","body":"Merged build finished. Test FAILed."}}
{"actor":{"display_login":"AmplabJenkins"}}