add columnNames function (by aggregation) in MongoDAO

Author: sfxcode

CHANGES.md

@@ -3,6 +3,9 @@
 
 ## Versions
 
+### 2.0.8
+* MongoDAO add method columnNames
+
 ### 2.0.7
 * mongo-scala-driver [4.1.0](https://mongodb.github.io/mongo-java-driver/4.1/whats-new/)
 

src/main/scala/com/sfxcode/nosql/mongo/MongoDAO.scala

@@ -3,14 +3,19 @@ package com.sfxcode.nosql.mongo
 import java.nio.charset.Charset
 
 import better.files.File
-import com.sfxcode.nosql.mongo.bson.DocumentHelper
+import com.sfxcode.nosql.mongo.bson.{BsonConverter, DocumentHelper}
 import com.sfxcode.nosql.mongo.database.{ChangeObserver, CollectionStatus, DatabaseProvider}
 import com.sfxcode.nosql.mongo.operation.Crud
 import org.bson.json.JsonParseException
+import org.mongodb.scala.model.Aggregates.project
+import org.mongodb.scala.model.Projections
 import org.mongodb.scala.{BulkWriteResult, Document, MongoCollection, Observable, SingleObservable}
 
 import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag
+import org.mongodb.scala.model.Filters._
+import org.mongodb.scala.model.Aggregates._
+import org.mongodb.scala.model.Accumulators._
 
 /**
   * Created by tom on 20.01.17.
@@ -32,18 +37,36 @@ abstract class MongoDAO[A](provider: DatabaseProvider, collectionName: String)(i
   def collectionStatus: Observable[CollectionStatus] =
     provider.runCommand(Map("collStats" -> collectionName)).map(document => CollectionStatus(document))
 
+  /**
+    *
+    * @param sampleSize use sample size greate 0 for better performance on big collections
+    * @return List of column names
+    */
+  def columnNames(sampleSize: Int = 0): List[String] = {
+    val projectStage = project(Projections.computed("tempArray", equal("$objectToArray", "$$ROOT")))
+    val unwindStage  = unwind("$tempArray")
+    val groupStage   = group("_id", addToSet("keySet", "$tempArray.k"))
+    val pipeline = {
+      if (sampleSize > 0)
+        List(projectStage, unwindStage, groupStage, sample(sampleSize))
+      else
+        List(projectStage, unwindStage, groupStage)
+    }
+
+    val aggregationResult: Document = Raw.findAggregated(pipeline).result()
+    BsonConverter.fromBson(aggregationResult.get("keySet").head).asInstanceOf[List[String]]
+  }
+
   protected def coll: MongoCollection[A] = collection
 
   // internal object for raw document access
   object Raw extends MongoDAO[Document](provider, collectionName)
 
   def importJsonFile(file: File): SingleObservable[BulkWriteResult] = {
     val docs = new ArrayBuffer[Document]()
-    try {
-      if (file.exists) {
-        val iterator = file.lineIterator(Charset.forName("UTF-8"))
-        iterator.foreach(line => docs.+=(DocumentHelper.documentFromJsonString(line).get))
-      }
+    try if (file.exists) {
+      val iterator = file.lineIterator(Charset.forName("UTF-8"))
+      iterator.foreach(line => docs.+=(DocumentHelper.documentFromJsonString(line).get))
     }
     catch {
       case e: JsonParseException =>

src/test/scala/com/sfxcode/nosql/mongo/dao/BookDAOSpec.scala

@@ -34,6 +34,11 @@ class BookDAOSpec extends Specification with BeforeAll {
       val count: Long = BookDAO.count().result()
       count mustEqual 431
     }
+
+    "support columnNames" in {
+      val columnNames = BookDAO.columnNames()
+      columnNames.size mustEqual 11
+    }
   }
 
   "BookDAO Aggregation" should {

src/test/scala/com/sfxcode/nosql/mongo/dao/PersonDAOSpec.scala

@@ -4,7 +4,6 @@ import java.util.concurrent.TimeUnit
 
 import com.sfxcode.nosql.MongoImplicits
 import com.sfxcode.nosql.mongo.test.TestDatabase.PersonDAO
-import com.sfxcode.nosql.mongo._
 import com.sfxcode.nosql.mongo.model.Person
 
 import scala.concurrent.ExecutionContext.Implicits.global
@@ -20,6 +19,11 @@ class PersonDAOSpec extends PersonSpecification with MongoImplicits {
       count mustEqual 200
     }
 
+    "support columnNames" in {
+      val columnNames = PersonDAO.columnNames(100)
+      columnNames.size mustEqual 18
+    }
+
     "support results" in {
       val seq: Seq[Person] = PersonDAO.find()
       seq must haveSize(200)