(dsl): Support Weight Avg aggregation (#358)

Author: vanjaftn

docs/overview/aggregations/elastic_aggregation_weighted_avg.md

@@ -0,0 +1,44 @@
+---
+id: elastic_aggregation_weighted_avg
+title: "Weighted Avg Aggregation"
+---
+
+The `Weighted Avg` aggregation is a single-value metrics aggregation that computes the average while taking into account the varying degrees of importance of numeric values. As a formula, a weighted average is the `∑(value * weight) / ∑(weight)`
+
+In order to use the `Weighted Avg` aggregation import the following:
+```scala
+import zio.elasticsearch.aggregation.WeightedAvgAggregation
+import zio.elasticsearch.ElasticAggregation.weightedAvgAggregation
+```
+
+You can create a `Weighted Avg` aggregation using the `weightedAvgAggregation` method this way:
+```scala
+val aggregation: WeightedAvgAggregation = weightedAvgAggregation(name = "weightedAvgAggregation", valueField = "doubleField", weightField = "intField")
+```
+
+You can create a [type-safe](https://lambdaworks.github.io/zio-elasticsearch/overview/overview_zio_prelude_schema) `Weighted Avg` aggregation using the `weightedAvgAggregation` method this way:
+```scala
+val aggregation: WeightedAvgAggregation = weightedAvgAggregation(name = "weightedAvgAggregation", valueField = Document.doubleField, weightField = Document.intField)
+```
+
+If you want to add aggregation (on the same level), you can use `withAgg` method:
+```scala
+val multipleAggregations: MultipleAggregations = weightedAvgAggregation(name = "weightedAvgAggregation1", valueField = Document.intField, weightField = Document.doubleField).withAgg(weightedAvgAggregation(name = "weightedAvgAggregation2", valueField = Document.doubleField, weightField = Document.intField))
+```
+
+If you want to change the `valueMissing`, you can use `valueMissing` method:
+```scala
+val aggregationWithValueMissing: WeightedAvgAggregation = weightedAvgAggregation(name = "weightedAvgAggregation", field = Document.intField).valueMissing(10.0)
+```
+
+If you want to change the `weightMissing`, you can use `weightMissing` method:
+```scala
+val aggregationWithWeightMissing: WeightedAvgAggregation = weightedAvgAggregation(name = "weightedAvgAggregation", field = Document.intField).weightMissing(5.0)
+```
+
+If you want to change the `weightMissing` and `valueMissing`, you can use `weightMissing` and `valueMissing` methods:
+```scala
+val aggregationWithValueAndWeightMissing: WeightedAvgAggregation = weightedAvgAggregation(name = "weightedAvgAggregation", field = Document.intField).valueMissing(5.0).weightMissing(10.0)
+```
+
+You can find more information about `Weighted Avg` aggregation [here](https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-metrics-weight-avg-aggregation.html).

modules/integration/src/test/scala/zio/elasticsearch/HttpExecutorSpec.scala

@@ -496,6 +496,45 @@ object HttpExecutorSpec extends IntegrationSpec {
           } @@ around(
             Executor.execute(ElasticRequest.createIndex(firstSearchIndex)),
             Executor.execute(ElasticRequest.deleteIndex(firstSearchIndex)).orDie
+          ),
+          test("aggregate using weighted avg aggregation") {
+            checkOnce(genDocumentId, genTestDocument, genDocumentId, genTestDocument) {
+              (firstDocumentId, firstDocument, secondDocumentId, secondDocument) =>
+                for {
+                  _ <- Executor.execute(ElasticRequest.deleteByQuery(firstSearchIndex, matchAll))
+                  _ <-
+                    Executor.execute(
+                      ElasticRequest
+                        .upsert[TestDocument](
+                          firstSearchIndex,
+                          firstDocumentId,
+                          firstDocument.copy(doubleField = 5, intField = 2)
+                        )
+                    )
+                  _ <-
+                    Executor.execute(
+                      ElasticRequest
+                        .upsert[TestDocument](
+                          firstSearchIndex,
+                          secondDocumentId,
+                          secondDocument.copy(doubleField = 10, intField = 3)
+                        )
+                        .refreshTrue
+                    )
+                  aggregation = weightedAvgAggregation(
+                                  name = "weightedAggregation",
+                                  valueField = TestDocument.doubleField,
+                                  weightField = TestDocument.intField
+                                )
+                  aggsRes <-
+                    Executor
+                      .execute(ElasticRequest.aggregate(selectors = firstSearchIndex, aggregation = aggregation))
+                      .asWeightedAvgAggregation("weightedAggregation")
+                } yield assert(aggsRes.head.value)(equalTo(8.0))
+            }
+          } @@ around(
+            Executor.execute(ElasticRequest.createIndex(firstSearchIndex)),
+            Executor.execute(ElasticRequest.deleteIndex(firstSearchIndex)).orDie
           )
         ),
         suite("search with aggregation")(

modules/library/src/main/scala/zio/elasticsearch/ElasticAggregation.scala

@@ -353,4 +353,53 @@ object ElasticAggregation {
   final def valueCountAggregation(name: String, field: String): ValueCountAggregation =
     ValueCount(name = name, field = field)
 
+  /**
+   * Constructs a type-safe instance of [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] using the specified
+   * parameters.
+   *
+   * @param name
+   *   aggregation name
+   * @param valueField
+   *   the type-safe field that represents value for which weighted avg aggregation will be executed
+   * @param weightField
+   *   the type-safe field that represents weight for which weighted avg aggregation will be executed
+   * @return
+   *   an instance of [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] that represents weighted avg aggregation
+   *   to be performed.
+   */
+  final def weightedAvgAggregation(
+    name: String,
+    valueField: Field[_, Any],
+    weightField: Field[_, Any]
+  ): WeightedAvgAggregation =
+    WeightedAvg(
+      name = name,
+      valueField = valueField.toString,
+      weightField = weightField.toString,
+      valueMissing = None,
+      weightMissing = None
+    )
+
+  /**
+   * Constructs an instance of [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] using the specified parameters.
+   *
+   * @param name
+   *   aggregation name
+   * @param valueField
+   *   the field that represents value for which weighted avg aggregation will be executed
+   * @param weightField
+   *   the field that represents weight for which weighted avg aggregation will be executed
+   * @return
+   *   an instance of [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] that represents weighted avg aggregation
+   *   to be performed.
+   */
+  final def weightedAvgAggregation(name: String, valueField: String, weightField: String): WeightedAvgAggregation =
+    WeightedAvg(
+      name = name,
+      valueField = valueField,
+      weightField = weightField,
+      valueMissing = None,
+      weightMissing = None
+    )
+
 }

modules/library/src/main/scala/zio/elasticsearch/aggregation/Aggregations.scala

@@ -356,3 +356,63 @@ private[elasticsearch] final case class ValueCount(name: String, field: String)
   private[elasticsearch] def toJson: Json =
     Obj(name -> Obj("value_count" -> Obj("field" -> field.toJson)))
 }
+
+sealed trait WeightedAvgAggregation extends SingleElasticAggregation with WithAgg {
+
+  /**
+   * Sets the `valueMissing` parameter for the [[zio.elasticsearch.aggregation.WeightedAvgAggregation]].
+   * The`valueMissing` parameter provides a value to use when a document is missing the value field that the aggregation
+   * is running on.
+   *
+   * @param value
+   *   the value to use for missing documents
+   * @return
+   *   an instance of the [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] enriched with the `valueMissing`
+   *   parameter.
+   */
+  def valueMissing(value: Double): WeightedAvgAggregation
+
+  /**
+   * Sets the `weightMissing` parameter for the [[zio.elasticsearch.aggregation.WeightedAvgAggregation]].
+   * The`weightMissing` parameter provides a value to use when a document is missing the weight field that the
+   * aggregation is running on.
+   *
+   * @param value
+   *   the value to use for missing documents
+   * @return
+   *   an instance of the [[zio.elasticsearch.aggregation.WeightedAvgAggregation]] enriched with the `weightMissing`
+   *   parameter.
+   */
+  def weightMissing(value: Double): WeightedAvgAggregation
+}
+
+private[elasticsearch] final case class WeightedAvg(
+  name: String,
+  valueField: String,
+  weightField: String,
+  valueMissing: Option[Double],
+  weightMissing: Option[Double]
+) extends WeightedAvgAggregation { self =>
+
+  def valueMissing(value: Double): WeightedAvgAggregation =
+    self.copy(valueMissing = Some(value))
+
+  def weightMissing(value: Double): WeightedAvgAggregation =
+    self.copy(weightMissing = Some(value))
+
+  def withAgg(agg: SingleElasticAggregation): MultipleAggregations =
+    multipleAggregations.aggregations(self, agg)
+
+  private[elasticsearch] def toJson: Json = {
+    val valueMissingJson: Json  = valueMissing.fold(Obj())(m => Obj("missing" -> m.toJson))
+    val weightMissingJson: Json = weightMissing.fold(Obj())(m => Obj("missing" -> m.toJson))
+
+    Obj(
+      name -> Obj(
+        "weighted_avg" -> (Obj("value" -> (Obj("field" -> valueField.toJson) merge valueMissingJson)) merge (Obj(
+          "weight" -> (Obj("field" -> weightField.toJson) merge weightMissingJson)
+        )))
+      )
+    )
+  }
+}

modules/library/src/main/scala/zio/elasticsearch/executor/response/AggregationResponse.scala

@@ -59,6 +59,8 @@ object AggregationResponse {
         )
       case ValueCountAggregationResponse(value) =>
         ValueCountAggregationResult(value)
+      case WeightedAvgAggregationResponse(value) =>
+        WeightedAvgAggregationResult(value)
     }
 }
 
@@ -118,7 +120,6 @@ private[elasticsearch] final case class SumAggregationResponse(value: Double) ex
 
 private[elasticsearch] object SumAggregationResponse {
   implicit val decoder: JsonDecoder[SumAggregationResponse] = DeriveJsonDecoder.gen[SumAggregationResponse]
-
 }
 
 private[elasticsearch] final case class TermsAggregationResponse(
@@ -154,6 +155,8 @@ private[elasticsearch] object TermsAggregationBucket {
           val objFields = data.unsafeAs[Obj].fields.toMap
 
           (field: @unchecked) match {
+            case str if str.contains("weighted_avg#") =>
+              Some(field -> WeightedAvgAggregationResponse(value = objFields("value").unsafeAs[Double]))
             case str if str.contains("avg#") =>
               Some(field -> AvgAggregationResponse(value = objFields("value").unsafeAs[Double]))
             case str if str.contains("cardinality#") =>
@@ -199,6 +202,8 @@ private[elasticsearch] object TermsAggregationBucket {
     val subAggs = allFields.collect {
       case (field, data) if field != "key" && field != "doc_count" =>
         (field: @unchecked) match {
+          case str if str.contains("weighted_avg#") =>
+            (field.split("#")(1), data.asInstanceOf[WeightedAvgAggregationResponse])
           case str if str.contains("avg#") =>
             (field.split("#")(1), data.asInstanceOf[AvgAggregationResponse])
           case str if str.contains("cardinality#") =>
@@ -239,3 +244,10 @@ private[elasticsearch] object ValueCountAggregationResponse {
   implicit val decoder: JsonDecoder[ValueCountAggregationResponse] =
     DeriveJsonDecoder.gen[ValueCountAggregationResponse]
 }
+
+private[elasticsearch] final case class WeightedAvgAggregationResponse(value: Double) extends AggregationResponse
+
+private[elasticsearch] object WeightedAvgAggregationResponse {
+  implicit val decoder: JsonDecoder[WeightedAvgAggregationResponse] =
+    DeriveJsonDecoder.gen[WeightedAvgAggregationResponse]
+}

modules/library/src/main/scala/zio/elasticsearch/executor/response/SearchWithAggregationsResponse.scala

@@ -72,6 +72,8 @@ private[elasticsearch] final case class SearchWithAggregationsResponse(
               res.fields.map { case (field, data) =>
                 ZValidation.fromEither(
                   (field: @unchecked) match {
+                    case str if str.contains("weighted_avg#") =>
+                      WeightedAvgAggregationResponse.decoder.decodeJson(data.toString).map(field.split("#")(1) -> _)
                     case str if str.contains("avg#") =>
                       AvgAggregationResponse.decoder.decodeJson(data.toString).map(field.split("#")(1) -> _)
                     case str if str.contains("max#") =>

modules/library/src/main/scala/zio/elasticsearch/package.scala

@@ -179,6 +179,19 @@ package object elasticsearch extends IndexNameNewtype with IndexPatternNewtype w
      */
     def asValueCountAggregation(name: String): RIO[R, Option[ValueCountAggregationResult]] =
       aggregationAs[ValueCountAggregationResult](name)
+
+    /**
+     * Executes the [[ElasticRequest.SearchRequest]] or the [[ElasticRequest.SearchAndAggregateRequest]].
+     *
+     * @param name
+     *   the name of the aggregation to retrieve
+     * @return
+     *   a [[RIO]] effect that, when executed, will produce the aggregation as instance of
+     *   [[result.WeightedAvgAggregationResult]].
+     */
+    def asWeightedAvgAggregation(name: String): RIO[R, Option[WeightedAvgAggregationResult]] =
+      aggregationAs[WeightedAvgAggregationResult](name)
+
   }
 
   final implicit class ZIODocumentOps[R, F[_]](zio: RIO[R, DocumentResult[F]]) {

modules/library/src/main/scala/zio/elasticsearch/result/AggregationResult.scala

@@ -70,3 +70,5 @@ final case class TermsAggregationBucketResult private[elasticsearch] (
 }
 
 final case class ValueCountAggregationResult private[elasticsearch] (value: Int) extends AggregationResult
+
+final case class WeightedAvgAggregationResult private[elasticsearch] (value: Double) extends AggregationResult

modules/library/src/main/scala/zio/elasticsearch/result/ElasticResult.scala

@@ -65,6 +65,9 @@ private[elasticsearch] sealed trait ResultWithAggregation {
 
   def asValueCountAggregation(name: String): IO[DecodingException, Option[ValueCountAggregationResult]] =
     aggregationAs[ValueCountAggregationResult](name)
+
+  def asWeightedAvgAggregation(name: String): IO[DecodingException, Option[WeightedAvgAggregationResult]] =
+    aggregationAs[WeightedAvgAggregationResult](name)
 }
 
 private[elasticsearch] sealed trait DocumentResult[F[_]] {