You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ku...@apache.org on 2021/06/02 06:18:12 UTC
[carbondata] branch master updated: [CARBONDATA-4191] update table
for primitive column not working when complex child column name and
primitive column name match
This is an automated email from the ASF dual-hosted git repository.
kunalkapoor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new 4c04f7c [CARBONDATA-4191] update table for primitive column not working when complex child column name and primitive column name match
4c04f7c is described below
commit 4c04f7c0e49efe5e3203ff2ce04bd7bf1b7fe6aa
Author: Mahesh Raju Somalaraju <ma...@huawei.com>
AuthorDate: Mon May 24 18:01:34 2021 +0530
[CARBONDATA-4191] update table for primitive column not working when complex child
column name and primitive column name match
Why is this PR needed?
Update primitive column not working when complex column child name and primitive
data type name same.
When an update for primitive is received, we are checking in complex child columns
if column name matches then returning UnsupportedOperationbException.
What changes were proposed in this PR?
Currently, we are ignoring the prefix of all columns and passing only columns/child
column info to the update command.
New Changes: Passing full column(alias name/table name.columnName) name which is given
by the user and added checks for handling the unsupported update operation of complex columns.
This closes #4139
---
.../apache/spark/sql/optimizer/CarbonIUDRule.scala | 8 ---
.../spark/sql/parser/CarbonSpark2SqlParser.scala | 34 ++++++++--
.../complexType/TestComplexDataType.scala | 72 +++++++++++++++++++++-
3 files changed, 100 insertions(+), 14 deletions(-)
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonIUDRule.scala b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonIUDRule.scala
index da1ca55..77d85a8 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonIUDRule.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonIUDRule.scala
@@ -41,14 +41,6 @@ class CarbonIUDRule extends Rule[LogicalPlan] with PredicateHelper {
case Project(pList, child) if !isTransformed =>
var (dest: Seq[NamedExpression], source: Seq[NamedExpression]) = pList
.splitAt(pList.size - cols.size)
- // check complex column
- cols.foreach { col =>
- val complexExists = "\"name\":\"" + col + "\""
- if (dest.exists(m => m.dataType.json.contains(complexExists))) {
- throw new UnsupportedOperationException(
- "Unsupported operation on Complex data type")
- }
- }
// check updated columns exists in table
val diff = cols.diff(dest.map(_.name.toLowerCase))
if (diff.nonEmpty) {
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index b77f16f..945a420 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -17,8 +17,9 @@
package org.apache.spark.sql.parser
-import scala.collection.JavaConverters.seqAsJavaListConverter
+import scala.collection.JavaConverters._
import scala.collection.mutable
+import scala.collection.mutable.ListBuffer
import scala.language.implicitConversions
import org.apache.commons.lang3.StringUtils
@@ -257,6 +258,25 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
(SET ~> "(" ~> repsep(element, ",") <~ ")") ~
("=" ~> restInput) <~ opt(";") ^^ {
case tab ~ columns ~ rest =>
+ // If update is received for complex data types then throw exception
+ var finalColumns = List.empty[String]
+ var updateColumns = new ListBuffer[String]()
+ columns.foreach { column =>
+ if (column.contains('.')) {
+ val columnFullName = column.split('.')
+ if (columnFullName.size >= 3) {
+ throw new UnsupportedOperationException("Unsupported operation on Complex data types")
+ } else if ((tab._3.isDefined && tab._3.get.equals(columnFullName(0)))
+ || tab._4.table.equals(columnFullName(0))) {
+ updateColumns += columnFullName(1)
+ } else {
+ throw new UnsupportedOperationException("Unsupported operation on Complex data types")
+ }
+ } else {
+ updateColumns += column
+ }
+ }
+ finalColumns = updateColumns.toList
val (sel, where) = splitQuery(rest)
val selectPattern = """^\s*select\s+""".r
// In case of "update = (subquery) where something"
@@ -349,9 +369,10 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
(sel, updateRelation(tab._1, tab._2, tab._4, tab._3))
}
val rel = tab._3 match {
- case Some(a) => UpdateTable(relation, columns, selectStmt, Some(tab._3.get), where)
+ case Some(a) => UpdateTable(relation, finalColumns, selectStmt, Some(tab._3.get),
+ where)
case None => UpdateTable(relation,
- columns,
+ finalColumns,
selectStmt,
Some(tab._1.tableIdentifier.table),
where)
@@ -377,7 +398,12 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val element: Parser[String] =
(ident <~ ".").? ~ ident ^^ {
- case table ~ column => column.toLowerCase
+ case table ~ column =>
+ if (table.isDefined) {
+ table.get.toLowerCase + "." + column.toLowerCase
+ } else {
+ column.toLowerCase
+ }
}
protected lazy val table: Parser[UnresolvedRelation] = {
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala
index 6c457dd..0610b3e 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala
@@ -757,14 +757,82 @@ class TestComplexDataType extends QueryTest with BeforeAndAfterAll {
sql("DROP TABLE IF EXISTS test")
sql("create table test(id int,a struct<b:int,c:int>,d array<int>) STORED AS carbondata")
sql("insert into test values(1, named_struct('b', 2, 'c', 3), array(4))")
- val structException = intercept[UnsupportedOperationException](
+ val structException = intercept[AnalysisException](
sql("update test set(a.b)=(4) where id=1").collect())
- assertResult("Unsupported operation on Complex data type")(structException.getMessage)
+ assert(structException.getMessage().contains("Unsupported operation on Complex data type"))
val arrayException = intercept[UnsupportedOperationException](
sql("update test set(a)=(4) where id=1").collect())
assertResult("Unsupported operation on Complex data type")(arrayException.getMessage)
}
+ test("check update operation on primitive data types when complex type present in table which " +
+ "has child name equal to primitive data types") {
+ sql("drop table if exists update_complex")
+ sql("create table update_complex (a int, b string, struct1 STRUCT<a:int, c:string>) " +
+ "stored as carbondata")
+ sql("insert into update_complex select 1,'c', named_struct('a',4,'b','d')")
+ sql("update update_complex set (a, b)=(4, 'y')")
+ checkAnswer(sql("select a,b from update_complex"),
+ Seq(Row(4, "y")))
+ sql("update update_complex m set (m.a, m.b)=(5, 'z')")
+ checkAnswer(sql("select a,b from update_complex"),
+ Seq(Row(5, "z")))
+ sql("update update_complex set (update_complex.a, update_complex.b)=(6, 'x')")
+ checkAnswer(sql("select a,b from update_complex"),
+ Seq(Row(6, "x")))
+ sql("update update_complex set (Update_Complex.B)=('g')")
+ checkAnswer(sql("select B from Update_Complex"),
+ Seq(Row("g")))
+ sql("drop table if exists update_complex")
+ }
+
+ test("check update operation on primitive data types when complex type present") {
+ sql("drop table if exists update_complex")
+ sql("create table update_complex (d int, b string, struct1 STRUCT<a:int, b:string>) " +
+ "stored as carbondata")
+ sql("insert into update_complex select 1,'c', named_struct('a',4,'b','d')")
+ sql("update update_complex set (d)=(4)")
+ checkAnswer(sql("select d from update_complex"),
+ Seq(Row(4)))
+ sql("drop table if exists update_complex")
+ }
+
+ test("check update operation on primitive data types when complex type present with alias") {
+ sql("drop table if exists update_complex")
+ sql("create table update_complex (d int, e int, b string, struct1 STRUCT<a:int, b:string>) " +
+ "stored as carbondata")
+ sql("insert into update_complex select 1, 2, 'c', named_struct('a',4,'b','d')")
+ sql("update update_complex m set (d, e)=(4, 4)")
+ checkAnswer(sql("select d,e from update_complex"),
+ Seq(Row(4, 4)))
+ sql("update update_complex m set (m.d, e)=(5, 5)")
+ checkAnswer(sql("select d,e from update_complex"),
+ Seq(Row(5, 5)))
+ sql("update update_complex m set (d, m.e)=(6, 6)")
+ checkAnswer(sql("select d,e from update_complex"),
+ Seq(Row(6, 6)))
+ sql("update update_complex m set (m.d, m.e)=(7, 7)")
+ checkAnswer(sql("select d,e from update_complex"),
+ Seq(Row(7, 7)))
+ sql("drop table if exists update_complex")
+ }
+
+ test("check update operation on complex columns with alias") {
+ sql("DROP TABLE IF EXISTS update_complex")
+ sql("create table update_complex(id int,a struct<b:int,c:int>,d array<int>) STORED AS carbondata")
+ sql("insert into update_complex values(1, named_struct('b', 2, 'c', 3), array(4))")
+ var exception1 = intercept[UnsupportedOperationException](
+ sql("update update_complex m set(a)=(4) where id=1").collect())
+ assert(exception1.getMessage().contains("Unsupported operation on Complex data type"))
+ val exception2 = intercept[AnalysisException](
+ sql("update update_complex m set(a.b)=(4) where id=1").collect())
+ assert(exception2.getMessage().contains("Unsupported operation on Complex data type"))
+ exception1 = intercept[UnsupportedOperationException](
+ sql("update update_complex set(update_complex.a)=(5) where id=1").collect())
+ assert(exception1.getMessage().contains("Unsupported operation on Complex data type"))
+ sql("DROP TABLE IF EXISTS update_complex")
+ }
+
test("test block partition column") {
sql("DROP TABLE IF EXISTS test")
val arrayException = intercept[AnalysisException](