Merge 9ae5324 into 14f170c

datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala

@@ -42,7 +42,7 @@ import org.apache.carbondata.core.datastore.impl.FileFactory
 import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
 import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, RelationIdentifier}
 import org.apache.carbondata.datamap.DataMapManager
-import org.apache.carbondata.mv.plans.modular.{GroupBy, Matchable, ModularPlan, Select}
+import org.apache.carbondata.mv.plans.modular._
 import org.apache.carbondata.mv.rewrite.{MVPlanWrapper, QueryRewrite, SummaryDatasetCatalog}
 import org.apache.carbondata.spark.util.CommonUtil
 
@@ -292,6 +292,7 @@ object MVHelper {
         g.child match {
           case s: Select =>
             isValidSelect(isValidExp, s)
+          case m: ModularRelation => isValidExp
         }
       case s: Select =>
         isValidSelect(true, s)

datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/Navigator.scala

@@ -194,6 +194,7 @@ private[mv] class Navigator(catalog: SummaryDatasetCatalog, session: MVSession)
               return intersetJoinEdges.exists(j => j.left == rIndex && j.left == eIndex ||
                 j.right == rIndex && j.right == eIndex)
             }
+          case _ => false
         }
     }
     true

datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/SelectAllColumnsSuite.scala

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.carbondata.mv.rewrite
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.test.util.QueryTest
+
+class SelectAllColumnsSuite extends QueryTest {
+
+  test ("table select all columns mv") {
+    sql("drop datamap if exists all_table_mv")
+    sql("drop table if exists all_table")
+    sql("create table all_table(name string, age int, height int)  stored by 'carbondata'")
+    sql("insert into all_table select 'tom',20,175")
+    sql("insert into all_table select 'tom',32,180")
+    sql("create datamap all_table_mv on table all_table using 'mv' as select avg(age),avg(height),name from all_table group by name")
+    sql("rebuild datamap all_table_mv")
+    checkAnswer(
+      sql("select avg(age),avg(height),name from all_table group by name"),
+      Seq(Row(26.0, 177.5, "tom")))
+    val frame = sql("select avg(age),avg(height),name from all_table group by name")
+    val analyzed = frame.queryExecution.analyzed
+    assert(verifyMVDataMap(analyzed, "all_table_mv"))
+    sql("drop table if exists all_table")
+  }
+
+  def verifyMVDataMap(logicalPlan: LogicalPlan, dataMapName: String): Boolean = {
+    val tables = logicalPlan collect {
+      case l: LogicalRelation => l.catalogTable.get
+    }
+    tables.exists(_.identifier.table.equalsIgnoreCase(dataMapName+"_table"))
+  }
+
+}

datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala

@@ -125,6 +125,8 @@ abstract class ModularPlan
       case modular.Select(_, _, _, _, _, children, _, _, _, _)
         if children.forall(_.isInstanceOf[modular.LeafNode]) => true
 
+      case modular.GroupBy(_, _, _, _, modular.ModularRelation(_, _, _, _, _), _, _, _) => true
+
       case _ => false
     }
   }

datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/SQLBuildDSL.scala

@@ -308,6 +308,32 @@ trait SQLBuildDSL {
       operator: ModularPlan,
       alias: Option[String]): Fragment = {
     operator match {
+      case g@modular.GroupBy(_, _, _, _, s@modular.Select(_, _, _, _, _, _, _, _, _, _), _, _, _) =>
+        val fragmentList = s.children.zipWithIndex
+          .map { case (child, index) => fragmentExtract(child, s.aliasMap.get(index)) }
+        val fList = s.joinEdges.map {
+          e => {
+            (e.right, (fragmentList(e.right), Some(e.joinType), s
+              .extractRightEvaluableConditions(s.children(e.left), s.children(e.right))))
+          }
+        }.toMap
+        val from = (0 to fragmentList.length - 1)
+          .map(index => fList.get(index).getOrElse((fragmentList(index), None, Nil)))
+        val excludesPredicate = from.flatMap(_._3).toSet
+        val windowExprs = s.windowSpec
+          .map { case Seq(expr) => expr.asInstanceOf[Seq[NamedExpression]] }
+          .foldLeft(Seq.empty.asInstanceOf[Seq[NamedExpression]])(_ ++ _)
+        val select = s.outputList ++ windowExprs
+
+        SPJGFragment(
+          select, // select
+          from, // from
+          s.predicateList.filter { p => !excludesPredicate(p) }, // where
+          (Nil, Nil), // group by
+          Nil, // having
+          alias,
+          (s.flags, s.flagSpec))
+
       case s@modular.Select(_, _, _, _, _, _, _, _, _, _) =>
         val fragmentList = s.children.zipWithIndex
           .map { case (child, index) => fragmentExtract(child, s.aliasMap.get(index)) }