You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/03/26 16:03:49 UTC
[1/2] incubator-carbondata git commit: Fixed union with alias error
Repository: incubator-carbondata
Updated Branches:
refs/heads/master 035a33228 -> e0f7aca28
Fixed union with alias error
Modified testcase
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/2390bc2b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/2390bc2b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/2390bc2b
Branch: refs/heads/master
Commit: 2390bc2b82d641554f7aff273ede852a997468d1
Parents: 035a332
Author: ravipesala <ra...@gmail.com>
Authored: Thu Mar 23 11:57:11 2017 +0530
Committer: jackylk <ja...@huawei.com>
Committed: Sun Mar 26 21:33:08 2017 +0530
----------------------------------------------------------------------
.../src/test/resources/join/data1.csv | 2 ++
.../src/test/resources/join/data2.csv | 2 ++
.../joinquery/AllDataTypesTestCaseJoin.scala | 34 ++++++++++++++++++++
.../CarbonDecoderOptimizerHelper.scala | 5 ++-
.../spark/sql/optimizer/CarbonOptimizer.scala | 23 +++++++++----
.../sql/optimizer/CarbonLateDecodeRule.scala | 14 +++++---
6 files changed, 68 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark-common-test/src/test/resources/join/data1.csv
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/resources/join/data1.csv b/integration/spark-common-test/src/test/resources/join/data1.csv
new file mode 100644
index 0000000..e94b3e5
--- /dev/null
+++ b/integration/spark-common-test/src/test/resources/join/data1.csv
@@ -0,0 +1,2 @@
+1,11,100,44.4,flink,2015/5/23 12:01:03,23.23,2015/5/23,ccc,2.5,'foo'#'bar'#'world'
+1,10,150,43.4,spark,2015/7/24 12:01:04,254.12,2015/7/24,ddd,2.5,'foo'#'bar'#'world'
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark-common-test/src/test/resources/join/data2.csv
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/resources/join/data2.csv b/integration/spark-common-test/src/test/resources/join/data2.csv
new file mode 100644
index 0000000..5d23a39
--- /dev/null
+++ b/integration/spark-common-test/src/test/resources/join/data2.csv
@@ -0,0 +1,2 @@
+1,10,100,48.4,spark,2015/4/23 12:01:01,1.23,2015/4/23,aaa,2.5,'foo'#'bar'#'world'
+5,17,140,43.4,spark,2015/7/27 12:01:02,3.45,2015/7/27,bbb,2.5,'foo'#'bar'#'world'
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/joinquery/AllDataTypesTestCaseJoin.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/joinquery/AllDataTypesTestCaseJoin.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/joinquery/AllDataTypesTestCaseJoin.scala
index ecd20e8..be0f8e6 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/joinquery/AllDataTypesTestCaseJoin.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/joinquery/AllDataTypesTestCaseJoin.scala
@@ -56,6 +56,40 @@ class AllDataTypesTestCaseJoin extends QueryTest with BeforeAndAfterAll {
)
}
+
+ test("Union with alias fails") {
+ sql("DROP TABLE IF EXISTS carbon_table1")
+ sql("DROP TABLE IF EXISTS carbon_table2")
+
+ sql("CREATE TABLE carbon_table1(shortField smallint,intField int,bigintField bigint,doubleField double,stringField string,timestampField timestamp,decimalField decimal(18,2),dateField date,charField char(5),floatField float) STORED BY 'carbondata' TBLPROPERTIES('DICTIONARY_INCLUDE'='dateField, charField')")
+
+ sql("CREATE TABLE carbon_table2(shortField smallint,intField int,bigintField bigint,doubleField double,stringField string,timestampField timestamp,decimalField decimal(18,2),dateField date,charField char(5),floatField float) STORED BY 'carbondata' TBLPROPERTIES('DICTIONARY_INCLUDE'='dateField, charField')")
+
+ val path1 = s"$resourcesPath/join/data1.csv"
+ val path2 = s"$resourcesPath/join/data2.csv"
+
+ sql(
+ s"""
+ LOAD DATA LOCAL INPATH '$path1'
+ INTO TABLE carbon_table1
+ options('FILEHEADER'='shortField,intField,bigintField,doubleField,stringField,timestampField,decimalField,dateField,charField,floatField,complexData','COMPLEX_DELIMITER_LEVEL_1'='#')
+ """.stripMargin)
+ sql(
+ s"""
+ LOAD DATA LOCAL INPATH '$path2'
+ INTO TABLE carbon_table2
+ options('FILEHEADER'='shortField,intField,bigintField,doubleField,stringField,timestampField,decimalField,dateField,charField,floatField,complexData','COMPLEX_DELIMITER_LEVEL_1'='#')
+ """.stripMargin)
+
+ checkAnswer(sql("""SELECT t.a a FROM (select charField a from carbon_table1 t1 union all select charField a from carbon_table2 t2) t order by a """),
+ Seq(Row("aaa"),Row("bbb"),Row("ccc"),Row("ddd"))
+ )
+
+ // Drop table
+ sql("DROP TABLE IF EXISTS carbon_table1")
+ sql("DROP TABLE IF EXISTS carbon_table2")
+ }
+
override def afterAll {
sql("drop table alldatatypestableJoin")
sql("drop table alldatatypestableJoin_hive")
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark-common/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
index 021e866..e098f69 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
@@ -25,6 +25,8 @@ import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.carbondata.spark.CarbonAliasDecoderRelation
+
abstract class AbstractNode
case class Node(cd: CarbonDictionaryTempDecoder) extends AbstractNode
@@ -36,7 +38,8 @@ case class CarbonDictionaryTempDecoder(
attrList: util.Set[AttributeReferenceWrapper],
attrsNotDecode: util.Set[AttributeReferenceWrapper],
child: LogicalPlan,
- isOuter: Boolean = false) extends UnaryNode {
+ isOuter: Boolean = false,
+ aliasMap: Option[CarbonAliasDecoderRelation] = None) extends UnaryNode {
var processed = false
def getAttrsNotDecode: util.Set[Attribute] = {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
index 11f5ea2..9aa8158 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
@@ -246,14 +246,21 @@ class ResolveCarbonFunctions(relations: Seq[CarbonDecoderRelation])
union.right.isInstanceOf[CarbonDictionaryTempDecoder]) =>
val leftCondAttrs = new util.HashSet[AttributeReferenceWrapper]
val rightCondAttrs = new util.HashSet[AttributeReferenceWrapper]
+ val leftLocalAliasMap = CarbonAliasDecoderRelation()
+ val rightLocalAliasMap = CarbonAliasDecoderRelation()
+ // collect alias information for the child plan again. It is required as global alias
+ // may have duplicated in case of aliases
+ collectInformationOnAttributes(union.left, leftLocalAliasMap)
+ collectInformationOnAttributes(union.right, rightLocalAliasMap)
union.left.output.foreach { attr =>
- if (isDictionaryEncoded(attr, attrMap, aliasMap)) {
- leftCondAttrs.add(AttributeReferenceWrapper(aliasMap.getOrElse(attr, attr)))
+ if (isDictionaryEncoded(attr, attrMap, leftLocalAliasMap)) {
+ leftCondAttrs.add(AttributeReferenceWrapper(leftLocalAliasMap.getOrElse(attr, attr)))
}
}
union.right.output.foreach { attr =>
- if (isDictionaryEncoded(attr, attrMap, aliasMap)) {
- rightCondAttrs.add(AttributeReferenceWrapper(aliasMap.getOrElse(attr, attr)))
+ if (isDictionaryEncoded(attr, attrMap, rightLocalAliasMap)) {
+ rightCondAttrs.add(
+ AttributeReferenceWrapper(rightLocalAliasMap.getOrElse(attr, attr)))
}
}
var leftPlan = union.left
@@ -262,13 +269,13 @@ class ResolveCarbonFunctions(relations: Seq[CarbonDecoderRelation])
!leftPlan.isInstanceOf[CarbonDictionaryCatalystDecoder]) {
leftPlan = CarbonDictionaryTempDecoder(leftCondAttrs,
new util.HashSet[AttributeReferenceWrapper](),
- union.left)
+ union.left, isOuter = false, Some(leftLocalAliasMap))
}
if (hasCarbonRelation(rightPlan) && rightCondAttrs.size() > 0 &&
!rightPlan.isInstanceOf[CarbonDictionaryCatalystDecoder]) {
rightPlan = CarbonDictionaryTempDecoder(rightCondAttrs,
new util.HashSet[AttributeReferenceWrapper](),
- union.right)
+ union.right, isOuter = false, Some(rightLocalAliasMap))
}
if (!decoder) {
decoder = true
@@ -545,16 +552,18 @@ class ResolveCarbonFunctions(relations: Seq[CarbonDecoderRelation])
}
private def updateTempDecoder(plan: LogicalPlan,
- aliasMap: CarbonAliasDecoderRelation,
+ aliasMapOriginal: CarbonAliasDecoderRelation,
attrMap: java.util.HashMap[AttributeReferenceWrapper, CarbonDecoderRelation]):
LogicalPlan = {
var allAttrsNotDecode: util.Set[AttributeReferenceWrapper] =
new util.HashSet[AttributeReferenceWrapper]()
val marker = new CarbonPlanMarker
+ var aliasMap = aliasMapOriginal
plan transformDown {
case cd: CarbonDictionaryTempDecoder if !cd.processed =>
cd.processed = true
allAttrsNotDecode = cd.attrsNotDecode
+ aliasMap = cd.aliasMap.getOrElse(aliasMap)
marker.pushMarker(allAttrsNotDecode)
if (cd.isOuter) {
CarbonDictionaryCatalystDecoder(relations,
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2390bc2b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
index 6555e9b..36478b4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonLateDecodeRule.scala
@@ -182,9 +182,13 @@ class CarbonLateDecodeRule extends Rule[LogicalPlan] with PredicateHelper {
if !union.children.exists(_.isInstanceOf[CarbonDictionaryTempDecoder]) =>
val children = union.children.map { child =>
val condAttrs = new util.HashSet[AttributeReferenceWrapper]
+ val localAliasMap = CarbonAliasDecoderRelation()
+ // collect alias information for the child plan again. It is required as global alias
+ // may have duplicated in case of aliases
+ collectInformationOnAttributes(child, localAliasMap)
child.output.foreach(attr =>
- if (isDictionaryEncoded(attr, attrMap, aliasMap)) {
- condAttrs.add(AttributeReferenceWrapper(aliasMap.getOrElse(attr, attr)))
+ if (isDictionaryEncoded(attr, attrMap, localAliasMap)) {
+ condAttrs.add(AttributeReferenceWrapper(localAliasMap.getOrElse(attr, attr)))
}
)
@@ -192,7 +196,7 @@ class CarbonLateDecodeRule extends Rule[LogicalPlan] with PredicateHelper {
!child.isInstanceOf[CarbonDictionaryCatalystDecoder]) {
CarbonDictionaryTempDecoder(condAttrs,
new util.HashSet[AttributeReferenceWrapper](),
- child)
+ child, false, Some(localAliasMap))
} else {
child
}
@@ -472,16 +476,18 @@ class CarbonLateDecodeRule extends Rule[LogicalPlan] with PredicateHelper {
}
private def updateTempDecoder(plan: LogicalPlan,
- aliasMap: CarbonAliasDecoderRelation,
+ aliasMapOriginal: CarbonAliasDecoderRelation,
attrMap: java.util.HashMap[AttributeReferenceWrapper, CarbonDecoderRelation]):
LogicalPlan = {
var allAttrsNotDecode: util.Set[AttributeReferenceWrapper] =
new util.HashSet[AttributeReferenceWrapper]()
val marker = new CarbonPlanMarker
+ var aliasMap: CarbonAliasDecoderRelation = aliasMapOriginal
plan transformDown {
case cd: CarbonDictionaryTempDecoder if !cd.processed =>
cd.processed = true
allAttrsNotDecode = cd.attrsNotDecode
+ aliasMap = cd.aliasMap.getOrElse(aliasMapOriginal)
marker.pushMarker(allAttrsNotDecode)
if (cd.isOuter) {
CarbonDictionaryCatalystDecoder(relations,
[2/2] incubator-carbondata git commit: [CARBONDATA-809] Fixed union
with alias error This closes #685
Posted by ja...@apache.org.
[CARBONDATA-809] Fixed union with alias error This closes #685
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/e0f7aca2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/e0f7aca2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/e0f7aca2
Branch: refs/heads/master
Commit: e0f7aca28fa1bbbaf3484eb020df7822fbe37164
Parents: 035a332 2390bc2
Author: jackylk <ja...@huawei.com>
Authored: Sun Mar 26 21:33:38 2017 +0530
Committer: jackylk <ja...@huawei.com>
Committed: Sun Mar 26 21:33:38 2017 +0530
----------------------------------------------------------------------
.../src/test/resources/join/data1.csv | 2 ++
.../src/test/resources/join/data2.csv | 2 ++
.../joinquery/AllDataTypesTestCaseJoin.scala | 34 ++++++++++++++++++++
.../CarbonDecoderOptimizerHelper.scala | 5 ++-
.../spark/sql/optimizer/CarbonOptimizer.scala | 23 +++++++++----
.../sql/optimizer/CarbonLateDecodeRule.scala | 14 +++++---
6 files changed, 68 insertions(+), 12 deletions(-)
----------------------------------------------------------------------