Better aggregation concatenation support with proper types

all/src/test/scala/spec/AggregationSpec.scala

@@ -81,20 +81,29 @@ class AggregationSpec extends AsyncWordSpec with AsyncIOSpec with Matchers {
       val names = Person.name.concat()
       val age = Person.age.group()
       val count = Person.age.count()
-      Person.withSearchContext { implicit context =>
-        Person.query
-          .aggregate(ids, names, age, count)
-          .sort(count, SortDirection.Descending)
-          .filter(count > 1)
-          .toList
-          .map { list =>
-          list.map(_(names)).map(_.toSet) should be(List(Set("Oscar", "Adam")))
-          list.map(_(ids)).map(_.toSet) should be(List(Set(oscar._id, adam._id).map(_.value)))
-          list.map(_(age)) should be(List(21))
-          list.map(_(count)) should be(List(2))
-        }
+      Person.query
+        .aggregate(ids, names, age, count)
+        .sort(count, SortDirection.Descending)
+        .filter(count > 1)
+        .toList
+        .map { list =>
+        list.map(_(names)).map(_.toSet) should be(List(Set("Oscar", "Adam")))
+        list.map(_(ids)).map(_.toSet) should be(List(Set(oscar._id, adam._id)))
+        list.map(_(age)) should be(List(21))
+        list.map(_(count)) should be(List(2))
       }
     }
+    "aggregate with age concatenation" in {
+      val ages = Person.age.concat()
+      Person.query
+        .aggregate(ages)
+        .toList
+        .map { list =>
+          list.map(_(ages).toSet) should be(List(
+            Set(2, 4, 11, 12, 15, 21, 21, 22, 23, 30, 33, 35, 42, 53, 62, 72, 81, 89, 99, 102)
+          ))
+        }
+    }
     "dispose" in {
       DB.dispose()
     }

core/src/main/scala/lightdb/aggregate/AggregateType.scala

@@ -8,6 +8,7 @@ object AggregateType {
   case object Avg extends AggregateType
   case object Sum extends AggregateType
   case object Count extends AggregateType
+  case object CountDistinct extends AggregateType
   case object Group extends AggregateType
   case object Concat extends AggregateType
 }

core/src/main/scala/lightdb/index/Index.scala

@@ -2,6 +2,7 @@ package lightdb.index
 
 import fabric.rw._
 import fabric._
+import fabric.define.DefType
 import lightdb.{Document, Unique}
 import lightdb.aggregate.{AggregateFilter, AggregateFunction, AggregateType}
 import lightdb.query.Filter
@@ -21,15 +22,26 @@ trait Index[F, D <: Document[D]] extends FilterSupport[F, D, Filter[D]] {
   def avg(name: String = un): AggregateFunction[Double, F, D] = AggregateFunction(name, this, AggregateType.Avg)
   def sum(name: String = un): AggregateFunction[F, F, D] = AggregateFunction(name, this, AggregateType.Sum)
   def count(name: String = un): AggregateFunction[Int, F, D] = AggregateFunction(name, this, AggregateType.Count)
+  def countDistinct(name: String = un): AggregateFunction[Int, F, D] = AggregateFunction(name, this, AggregateType.CountDistinct)
   def group(name: String = un): AggregateFunction[F, F, D] = AggregateFunction(name, this, AggregateType.Group)
-  def concat(name: String = un): AggregateFunction[List[String], F, D] = AggregateFunction(name, this, AggregateType.Concat)(Index.ConcatRW)
+  def concat(name: String = un): AggregateFunction[List[F], F, D] = AggregateFunction(name, this, AggregateType.Concat)(Index.ConcatRW)
 
   def aggregateFilterSupport(name: String): FilterSupport[F, D, AggregateFilter[D]]
 }
 
 object Index {
-  private val ConcatRW: RW[List[String]] = RW.string[List[String]](
-    asString = _.mkString(","),
-    fromString = _.split(',').toList
+  private def ConcatRW[F](implicit fRW: RW[F]): RW[List[F]] = RW.from[List[F]](
+    r = list => arr(list.map(fRW.read)),
+    w = _.asString.split(',').toList.map { s =>
+      val json = fRW.definition match {
+        case DefType.Str => str(s)
+        case DefType.Int => num(s.toLong)
+        case DefType.Dec => num(BigDecimal(s))
+        case DefType.Bool => bool(s.toBoolean)
+        case d => throw new RuntimeException(s"Unsupported DefType $d ($s)")
+      }
+      json.as[F]
+    },
+    d = DefType.Json
   )
 }

sql/src/main/scala/lightdb/sql/SQLSupport.scala

@@ -243,12 +243,17 @@ trait SQLSupport[D <: Document[D]] extends IndexSupport[D] {
           case AggregateType.Min => Some("MIN")
           case AggregateType.Avg => Some("AVG")
           case AggregateType.Sum => Some("SUM")
-          case AggregateType.Count => Some("COUNT")
+          case AggregateType.Count | AggregateType.CountDistinct => Some("COUNT")
           case AggregateType.Concat => Some("GROUP_CONCAT")
           case AggregateType.Group => None
         }
         val fieldName = af match {
-          case Some(s) => s"$s(${f.index.fieldName})"
+          case Some(s) =>
+            val extra = f.`type` match {
+              case AggregateType.CountDistinct => "DISTINCT "
+              case _ => ""
+            }
+            s"$s($extra${f.index.fieldName})"
           case None => f.index.fieldName
         }
         s"$fieldName AS ${f.name}"