Repository: slick/slick (scala)

Issue:
group by with map to use filter on group and calculate length fails
**DESCRIPTION:**
```scala
 val subQuery = some_great_table
      .groupBy(_.queryId)
      .map {
        case (queryId, group) =>
          (queryId, group.filter(_.isEnabled).length, group.filter(!_.isEnabled).length)
      }
```
Following query should work given table : 
```scala
 class SavedQueryCronDefinitionTable(tag: Tag) extends Table[SavedQueryCronDefinition](tag, "saved_query_cron_tab") {
      def * =
        (
          queryId,
          cronExpression,
          tabIdentifier,
          creation,
          startDate,
          endDate,
          niceLevel,
          executionJSON,
          isEnabled
        ) <> (SavedQueryCronDefinition.tupled, SavedQueryCronDefinition.unapply)
      def tabIdentifier: Rep[UUID]                     = column[UUID]("tab_identifier")
      def creation: Rep[LocalDateTime]                 = column[LocalDateTime]("creation")
      def startDate: Rep[Option[LocalDateTime]]        = column[Option[LocalDateTime]]("start_date")
      def endDate: Rep[Option[LocalDateTime]]          = column[Option[LocalDateTime]]("end_date")
      def niceLevel: Rep[Int]                          = column[Int]("niceness")
      def executionJSON: Rep[ExecuteSavedQueryRequest] = column[ExecuteSavedQueryRequest]("execution_json")
      def isEnabled: Rep[Boolean]                      = column[Boolean]("is_enabled")
      def pk: PrimaryKey                               = primaryKey("saved_query_current_schedule_pkey", (queryId, cronExpression))
      def queryId: Rep[String]                         = column[String]("query_id")
      def cronExpression: Rep[String]                  = column[String]("cron_expression")
    }
```


However when `joinLeft` with other tables this gives an error (even without):
```scala
slick.SlickTreeException: Cannot convert node to SQL Comprehension
| Path s5._2 : Vector[t2<{s3: String', s4: Boolean'}>]

	at slick.compiler.MergeToComprehensions.$anonfun$convert$60(MergeToComprehensions.scala:179)
	at scala.Option.getOrElse(Option.scala:189)
	at slick.compiler.MergeToComprehensions.convertBase$1(MergeToComprehensions.scala:179)
	at slick.compiler.MergeToComprehensions.$anonfun$convert$59(MergeToComprehensions.scala:173)
	at slick.compiler.MergeToComprehensions.$anonfun$convert$59$adapted(MergeToComprehensions.scala:173)
	at slick.compiler.MergeToComprehensions.mergeCommon(MergeToComprehensions.scala:356)
	at slick.compiler.MergeToComprehensions.mergeFilterWhere$1(MergeToComprehensions.scala:173)
	at slick.compiler.MergeToComprehensions.mergeGroupBy$1(MergeToComprehensions.scala:168)
	at slick.compiler.MergeToComprehensions.$anonfun$convert$36(MergeToComprehensions.scala:103)
```

**SETUP:**

slick version: `3.3.3` 
db: postgres
sbt: `1.5.0`

Pull request description:
Fix a case when filtered aggregate produces runtime exception by interpreting it with CASE expressions.

Closes #2235

Patch:
diff --git a/slick/src/main/scala/slick/compiler/CreateAggregates.scala b/slick/src/main/scala/slick/compiler/CreateAggregates.scala
index bde1149943..f1e4f80c60 100644
--- a/slick/src/main/scala/slick/compiler/CreateAggregates.scala
+++ b/slick/src/main/scala/slick/compiler/CreateAggregates.scala
@@ -4,7 +4,7 @@ import slick.ast.Library.AggregateFunctionSymbol
 import slick.ast.TypeUtil._
 import slick.ast.Util._
 import slick.ast._
-import slick.util.ConstArray
+import slick.util.{ConstArray, Logging}
 
 /** Rewrite aggregation function calls to Aggregate nodes. */
 class CreateAggregates extends Phase {
@@ -22,10 +22,10 @@ class CreateAggregates extends Phase {
             case _ => Select(Ref(s) :@ elType, els.head._1) :@ els.head._2
           }))(n.nodeType)).infer()
           logger.debug("Converted aggregation function application", a)
-          inlineMap(a)
+          CreateAggregates.inlineMap(a)
 
         case n @ Bind(s1, from1, Pure(sel1, ts1)) if !from1.isInstanceOf[GroupBy] =>
-          val (sel2, temp) = liftAggregates(sel1, s1)
+          val (sel2, temp) = CreateAggregates.liftAggregates(sel1, s1)
           if(temp.isEmpty) n else {
             logger.debug("Lifting aggregates into join in:", n)
             logger.debug("New mapping with temporary refs:", sel2)
@@ -59,7 +59,9 @@ class CreateAggregates extends Phase {
       }, keepType = true, bottomUp = true))
     else state
   }
+}
 
+object CreateAggregates extends Logging {
   /** Recursively inline mapping Bind calls under an Aggregate */
   def inlineMap(a: Aggregate): Aggregate = a.from match {
     case Bind(s1, f1, Pure(StructNode(defs1), ts1)) if !f1.isInstanceOf[GroupBy] => // mergeToComprehensions always needs a Bind around a GroupBy
diff --git a/slick/src/main/scala/slick/compiler/MergeToComprehensions.scala b/slick/src/main/scala/slick/compiler/MergeToComprehensions.scala
index 3a45ea8998..da8683e7a4 100644
--- a/slick/src/main/scala/slick/compiler/MergeToComprehensions.scala
+++ b/slick/src/main/scala/slick/compiler/MergeToComprehensions.scala
@@ -109,10 +109,26 @@ class MergeToComprehensions extends Phase {
 
     /** Merge GroupBy into an existing Comprehension or create new Comprehension from non-grouping Aggregation */
     def mergeGroupBy(n: Node, buildBase: Boolean): (Comprehension.Base, Replacements) = n match {
-      case Bind(s1, GroupBy(_, f1, b1, _), Pure(str1, ts2)) =>
-        val (c1, replacements1) = mergeFilterWhere(f1, buildBase = true)
+      case Bind(s1, GroupBy(_, f1, b1, _), Pure(str0, ts2)) =>
         logger.debug("Merging GroupBy into Comprehension:", Ellipsis(n, List(0, 0)))
+        val str1 = str0.replace {
+          case a@Aggregate(_, fi@Filter(_, from, where), ap: Apply) =>
+            //replace aggregate filters with Case as it is supported by most dbs
+            logger.debug("Replacing Filter on Aggregate with Case:", a)
+            val thenElse = IfThenElse(ConstArray(where) ++ ap.children :+ LiteralNode(null))
+            val select = ap.withChildren(ConstArray(thenElse))
+            val replaced = a.copy(from = from, select = select).replace {
+              case FwdPath(f :: rest) if f == fi.generator =>
+                FwdPath(a.sym +: rest)
+            } match {
+              case a: Aggregate => a
+            }
+            logger.debug("Replaced Filter on Aggregate with Case:", replaced)
+            val inlined = CreateAggregates.inlineMap(replaced)
+            inlined.infer()
+        }
         val (c1a, replacements1a, b2a) = {
+          val (c1, replacements1) = mergeFilterWhere(f1, buildBase = true)
           val b2 = applyReplacements(b1, replacements1, c1)
           // Check whether groupBy keys containing bind variables are returned for further use
           // and push the current Comprehension into a subquery if this is the case.
diff --git a/slick/src/main/scala/slick/memory/QueryInterpreter.scala b/slick/src/main/scala/slick/memory/QueryInterpreter.scala
index 0fe6552c80..47575db1a8 100644
--- a/slick/src/main/scala/slick/memory/QueryInterpreter.scala
+++ b/slick/src/main/scala/slick/memory/QueryInterpreter.scala
@@ -380,6 +380,13 @@ class QueryInterpreter(db: HeapBackend#Database, params: Any) extends Logging {
         case ProductType(_) => coll
         case _ => coll.iterator.filter(v => v != null && v != None)
       }).size
+    case Library.CountDistinct =>
+      val CollectionType(_, elType) = args(0)._1: @unchecked
+      val coll = args(0)._2.asInstanceOf[Coll]
+      (elType match {
+        case ProductType(_) => coll.toSeq.distinct
+        case _ => coll.toSeq.filter(v => v != null && v != None).distinct
+      }).size
     case Library.Database => ""
     case Library.Degrees =>
       val t = args(0)._1.asInstanceOf[ScalaNumericType[Any]]


Test patch:
diff --git a/slick-testkit/src/main/scala/com/typesafe/slick/testkit/tests/AggregateTest.scala b/slick-testkit/src/main/scala/com/typesafe/slick/testkit/tests/AggregateTest.scala
index 5486bac452..3f7a9cf49b 100644
--- a/slick-testkit/src/main/scala/com/typesafe/slick/testkit/tests/AggregateTest.scala
+++ b/slick-testkit/src/main/scala/com/typesafe/slick/testkit/tests/AggregateTest.scala
@@ -313,6 +313,36 @@ class AggregateTest extends AsyncTest[RelationalTestDB] {
     )
   }
 
+  def testFilteredGroupBy = {
+    class A(tag: Tag) extends Table[(Int, Int)](tag, "A_FILTEREDGROUPBY") {
+      def id = column[Int]("id", O.PrimaryKey)
+      def value = column[Int]("value")
+      def * = (id, value)
+    }
+    val as = TableQuery[A]
+    val q0 = as.groupBy(_.value).map(_._2.map(x => Case.If(x.id > 1).Then(x.value)).countDistinct)
+    val q1 = as.groupBy(_.value).map(_._2.filter(_.id > 1).length)
+    val q2 = as.groupBy(_.value).map(_._2.filter(_.id > 1).map(_.value + 1).sum)
+    val q3 = as.groupBy(_.value).map(_._2.filter(_.id > 1).map(_.value).countDistinct)
+    val q4 = as.groupBy(_.value).map(_._2.map(_.id).filter(_ > 1).sum)
+    val q5 = as.groupBy(_.value).map(_._2.map(identity).map(_.id).filter(_ > 1).map(_ + 1).sum)
+    val q6 = as.groupBy(_.value).map(_._2.map(x => (x.id + 1, x.id + x.value)).filter(x => x._1 > 2 && x._2 < 25).map(x => x._1 * x._2).sum)
+    val q7 = as.groupBy(_.value).map { x => (x._1, x._2.filter(_.id > 2).length, x._2.filter(_.id < 2).map(_.value).sum) }
+
+    DBIO.seq(
+      as.schema.create,
+      as ++= Seq((1, 10), (2, 20), (3, 20)),
+      mark("q0", q0.result).map(_.toSet shouldBe Set(0, 1)),
+      mark("q1", q1.result).map(_.toSet shouldBe Set(0, 2)),
+      mark("q2", q2.result).map(_.toSet shouldBe Set(None, Some(42))),
+      mark("q3", q3.result).map(_.toSet shouldBe Set(0, 1)),
+      mark("q4", q4.result).map(_.toSet shouldBe Set(None, Some(5))),
+      mark("q5", q5.result).map(_.toSet shouldBe Set(None, Some(7))),
+      mark("q6", q6.result).map(_.toSet shouldBe Set(None, Some(158))),
+      mark("q7", q7.result).map(_.toSet shouldBe Set((20,1,None), (10,0,Some(10)))),
+    )
+  }
+
   def testDistinct = {
     class A(tag: Tag) extends Table[String](tag, "A_DISTINCT") {
       def id = column[Int]("id", O.PrimaryKey)