You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ak...@apache.org on 2020/02/03 05:04:09 UTC
[carbondata] branch master updated: [CARBONDATA-3636]Timeseries
query is not hitting datamap if granularity in query is given case
insensitive
This is an automated email from the ASF dual-hosted git repository.
akashrn5 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 cf47039 [CARBONDATA-3636]Timeseries query is not hitting datamap if granularity in query is given case insensitive
cf47039 is described below
commit cf4703942d95c2c30b6ab370dd5976fc0aaabb86
Author: Indhumathi27 <in...@gmail.com>
AuthorDate: Fri Dec 27 18:10:06 2019 +0530
[CARBONDATA-3636]Timeseries query is not hitting datamap if granularity in query is given case insensitive
Problem:
TimeSeriesUDF function has two parameters:
1.Column -> AttributeReference
2.Granularity -> Literal
For timeseries function having granularity with different case letters in Create datamap and actual query,
select query is not hitting the datamap. This is because, since we store granularity as Literal,
semanticEquals returns false, if granularity is of different case.
Solution:
For TimeSeriesFunction, check Literal value case insensitive
This closes #3541
---
.../carbondata/mv/datamap/MVAnalyzerRule.scala | 2 +-
.../carbondata/mv/rewrite/DefaultMatchMaker.scala | 94 ++++++++++++++++------
.../org/apache/carbondata/mv/rewrite/Utils.scala | 92 ++++++++++++++++++++-
.../TestMVTimeSeriesCreateDataMapCommand.scala | 11 +++
.../timeseries/TestMVTimeSeriesLoadAndQuery.scala | 6 +-
5 files changed, 173 insertions(+), 32 deletions(-)
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVAnalyzerRule.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVAnalyzerRule.scala
index cf76e48..30f795d 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVAnalyzerRule.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVAnalyzerRule.scala
@@ -149,7 +149,7 @@ class MVAnalyzerRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {
if (!outPutUDFColumn.equalsIgnoreCase("") && compactSQL.contains("WHERE")) {
queryArray = compactSQL.split("\n")
queryArray(queryArray.indexOf("WHERE") + 1) = queryArray(
- queryArray.indexOf("WHERE") + 1).replace(outPutUDFColumn,
+ queryArray.indexOf("WHERE") + 1).toLowerCase.replace(outPutUDFColumn.toLowerCase,
s"gen_subsumer_0.`$outPutUDFColumn`")
reWrittenQuery = queryArray.mkString("\n")
}
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
index 6fbc87f..06dd7a1 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter}
+import org.apache.spark.sql.execution.command.timeseries.TimeSeriesFunction
import org.apache.spark.sql.types.{DataType, Metadata}
import org.apache.carbondata.mv.datamap.MVHelper
@@ -59,7 +60,12 @@ abstract class DefaultMatchPattern extends MatchPattern[ModularPlan] {
subsumer.outputList.collect {
case a: Alias if a.child.isInstanceOf[Expression] &&
!a.child.isInstanceOf[AggregateExpression] =>
- a.child -> a.toAttribute
+ a.child match {
+ case s: ScalaUDF if s.function.isInstanceOf[TimeSeriesFunction] =>
+ Utils.getTransformedTimeSeriesUDF(s) -> a.toAttribute
+ case _ =>
+ a.child -> a.toAttribute
+ }
}.toMap
// Check and replace all alias references with subsumer alias map references.
@@ -76,8 +82,18 @@ abstract class DefaultMatchPattern extends MatchPattern[ModularPlan] {
qualifier = a.qualifier)
}.getOrElse(a)
case a: Expression =>
- aliasMapExp
+ var attribute = aliasMapExp
.get(a)
+ // attribute will be empty, if attribute name is of different case. If empty, change
+ // case of scalaUDF present in expression and get updated expression from aliasMap
+ if (attribute.isEmpty) {
+ val newExp = a transform {
+ case s: ScalaUDF if s.function.isInstanceOf[TimeSeriesFunction] =>
+ Utils.getTransformedTimeSeriesUDF(s)
+ }
+ attribute = aliasMapExp.get(newExp)
+ }
+ attribute
.map { ref =>
AttributeReference(
ref.name, ref.dataType)(
@@ -137,17 +153,20 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
compensation: Option[ModularPlan]): Boolean = {
if (subsumee.asInstanceOf[Select].predicateList.contains(exprE)) {
subsumer.asInstanceOf[Select].predicateList.exists(_.semanticEquals(exprE)) ||
- canEvaluate(exprE, subsumer)
+ canEvaluate(exprE, subsumer) ||
+ Utils.isExpressionMatchesUDF(exprE, subsumer.asInstanceOf[Select].predicateList)
} else if (subsumee.asInstanceOf[Select].outputList.contains(exprE)) {
exprE match {
case a@Alias(_, _) =>
exprListR.exists(a1 => a1.isInstanceOf[Alias] &&
a1.asInstanceOf[Alias].child.semanticEquals(a.child)) ||
- exprListR.exists(_.semanticEquals(exprE) || canEvaluate(exprE, subsumer))
+ exprListR.exists(_.semanticEquals(exprE) || canEvaluate(exprE, subsumer)) ||
+ Utils.isExpressionMatchesUDF(exprE, exprListR)
case exp =>
exprListR.exists(a1 => a1.isInstanceOf[Alias] &&
a1.asInstanceOf[Alias].child.semanticEquals(exp)) ||
- exprListR.exists(_.semanticEquals(exprE) || canEvaluate(exprE, subsumer))
+ exprListR.exists(_.semanticEquals(exprE) || canEvaluate(exprE, subsumer)) ||
+ Utils.isExpressionMatchesUDF(exprE, exprListR)
}
} else {
false
@@ -199,7 +218,8 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
val rejoinOutputList = rejoin.flatMap(_.output)
val isPredicateRmE = sel_1a.predicateList.forall(expr =>
- sel_1q.predicateList.exists(_.semanticEquals(expr)))
+ sel_1q.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_1q.predicateList))
val isPredicateEmdR = sel_1q.predicateList.forall(expr =>
isDerivable(expr, sel_1a.outputList ++ rejoinOutputList, sel_1q, sel_1a, None))
// Check if sel_1q.outputList is non empty and then check whether
@@ -244,19 +264,24 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
val sel_1q_join = sel_1q.extractJoinConditions(
sel_1q.children(mappedEdge.left),
sel_1q.children(mappedEdge.right))
- sel_1a_join.forall(e => sel_1q_join.exists(e.semanticEquals(_))) &&
- sel_1q_join.forall(e => sel_1a_join.exists(e.semanticEquals(_)))
+ sel_1a_join.forall(e => sel_1q_join.exists(e.semanticEquals(_)) ||
+ Utils.isExpressionMatchesUDF(e, sel_1q_join)) &&
+ sel_1q_join.forall(e => sel_1a_join.exists(e.semanticEquals(_)) ||
+ Utils.isExpressionMatchesUDF(e, sel_1a_join))
} else false
case _ => false
}
}
val isPredicateEmR = sel_1q.predicateList.forall(expr =>
- sel_1a.predicateList.exists(_.semanticEquals(expr)))
+ sel_1a.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_1a.predicateList))
val isOutputEmR = sel_1q.outputList.forall(expr =>
- sel_1a.outputList.exists(_.semanticEquals(expr)))
+ sel_1a.outputList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_1a.outputList))
val isOutputRmE = sel_1a.outputList.forall(expr =>
- sel_1q.outputList.exists(_.semanticEquals(expr)))
+ sel_1q.outputList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_1q.outputList))
val isLOEmLOR = !(isLeftJoinView(sel_1a) && sel_1q.joinEdges.head.joinType == Inner)
if (r2eJoinsMatch) {
@@ -329,10 +354,13 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
sel_3q.children.forall(_.isInstanceOf[GroupBy]) =>
val isPredicateRmE = sel_3a.predicateList.isEmpty ||
sel_3a.predicateList.forall(expr =>
- sel_3q.predicateList.exists(_.semanticEquals(expr)))
+ sel_3q.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_3q.predicateList))
val isPredicateEmdR = sel_3q.predicateList.isEmpty ||
sel_3q.predicateList.forall(expr =>
- sel_3a.predicateList.exists(_.semanticEquals(expr)) ||
+ sel_3a.predicateList.exists(_.semanticEquals(expr) ||
+ Utils.isExpressionMatchesUDF(expr,
+ sel_3a.predicateList)) ||
isDerivable(expr, sel_3a.outputList, sel_3q, sel_3a, None))
val isOutputEdR = sel_3q.outputList.forall(expr =>
isDerivable(expr, sel_3a.outputList, sel_3q, sel_3a, None))
@@ -341,7 +369,8 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
if (isPredicateRmE && isPredicateEmdR && isOutputEdR && isSingleChild) {
val isPredicateEmR = sel_3q.predicateList.isEmpty ||
sel_3q.predicateList.forall(expr =>
- sel_3a.predicateList.exists(_.semanticEquals(expr)))
+ sel_3a.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_3a.predicateList))
val isOutputRmE = sel_3a.outputList.forall(expr =>
isDerivable(expr, sel_3q.outputList, sel_3a, sel_3q, None))
val isOutputEmR = sel_3q.outputList.forall(expr =>
@@ -355,7 +384,8 @@ object SelectSelectNoChildDelta extends DefaultMatchPattern with PredicateHelper
case a: Alias => sel_3a.outputList
.find { a1 =>
a1.isInstanceOf[Alias] &&
- a1.asInstanceOf[Alias].child.semanticEquals(a.child)
+ (a1.asInstanceOf[Alias].child.semanticEquals(a.child) ||
+ Utils.isExpressionMatchesUDF(a1.asInstanceOf[Alias].child, a.child))
}.map(_.toAttribute).get
})
val wip = sel_3q_exp.copy(
@@ -386,9 +416,11 @@ object GroupbyGroupbyNoChildDelta extends DefaultMatchPattern {
gb_2q @ modular.GroupBy(_, _, _, _, _, _, _, _),
None) =>
val isGroupingEmR = gb_2q.predicateList.forall(expr =>
- gb_2a.predicateList.exists(_.semanticEquals(expr)))
+ gb_2a.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, gb_2a.predicateList))
val isGroupingRmE = gb_2a.predicateList.forall(expr =>
- gb_2q.predicateList.exists(_.semanticEquals(expr)))
+ gb_2q.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, gb_2q.predicateList))
val isOutputEmR = gb_2q.outputList.forall {
case a @ Alias(_, _) =>
gb_2a.outputList.exists{
@@ -403,7 +435,8 @@ object GroupbyGroupbyNoChildDelta extends DefaultMatchPattern {
val mappings = gb_2a.outputList.zipWithIndex.map { case(exp, index) =>
(exp, gb_2q.outputList.find {
case a: Alias if exp.isInstanceOf[Alias] =>
- a.child.semanticEquals(exp.children.head)
+ a.child.semanticEquals(exp.children.head) ||
+ Utils.isExpressionMatchesUDF(a.child, exp.children.head)
case a: Alias => a.child.semanticEquals(exp)
case other => other.semanticEquals(exp)
}.getOrElse(gb_2a.outputList(index)))
@@ -476,7 +509,7 @@ object GroupbyGroupbySelectOnlyChildDelta extends DefaultMatchPattern with Predi
compensation: Option[ModularPlan]) = {
if (subsumee.asInstanceOf[GroupBy].predicateList.contains(exprE)) {
if (exprListR.exists(_.semanticEquals(exprE)) || canEvaluate(exprE, exprListR) ||
- isDerivableForUDF(exprE, exprListR)) {
+ isDerivableForUDF(exprE, exprListR) || Utils.isExpressionMatchesUDF(exprE, exprListR)) {
true
} else {
false
@@ -484,7 +517,7 @@ object GroupbyGroupbySelectOnlyChildDelta extends DefaultMatchPattern with Predi
} else if (compensation.getOrElse(throw new RuntimeException("compensation cannot be None"))
.asInstanceOf[Select].predicateList.contains(exprE)) {
if (canEvaluate(exprE, exprListR) || exprListR.exists(_.semanticEquals(exprE)) ||
- isDerivableForUDF(exprE, exprListR)) {
+ isDerivableForUDF(exprE, exprListR) || Utils.isExpressionMatchesUDF(exprE, exprListR)) {
true
} else {
false
@@ -546,7 +579,9 @@ object GroupbyGroupbySelectOnlyChildDelta extends DefaultMatchPattern with Predi
val rejoinOutputList = sel_1c1.children.tail.flatMap(_.output)
val isGroupingEdR = gb_2q.predicateList.forall(expr =>
isDerivable(expr, gb_2a.predicateList ++ rejoinOutputList, gb_2q, gb_2a, compensation))
- val needRegrouping = !gb_2a.predicateList.forall(gb_2q.predicateList.contains)
+ val needRegrouping = !gb_2a.predicateList
+ .forall(f => gb_2q.predicateList.contains(f) ||
+ Utils.isExpressionMatchesUDF(f, gb_2q.predicateList))
val canPullup = sel_1c1.predicateList.forall(expr =>
isDerivable(expr, gb_2a.predicateList ++ rejoinOutputList, gb_2q, gb_2a, compensation))
val isAggEmR = gb_2q.outputList.collect {
@@ -558,9 +593,12 @@ object GroupbyGroupbySelectOnlyChildDelta extends DefaultMatchPattern with Predi
// pull up
val pullupOutputList = gb_2a.outputList.map(_.toAttribute) ++ rejoinOutputList
val myOutputList = gb_2a.outputList.filter {
- case alias: Alias => gb_2q.outputList.filter(_.isInstanceOf[Alias])
- .exists(_.asInstanceOf[Alias].child.semanticEquals(alias.child))
- case attr: Attribute => gb_2q.outputList.exists(_.semanticEquals(attr))
+ case alias: Alias =>
+ val aliasList = gb_2q.outputList.filter(_.isInstanceOf[Alias])
+ aliasList.exists(_.asInstanceOf[Alias].child.semanticEquals(alias.child)) ||
+ Utils.isExpressionMatchesUDF(alias.child, aliasList)
+ case attr: Attribute =>
+ gb_2q.outputList.exists(_.semanticEquals(attr))
}.map(_.toAttribute) ++ rejoinOutputList
// TODO: find out if we really need to check needRegrouping or just use myOutputList
val sel_2c1 = if (needRegrouping) {
@@ -707,10 +745,13 @@ object SelectSelectGroupbyChildDelta extends DefaultMatchPattern with PredicateH
val isPredicateRmE = sel_3a.predicateList.forall(expr =>
sel_3q.predicateList.exists(_.semanticEquals(expr)) ||
- gb_2c.predicateList.exists(_.semanticEquals(expr)))
+ Utils.isExpressionMatchesUDF(expr, sel_3q.predicateList) ||
+ gb_2c.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, gb_2c.predicateList))
val isPredicateEmdR = sel_3q.predicateList
.forall(expr =>
sel_3a.predicateList.exists(_.semanticEquals(expr)) ||
+ Utils.isExpressionMatchesUDF(expr, sel_3a.predicateList) ||
isDerivable(
expr,
sel_3a.outputList ++ rejoinOutputList,
@@ -769,7 +810,8 @@ object SelectSelectGroupbyChildDelta extends DefaultMatchPattern with PredicateH
val mappings = sel_3q_exp.outputList.zipWithIndex.map { case(exp, index) =>
(exp, gb_2c.outputList.find {
case a: Alias if exp.isInstanceOf[Alias] =>
- a.child.semanticEquals(exp.children.head)
+ a.child.semanticEquals(exp.children.head) ||
+ Utils.isExpressionMatchesUDF(a.child, exp.children.head)
case a: Alias => a.child.semanticEquals(exp)
case other => other.semanticEquals(exp)
}.getOrElse(gb_2c.outputList(index)))
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/Utils.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/Utils.scala
index 802be83..2733a5a 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/Utils.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/Utils.scala
@@ -17,8 +17,9 @@
package org.apache.carbondata.mv.rewrite
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, Cast, Divide, Expression, Multiply, PredicateHelper, ScalaUDF}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, Cast, Divide, Expression, Literal, Multiply, PredicateHelper, ScalaUDF}
import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.execution.command.timeseries.TimeSeriesFunction
import org.apache.carbondata.mv.plans.modular
import org.apache.carbondata.mv.plans.modular.{ModularPlan, Select}
@@ -321,7 +322,8 @@ object Utils extends PredicateHelper {
case expr: Expression if !expr.isInstanceOf[AggregateFunction] =>
operator_a.outputList.find {
case alias: Alias if alias_m.contains(alias.toAttribute) &&
- alias_m(alias.toAttribute).child.semanticEquals(expr) &&
+ (alias_m(alias.toAttribute).child.semanticEquals(expr) ||
+ Utils.isExpressionMatchesUDF(alias_m(alias.toAttribute), expr)) &&
!alias_m(alias.toAttribute).child
.isInstanceOf[AggregateExpression] => true
case attr: Attribute if alias_m.contains(attr) &&
@@ -459,14 +461,14 @@ object Utils extends PredicateHelper {
subsumer.asInstanceOf[Select].outputList.forall {
case Alias(s: ScalaUDF, _) =>
expE.children.foreach { expr =>
- if (s.semanticEquals(expr)) {
+ if (s.semanticEquals(expr)|| isExpressionMatchesUDF(s, expr)) {
canBeDerived = true
}
// It is because when expression is like between filter, the expr will be as Cast
// expression and its child will be ScalaUDF(timeseries), So compare the child also.
if (!canBeDerived && null != expr.children) {
expr.children.foreach { expC =>
- if (s.semanticEquals(expC)) {
+ if (s.semanticEquals(expC) || isExpressionMatchesUDF(s, expC)) {
canBeDerived = true
}
}
@@ -519,4 +521,86 @@ object Utils extends PredicateHelper {
}
}
+ /**
+ * Check's if timeseries udf function exists. If exists, compare literal with case insensitive
+ * value
+ */
+ def isExpressionMatchesUDF(subsumeExp: Expression, subsumerExprList: Seq[Expression]): Boolean = {
+ // Check if expression has a ScalaUDF of timeSeries function and verify it's children
+ // irrespective of case. The structure of scalaUDF function will look like,
+ // ScalaUDF
+ // |
+ // TimeSeriesFunction
+ // / \
+ // / \
+ // AttributeReference Literal
+ subsumeExp match {
+ case Alias(udf: ScalaUDF, _) if udf.function.isInstanceOf[TimeSeriesFunction] =>
+ val children = udf.children
+ val subsumerTimeSeriesExp = subsumerExprList.filter(a1 =>
+ a1.isInstanceOf[Alias] && a1.asInstanceOf[Alias].child.isInstanceOf[ScalaUDF] &&
+ a1.asInstanceOf[Alias].child.asInstanceOf[ScalaUDF].function.
+ isInstanceOf[TimeSeriesFunction])
+ subsumerTimeSeriesExp.exists(f => {
+ val childExprsOfTimeSeriesUDF = f.asInstanceOf[Alias].child
+ .asInstanceOf[ScalaUDF].children
+ childExprsOfTimeSeriesUDF.head.semanticEquals(children.head) &&
+ childExprsOfTimeSeriesUDF.last.asInstanceOf[Literal].toString().equalsIgnoreCase(
+ children.last.asInstanceOf[Literal].toString())
+ })
+ case udf: ScalaUDF if udf.function.isInstanceOf[TimeSeriesFunction] =>
+ val children = udf.children
+ var subsumerTimeSeriesExprList: Seq[Expression] = Seq.empty
+ subsumerExprList foreach {
+ case s: ScalaUDF if s.function.isInstanceOf[TimeSeriesFunction] =>
+ subsumerTimeSeriesExprList = subsumerTimeSeriesExprList.+:(s)
+ case Alias(s: ScalaUDF, _) if s.function.isInstanceOf[TimeSeriesFunction] =>
+ subsumerTimeSeriesExprList = subsumerTimeSeriesExprList.+:(s.asInstanceOf[Expression])
+ case _ =>
+ }
+ subsumerTimeSeriesExprList.exists(f => {
+ val childExprsOfTimeSeriesUDF = f.asInstanceOf[ScalaUDF].children
+ childExprsOfTimeSeriesUDF.head.semanticEquals(children.head) &&
+ childExprsOfTimeSeriesUDF.last.asInstanceOf[Literal].toString().equalsIgnoreCase(
+ children.last.asInstanceOf[Literal].toString())
+ })
+ case exp: Expression =>
+ val transformedExpwithLowerCase = exp.transform {
+ case s: ScalaUDF if s.function.isInstanceOf[TimeSeriesFunction] =>
+ getTransformedTimeSeriesUDF(s)
+ case other => other
+ }
+ val transformedExprListWithLowerCase = subsumerExprList map { expr =>
+ expr.transform {
+ case s: ScalaUDF if s.function.isInstanceOf[TimeSeriesFunction] =>
+ getTransformedTimeSeriesUDF(s)
+ case other => other
+ }
+ }
+ transformedExprListWithLowerCase.exists(_.semanticEquals(transformedExpwithLowerCase))
+ case _ => false
+ }
+ }
+
+ def getTransformedTimeSeriesUDF(s: ScalaUDF): Expression = {
+ s.transform {
+ case l: Literal =>
+ Literal(l.toString().toLowerCase, l.dataType)
+ }
+ }
+
+ /**
+ * Check if expr1 and expr2 matches TimeSeriesUDF function. If both expressions are
+ * timeseries udf functions, then check it's childrens are same irrespective of case.
+ */
+ def isExpressionMatchesUDF(expr1: Expression, expr2: Expression): Boolean = {
+ (expr1, expr2) match {
+ case (s1: ScalaUDF, s2: ScalaUDF) if s1.function.isInstanceOf[TimeSeriesFunction] &&
+ s2.function.isInstanceOf[TimeSeriesFunction] =>
+ s1.children.head.semanticEquals(s2.children.head) &&
+ s1.children.last.asInstanceOf[Literal].toString()
+ .equalsIgnoreCase(s2.children.last.asInstanceOf[Literal].toString())
+ case _ => false
+ }
+ }
}
diff --git a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesCreateDataMapCommand.scala b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesCreateDataMapCommand.scala
index 375eb64..160332f 100644
--- a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesCreateDataMapCommand.scala
+++ b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesCreateDataMapCommand.scala
@@ -210,6 +210,17 @@ class TestMVTimeSeriesCreateDataMapCommand extends QueryTest with BeforeAndAfter
}.getMessage.contains("MV Timeseries is only supported on Timestamp/Date column")
}
+ test("test timeseries with case sensitive granularity") {
+ sql("drop datamap if exists datamap1")
+ sql("create datamap datamap1 on table maintable using 'mv'" +
+ " as select timeseries(projectjoindate,'Second'), sum(projectcode) from maintable group by timeseries(projectjoindate,'Second')")
+ val df1 = sql("select timeseries(projectjoindate,'SECOND'), sum(projectcode) from maintable group by timeseries(projectjoindate,'SECOND')")
+ val df2 = sql("select timeseries(projectjoinDATE,'SECOnd'), sum(projectcode) from maintable where projectcode=8 group by timeseries(projectjoinDATE,'SECOnd')")
+ TestUtil.verifyMVDataMap(df1.queryExecution.optimizedPlan, "datamap1")
+ TestUtil.verifyMVDataMap(df2.queryExecution.optimizedPlan, "datamap1")
+ sql("drop datamap if exists datamap1")
+ }
+
class QueryTask(query: String) extends Callable[String] {
override def call(): String = {
var result = "PASS"
diff --git a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesLoadAndQuery.scala b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesLoadAndQuery.scala
index 3916ea3..fdc3190 100644
--- a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesLoadAndQuery.scala
+++ b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/timeseries/TestMVTimeSeriesLoadAndQuery.scala
@@ -48,6 +48,8 @@ class TestMVTimeSeriesLoadAndQuery extends QueryTest with BeforeAndAfterAll {
val df1 = sql("select timeseries(projectjoindate,'minute'),sum(projectcode) from maintable where timeseries(projectjoindate,'minute') = '2016-02-23 09:17:00'" +
"group by timeseries(projectjoindate,'minute')")
assert(TestUtil.verifyMVDataMap(df1.queryExecution.optimizedPlan, "datamap1"))
+ val df2 = sql("select timeseries(projectjoindate,'MINUTE'), sum(projectcode) from maintable where timeseries(projectjoindate,'MINute') = '2016-02-23 09:17:00' group by timeseries(projectjoindate,'MINUTE')")
+ TestUtil.verifyMVDataMap(df2.queryExecution.optimizedPlan, "datamap1")
dropDataMap("datamap1")
}
@@ -181,8 +183,10 @@ class TestMVTimeSeriesLoadAndQuery extends QueryTest with BeforeAndAfterAll {
"create datamap datamap1 on table maintable using 'mv' as " +
"select timeseries(projectjoindate,'month'), max(salary) from maintable where timeseries(projectjoindate,'month') = '2016-03-01 00:00:00' or timeseries(projectjoindate,'month') = '2016-02-01 00:00:00' group by timeseries(projectjoindate,'month')")
loadData("maintable")
- val df1 = sql("select timeseries(projectjoindate,'month'), max(salary) from maintable where timeseries(projectjoindate,'month') = '2016-03-01 00:00:00' or timeseries(projectjoindate,'month') = '2016-02-01 00:00:00' group by timeseries(projectjoindate,'month')")
+ var df1 = sql("select timeseries(projectjoindate,'month'), max(salary) from maintable where timeseries(projectjoindate,'month') = '2016-03-01 00:00:00' or timeseries(projectjoindate,'month') = '2016-02-01 00:00:00' group by timeseries(projectjoindate,'month')")
checkPlan("datamap1", df1)
+ df1 = sql("select timeseries(projectjoindate,'MONth'), max(salary) from maintable where timeseries(projectjoindate,'MoNtH') = '2016-03-01 00:00:00' or timeseries(projectjoinDATE,'MONth') = '2016-02-01 00:00:00' group by timeseries(projectjoindate,'MONth')")
+ TestUtil.verifyMVDataMap(df1.queryExecution.optimizedPlan, "datamap1")
sql(
"create datamap datamap2 on table maintable using 'mv' as " +
"select timeseries(projectjoindate,'month'), max(salary) from maintable where timeseries(projectjoindate,'month') = '2016-03-01 00:00:00' and timeseries(projectjoindate,'month') = '2016-02-01 00:00:00' group by timeseries(projectjoindate,'month')")