{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/20858","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/20858/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/20858/comments","events_url":"https://api.github.com/repos/apache/spark/issues/20858/events","html_url":"https://github.com/apache/spark/pull/20858","id":306497914,"number":20858,"title":"[SPARK-23736][SQL] Extending the concat function to support array columns","user":{"login":"mn-mikke","id":3674621,"avatar_url":"https://avatars3.githubusercontent.com/u/3674621?v=4","gravatar_id":"","url":"https://api.github.com/users/mn-mikke","html_url":"https://github.com/mn-mikke","followers_url":"https://api.github.com/users/mn-mikke/followers","following_url":"https://api.github.com/users/mn-mikke/following{/other_user}","gists_url":"https://api.github.com/users/mn-mikke/gists{/gist_id}","starred_url":"https://api.github.com/users/mn-mikke/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mn-mikke/subscriptions","organizations_url":"https://api.github.com/users/mn-mikke/orgs","repos_url":"https://api.github.com/users/mn-mikke/repos","events_url":"https://api.github.com/users/mn-mikke/events{/privacy}","received_events_url":"https://api.github.com/users/mn-mikke/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":41,"created_at":"2018-03-19T14:56:27Z","updated_at":"2018-04-16T20:18:41Z","closed_at":null,"author_association":"NONE","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/20858","html_url":"https://github.com/apache/spark/pull/20858","diff_url":"https://github.com/apache/spark/pull/20858.diff","patch_url":"https://github.com/apache/spark/pull/20858.patch"},"body":"## What changes were proposed in this pull request?\r\nThe PR adds a logic for easy concatenation of multiple array columns and covers:\r\n- Concat expression has been extended to support array columns\r\n- A Python wrapper\r\n\r\n## How was this patch tested?\r\nNew tests added into:\r\n- CollectionExpressionsSuite\r\n- DataFrameFunctionsSuite\r\n- typeCoercion/native/concat.sql\r\n\r\n## Codegen examples\r\n### Primitive-type elements\r\n```\r\nval df = Seq(\r\n  (Seq(1 ,2), Seq(3, 4)),\r\n  (Seq(1, 2, 3), null)\r\n).toDF(\"a\", \"b\")\r\ndf.filter('a.isNotNull).select(concat('a, 'b)).debugCodegen()\r\n```\r\nResult:\r\n```\r\n/* 033 */         boolean inputadapter_isNull = inputadapter_row.isNullAt(0);\r\n/* 034 */         ArrayData inputadapter_value = inputadapter_isNull ?\r\n/* 035 */         null : (inputadapter_row.getArray(0));\r\n/* 036 */\r\n/* 037 */         if (!(!inputadapter_isNull)) continue;\r\n/* 038 */\r\n/* 039 */         ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(1);\r\n/* 040 */\r\n/* 041 */         ArrayData[] project_args = new ArrayData[2];\r\n/* 042 */\r\n/* 043 */         if (!false) {\r\n/* 044 */           project_args[0] = inputadapter_value;\r\n/* 045 */         }\r\n/* 046 */\r\n/* 047 */         boolean inputadapter_isNull1 = inputadapter_row.isNullAt(1);\r\n/* 048 */         ArrayData inputadapter_value1 = inputadapter_isNull1 ?\r\n/* 049 */         null : (inputadapter_row.getArray(1));\r\n/* 050 */         if (!inputadapter_isNull1) {\r\n/* 051 */           project_args[1] = inputadapter_value1;\r\n/* 052 */         }\r\n/* 053 */\r\n/* 054 */         ArrayData project_value = new Object() {\r\n/* 055 */           public ArrayData concat(ArrayData[] args) {\r\n/* 056 */             for (int z = 0; z < 2; z++) {\r\n/* 057 */               if (args[z] == null) return null;\r\n/* 058 */             }\r\n/* 059 */\r\n/* 060 */             long project_numElements = 0L;\r\n/* 061 */             for (int z = 0; z < 2; z++) {\r\n/* 062 */               project_numElements += args[z].numElements();\r\n/* 063 */             }\r\n/* 064 */             if (project_numElements > 2147483632) {\r\n/* 065 */               throw new RuntimeException(\"Unsuccessful try to concat arrays with project_numElements\" +\r\n/* 066 */                 \" elements due to exceeding the array size limit 2147483632.\");\r\n/* 067 */             }\r\n/* 068 */\r\n/* 069 */             long project_size = UnsafeArrayData.calculateSizeOfUnderlyingByteArray(\r\n/* 070 */               project_numElements,\r\n/* 071 */               4);\r\n/* 072 */             if (project_size > 2147483632) {\r\n/* 073 */               throw new RuntimeException(\"Unsuccessful try to concat arrays with project_size bytes\" +\r\n/* 074 */                 \" of data due to exceeding the limit 2147483632 bytes for UnsafeArrayData.\");\r\n/* 075 */             }\r\n/* 076 */\r\n/* 077 */             byte[] project_array = new byte[(int)project_size];\r\n/* 078 */             UnsafeArrayData project_arrayData = new UnsafeArrayData();\r\n/* 079 */             Platform.putLong(project_array, 16, project_numElements);\r\n/* 080 */             project_arrayData.pointTo(project_array, 16, (int)project_size);\r\n/* 081 */             int project_counter = 0;\r\n/* 082 */             for (int y = 0; y < 2; y++) {\r\n/* 083 */               for (int z = 0; z < args[y].numElements(); z++) {\r\n/* 084 */                 if (args[y].isNullAt(z)) {\r\n/* 085 */                   project_arrayData.setNullAt(project_counter);\r\n/* 086 */                 } else {\r\n/* 087 */                   project_arrayData.setInt(\r\n/* 088 */                     project_counter,\r\n/* 089 */                     args[y].getInt(z)\r\n/* 090 */                   );\r\n/* 091 */                 }\r\n/* 092 */                 project_counter++;\r\n/* 093 */               }\r\n/* 094 */             }\r\n/* 095 */             return project_arrayData;\r\n/* 096 */           }\r\n/* 097 */         }.concat(project_args);\r\n/* 098 */         boolean project_isNull = project_value == null;\r\n```\r\n\r\n### Non-primitive-type elements\r\n```\r\nval df = Seq(\r\n  (Seq(\"aa\" ,\"bb\"), Seq(\"ccc\", \"ddd\")),\r\n  (Seq(\"x\", \"y\"), null)\r\n).toDF(\"a\", \"b\")\r\ndf.filter('a.isNotNull).select(concat('a, 'b)).debugCodegen()\r\n```\r\nResult:\r\n```\r\n/* 033 */         boolean inputadapter_isNull = inputadapter_row.isNullAt(0);\r\n/* 034 */         ArrayData inputadapter_value = inputadapter_isNull ?\r\n/* 035 */         null : (inputadapter_row.getArray(0));\r\n/* 036 */\r\n/* 037 */         if (!(!inputadapter_isNull)) continue;\r\n/* 038 */\r\n/* 039 */         ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(1);\r\n/* 040 */\r\n/* 041 */         ArrayData[] project_args = new ArrayData[2];\r\n/* 042 */\r\n/* 043 */         if (!false) {\r\n/* 044 */           project_args[0] = inputadapter_value;\r\n/* 045 */         }\r\n/* 046 */\r\n/* 047 */         boolean inputadapter_isNull1 = inputadapter_row.isNullAt(1);\r\n/* 048 */         ArrayData inputadapter_value1 = inputadapter_isNull1 ?\r\n/* 049 */         null : (inputadapter_row.getArray(1));\r\n/* 050 */         if (!inputadapter_isNull1) {\r\n/* 051 */           project_args[1] = inputadapter_value1;\r\n/* 052 */         }\r\n/* 053 */\r\n/* 054 */         ArrayData project_value = new Object() {\r\n/* 055 */           public ArrayData concat(ArrayData[] args) {\r\n/* 056 */             for (int z = 0; z < 2; z++) {\r\n/* 057 */               if (args[z] == null) return null;\r\n/* 058 */             }\r\n/* 059 */\r\n/* 060 */             long project_numElements = 0L;\r\n/* 061 */             for (int z = 0; z < 2; z++) {\r\n/* 062 */               project_numElements += args[z].numElements();\r\n/* 063 */             }\r\n/* 064 */             if (project_numElements > 2147483632) {\r\n/* 065 */               throw new RuntimeException(\"Unsuccessful try to concat arrays with project_numElements\" +\r\n/* 066 */                 \" elements due to exceeding the array size limit 2147483632.\");\r\n/* 067 */             }\r\n/* 068 */\r\n/* 069 */             Object[] project_arrayObjects = new Object[(int)project_numElements];\r\n/* 070 */             int project_counter = 0;\r\n/* 071 */             for (int y = 0; y < 2; y++) {\r\n/* 072 */               for (int z = 0; z < args[y].numElements(); z++) {\r\n/* 073 */                 project_arrayObjects[project_counter] = args[y].getUTF8String(z);\r\n/* 074 */                 project_counter++;\r\n/* 075 */               }\r\n/* 076 */             }\r\n/* 077 */             return new org.apache.spark.sql.catalyst.util.GenericArrayData(project_arrayObjects);\r\n/* 078 */           }\r\n/* 079 */         }.concat(project_args);\r\n/* 080 */         boolean project_isNull = project_value == null;\r\n```"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/381735618","html_url":"https://github.com/apache/spark/pull/20858#issuecomment-381735618","issue_url":"https://api.github.com/repos/apache/spark/issues/20858","id":381735618,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars1.githubusercontent.com/u/8108735?v=4","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2018-04-16T20:18:41Z","updated_at":"2018-04-16T20:18:41Z","author_association":"NONE","body":"**[Test build #89402 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/89402/testReport)** for PR 20858 at commit [`600ae89`](https://github.com/apache/spark/commit/600ae891006b8edda6f817f494e576b65886db47).\n * This patch passes all tests.\n * This patch merges cleanly.\n * This patch adds no public classes."}}
{"actor":{"display_login":"SparkQA"}}