Keep Bind around GroupBy and don’t merge aggregates into GroupBy

Fixes #1355.

slick-testkit/src/main/scala/com/typesafe/slick/testkit/tests/AggregateTest.scala

@@ -47,6 +47,9 @@ class AggregateTest extends AsyncTest[RelationalTestDB] {
       val q0 = ts.groupBy(_.a)
       val q1 = q0.map(_._2.length).sortBy(identity)
       db.run(mark("q1", q1.result)).map { r0t: Seq[Int] => r0t shouldBe Vector(2, 3, 3) }
+    }.flatMap { _ =>
+      val q0 = ts.groupBy(_.a).map(_._1).length
+      db.run(mark("q0", q0.result)).map { r0t: Int => r0t shouldBe 3 }
     }.flatMap { _ =>
       val q = (for {
         (k, v) <- ts.groupBy(t => t.a)

slick/src/main/scala/slick/compiler/CreateAggregates.scala

@@ -62,7 +62,7 @@ class CreateAggregates extends Phase {
 
   /** Recursively inline mapping Bind calls under an Aggregate */
   def inlineMap(a: Aggregate): Aggregate = a.from match {
-    case Bind(s1, f1, Pure(StructNode(defs1), ts1)) =>
+    case Bind(s1, f1, Pure(StructNode(defs1), ts1)) if !f1.isInstanceOf[GroupBy] => // mergeToComprehensions always needs a Bind around a GroupBy
       logger.debug("Inlining mapping Bind under Aggregate", a)
       val defs1M = defs1.iterator.toMap
       val sel = a.select.replace({

slick/src/main/scala/slick/compiler/ForceOuterBinds.scala

@@ -50,7 +50,7 @@ class ForceOuterBinds extends Phase {
   }
 
   def maybewrap(n: Node): Node = n match {
-    case _: Join | _: Pure | _: Union | _: FilteredQuery | _:TableNode => wrap(n)
+    case _: Join | _: Pure | _: Union | _: FilteredQuery | _:TableNode | _: GroupBy => wrap(n)
     case _ => nowrap(n)
   }
 }

slick/src/main/scala/slick/compiler/MergeToComprehensions.scala

@@ -150,8 +150,13 @@ class MergeToComprehensions extends Phase {
         (c2, replacements)
 
       case n @ Pure(Aggregate(s1, f1, str1), ts) =>
-        val (c1, replacements1) = mergeFilterWhere(f1, true)
         logger.debug("Merging Aggregate source into Comprehension:", Ellipsis(n, List(0, 0)))
+        val (c1, replacements1) = {
+          val (c, r) = mergeFilterWhere(f1, true)
+          if(c.groupBy.isDefined || c.distinct.isDefined || c.fetch.isDefined || c.offset.isDefined) toSubquery(c, r)
+          else (c, r)
+        }
+        logger.debug("New source at this point is:", c1)
         val str2 = applyReplacements(str1, replacements1, c1)
         val c2 = c1.copy(select = Pure(str2, ts)).infer()
         logger.debug("Merged Aggregate source into Comprehension:", c2)