NEW: FetchCountDistinct - query can return distinct values with their counts (#1300)

* NEW: FetchCountDistinct - query can return distinct values with their counts

* Have to disable asserts, because of #1298

src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java

@@ -1,6 +1,7 @@
 package io.ebeaninternal.server.query;
 
 import io.ebean.util.JdbcClose;
+import io.ebean.CountedValue;
 import io.ebeaninternal.api.SpiProfileTransactionEvent;
 import io.ebeaninternal.api.SpiQuery;
 import io.ebeaninternal.api.SpiTransaction;
@@ -60,18 +61,21 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
 
   private final ScalarType<?> scalarType;
 
+  private final boolean containsCounts;
+
   private long profileOffset;
 
   /**
    * Create the Sql select based on the request.
    */
-  CQueryFetchSingleAttribute(OrmQueryRequest<?> request, CQueryPredicates predicates, CQueryPlan queryPlan) {
+  CQueryFetchSingleAttribute(OrmQueryRequest<?> request, CQueryPredicates predicates, CQueryPlan queryPlan, boolean containsCounts) {
     this.request = request;
     this.queryPlan = queryPlan;
     this.query = request.getQuery();
     this.sql = queryPlan.getSql();
     this.desc = request.getBeanDescriptor();
     this.predicates = predicates;
+    this.containsCounts = containsCounts;
     this.scalarType = queryPlan.getSingleAttributeScalarType();
     query.setGeneratedSql(sql);
   }
@@ -101,7 +105,11 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent {
 
       List<Object> result = new ArrayList<>();
       while (dataReader.next()) {
-        result.add(scalarType.read(dataReader));
+        Object value = scalarType.read(dataReader);
+        if (containsCounts) {
+          value = new CountedValue<>(value, dataReader.getLong());
+        }
+        result.add(value);
         dataReader.resetColumnPosition();
         rowCount++;
       }