(dsl): Support Update By Query request

modules/library/src/it/scala/zio/elasticsearch/HttpExecutorSpec.scala

@@ -27,7 +27,7 @@ import zio.elasticsearch.query.sort.SortMode.Max
 import zio.elasticsearch.query.sort.SortOrder._
 import zio.elasticsearch.query.sort.SourceType.NumberType
 import zio.elasticsearch.request.{CreationOutcome, DeletionOutcome}
-import zio.elasticsearch.result.Item
+import zio.elasticsearch.result.{Item, UpdateByQueryResult}
 import zio.elasticsearch.script.Script
 import zio.json.ast.Json.{Arr, Str}
 import zio.stream.{Sink, ZSink}
@@ -1301,6 +1301,59 @@ object HttpExecutorSpec extends IntegrationSpec {
               } yield assert(doc)(isSome(equalTo(secondDocument)))
             }
           }
+        ),
+        suite("updating document by query")(
+          test("successfully update document with only script") {
+            checkOnce(genDocumentId, genTestDocument) { (documentId, document) =>
+              val stringField = "StringField"
+              for {
+                _ <- Executor.execute(ElasticRequest.deleteByQuery(updateByQueryIndex, matchAll).refreshTrue)
+                _ <- Executor.execute(
+                       ElasticRequest.upsert[TestDocument](updateByQueryIndex, documentId, document).refreshTrue
+                     )
+                updateRes <- Executor.execute(
+                               ElasticRequest
+                                 .updateAllByQuery(
+                                   updateByQueryIndex,
+                                   Script("ctx._source['stringField'] = params['str']").withParams("str" -> stringField)
+                                 )
+                                 .refreshTrue
+                             )
+                doc <- Executor.execute(ElasticRequest.getById(updateByQueryIndex, documentId)).documentAs[TestDocument]
+              } yield assert(updateRes)(
+                equalTo(
+                  UpdateByQueryResult(took = updateRes.took, total = 1, updated = 1, deleted = 0, versionConflicts = 0)
+                )
+              ) && assert(doc)(isSome(equalTo(document.copy(stringField = stringField))))
+            }
+          },
+          test("successfully update document with script and query") {
+            checkOnce(genDocumentId, genTestDocument) { (documentId, document) =>
+              val newDocument = document.copy(stringField = "StringField")
+              for {
+                _ <- Executor.execute(ElasticRequest.deleteByQuery(updateByQueryIndex, matchAll).refreshTrue)
+                _ <- Executor.execute(
+                       ElasticRequest.upsert[TestDocument](updateByQueryIndex, documentId, newDocument).refreshTrue
+                     )
+                updateRes <-
+                  Executor.execute(
+                    ElasticRequest
+                      .updateByQuery(
+                        index = updateByQueryIndex,
+                        query =
+                          term(field = TestDocument.stringField, multiField = Some("keyword"), value = "StringField"),
+                        script = Script("ctx._source['intField']++")
+                      )
+                      .refreshTrue
+                  )
+                doc <- Executor.execute(ElasticRequest.getById(updateByQueryIndex, documentId)).documentAs[TestDocument]
+              } yield assert(updateRes)(
+                equalTo(
+                  UpdateByQueryResult(took = updateRes.took, total = 1, updated = 1, deleted = 0, versionConflicts = 0)
+                )
+              ) && assert(doc)(isSome(equalTo(newDocument.copy(intField = newDocument.intField + 1))))
+            }
+          }
         )
       ) @@ nondeterministic @@ sequential @@ prepareElasticsearchIndexForTests @@ afterAll(
         Executor.execute(ElasticRequest.deleteIndex(index)).orDie

modules/library/src/it/scala/zio/elasticsearch/IntegrationSpec.scala

@@ -46,6 +46,8 @@ trait IntegrationSpec extends ZIOSpecDefault {
 
   val secondCountIndex: IndexName = IndexName("count-index-2")
 
+  val updateByQueryIndex: IndexName = IndexName("update-by-query-index")
+
   val prepareElasticsearchIndexForTests: TestAspect[Nothing, Any, Throwable, Any] = beforeAll((for {
     _ <- Executor.execute(ElasticRequest.createIndex(index))
     _ <- Executor.execute(ElasticRequest.deleteByQuery(index, matchAll).refreshTrue)

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

@@ -22,7 +22,13 @@ import zio.elasticsearch.highlights.Highlights
 import zio.elasticsearch.query.ElasticQuery
 import zio.elasticsearch.query.sort.Sort
 import zio.elasticsearch.request._
-import zio.elasticsearch.result.{AggregationResult, GetResult, SearchAndAggregateResult, SearchResult}
+import zio.elasticsearch.result.{
+  AggregationResult,
+  GetResult,
+  SearchAndAggregateResult,
+  SearchResult,
+  UpdateByQueryResult
+}
 import zio.elasticsearch.script.Script
 import zio.json.ast.Json
 import zio.json.ast.Json.{Arr, Obj}
@@ -109,6 +115,12 @@ object ElasticRequest {
       upsert = None
     )
 
+  def updateAllByQuery(index: IndexName, script: Script): UpdateByQueryRequest =
+    UpdateByQuery(index = index, script = script, conflicts = None, query = None, refresh = None, routing = None)
+
+  def updateByQuery(index: IndexName, query: ElasticQuery[_], script: Script): UpdateByQueryRequest =
+    UpdateByQuery(index = index, script = script, conflicts = None, query = Some(query), refresh = None, routing = None)
+
   def updateByScript(index: IndexName, id: DocumentId, script: Script): UpdateRequest =
     Update(index = index, id = id, doc = None, refresh = None, routing = None, script = Some(script), upsert = None)
 
@@ -481,6 +493,34 @@ object ElasticRequest {
     }
   }
 
+  sealed trait UpdateByQueryRequest
+      extends ElasticRequest[UpdateByQueryResult]
+      with HasRefresh[UpdateByQueryRequest]
+      with HasRouting[UpdateByQueryRequest] {
+    def conflicts(value: UpdateConflicts): UpdateByQueryRequest
+  }
+
+  private[elasticsearch] final case class UpdateByQuery(
+    index: IndexName,
+    script: Script,
+    conflicts: Option[UpdateConflicts],
+    query: Option[ElasticQuery[_]],
+    refresh: Option[Boolean],
+    routing: Option[Routing]
+  ) extends UpdateByQueryRequest { self =>
+    def conflicts(value: UpdateConflicts): UpdateByQueryRequest =
+      self.copy(conflicts = Some(value))
+
+    def refresh(value: Boolean): UpdateByQueryRequest =
+      self.copy(refresh = Some(value))
+
+    def routing(value: Routing): UpdateByQueryRequest =
+      self.copy(routing = Some(value))
+
+    def toJson: Json =
+      query.foldLeft(Obj("script" -> script.toJson))((scriptJson, q) => scriptJson merge q.toJson)
+  }
+
   private def getActionAndMeta(requestType: String, parameters: List[(String, Any)]): String =
     parameters.collect { case (name, Some(value)) => s""""$name" : "$value"""" }
       .mkString(s"""{ "$requestType" : { """, ", ", " } }")

modules/library/src/main/scala/zio/elasticsearch/executor/HttpExecutor.scala

@@ -32,7 +32,13 @@ import sttp.model.Uri.QuerySegment
 import zio.ZIO.logDebug
 import zio.elasticsearch.ElasticRequest._
 import zio.elasticsearch._
-import zio.elasticsearch.executor.response.{CountResponse, CreateResponse, GetResponse, SearchWithAggregationsResponse}
+import zio.elasticsearch.executor.response.{
+  CountResponse,
+  CreateResponse,
+  GetResponse,
+  SearchWithAggregationsResponse,
+  UpdateByQueryResponse
+}
 import zio.elasticsearch.request.{CreationOutcome, DeletionOutcome, UpdateOutcome}
 import zio.elasticsearch.result._
 import zio.json.ast.Json
@@ -71,6 +77,7 @@ private[elasticsearch] final class HttpExecutor private (esConfig: ElasticConfig
       case r: Search             => executeSearch(r)
       case r: SearchAndAggregate => executeSearchAndAggregate(r)
       case r: Update             => executeUpdate(r)
+      case r: UpdateByQuery      => executeUpdateByQuery(r)
     }
 
   def stream(r: SearchRequest): Stream[Throwable, Item] =
@@ -520,6 +527,34 @@ private[elasticsearch] final class HttpExecutor private (esConfig: ElasticConfig
       }
     }
 
+  private def executeUpdateByQuery(r: UpdateByQuery): Task[UpdateByQueryResult] =
+    sendRequestWithCustomResponse(
+      baseRequest
+        .post(
+          uri"${esConfig.uri}/${r.index}/$UpdateByQuery".withParams(
+            getQueryParams(List(("conflicts", r.conflicts), ("refresh", r.refresh), ("routing", r.routing)))
+          )
+        )
+        .response(asJson[UpdateByQueryResponse])
+        .contentType(ApplicationJson)
+        .body(r.toJson)
+    ).flatMap { response =>
+      response.code match {
+        case HttpOk =>
+          response.body.fold(
+            e => ZIO.fail(new ElasticException(s"Exception occurred: ${e.getMessage}")),
+            value => ZIO.succeed(UpdateByQueryResult.from(response = value))
+          )
+        case HttpConflict =>
+          response.body.fold(
+            e => ZIO.fail(new ElasticException(s"Exception occurred: ${e.getMessage}")),
+            value => ZIO.fail(VersionConflictException(value.updated + value.deleted, value.versionConflicts))
+          )
+        case _ =>
+          ZIO.fail(handleFailuresFromCustomResponse(response))
+      }
+    }
+
   private def getQueryParams(parameters: List[(String, Any)]): ScalaMap[String, String] =
     parameters.collect { case (name, Some(value)) => (name, value.toString) }.toMap
 
@@ -545,15 +580,15 @@ private[elasticsearch] final class HttpExecutor private (esConfig: ElasticConfig
         )
     }
 
-  private def itemFromResultsWithHighlights(results: List[(Json, Option[Json])]) =
+  private def itemFromResultsWithHighlights(results: List[(Json, Option[Json])]): Chunk[Item] =
     Chunk.fromIterable(results).map { case (source, highlight) =>
       Item(source, highlight)
     }
 
   private def itemFromResultsWithHighlightsAndInnerHits(
     results: List[(Json, Option[Json])],
     innerHits: List[Map[String, List[Json]]]
-  ) =
+  ): Chunk[Item] =
     Chunk.fromIterable(results).zip(innerHits).map { case ((source, highlight), innerHits) =>
       Item(source, highlight, innerHits)
     }
@@ -591,6 +626,7 @@ private[elasticsearch] object HttpExecutor {
   private final val Search        = "_search"
   private final val ShardDoc      = "_shard_doc"
   private final val Update        = "_update"
+  private final val UpdateByQuery = "_update_by_query"
 
   private[elasticsearch] final case class PointInTimeResponse(id: String)
   object PointInTimeResponse {

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2022 LambdaWorks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zio.elasticsearch.executor.response
+
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
+
+private[elasticsearch] final case class UpdateByQueryResponse(
+  took: Int,
+  total: Int,
+  updated: Int,
+  deleted: Int,
+  @jsonField("version_conflicts")
+  versionConflicts: Int
+)
+
+private[elasticsearch] object UpdateByQueryResponse {
+  implicit val decoder: JsonDecoder[UpdateByQueryResponse] = DeriveJsonDecoder.gen[UpdateByQueryResponse]
+}

modules/library/src/main/scala/zio/elasticsearch/query/Queries.scala

@@ -26,7 +26,7 @@ import scala.annotation.unused
 sealed trait ElasticQuery[-S] { self =>
   def paramsToJson(fieldPath: Option[String]): Json
 
-  final def toJson: Json =
+  final def toJson: Obj =
     Obj("query" -> self.paramsToJson(None))
 }
 

modules/library/src/main/scala/zio/elasticsearch/request/UpdateConflicts.scala

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2022 LambdaWorks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zio.elasticsearch.request
+
+sealed abstract class UpdateConflicts
+
+object UpdateConflicts {
+  case object Abort extends UpdateConflicts {
+    override def toString: String = "abort"
+  }
+
+  case object Proceed extends UpdateConflicts {
+    override def toString: String = "proceed"
+  }
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2022 LambdaWorks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zio.elasticsearch.result
+
+import zio.elasticsearch.executor.response.UpdateByQueryResponse
+
+final case class UpdateByQueryResult(
+  took: Int,
+  total: Int,
+  updated: Int,
+  deleted: Int,
+  versionConflicts: Int
+)
+
+object UpdateByQueryResult {
+  def from(response: UpdateByQueryResponse): UpdateByQueryResult =
+    UpdateByQueryResult(response.took, response.total, response.updated, response.deleted, response.versionConflicts)
+}

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

@@ -22,4 +22,9 @@ package object result {
   final case class DecodingException(message: String) extends ElasticException(message)
 
   case object UnauthorizedException extends ElasticException("Wrong credentials provided.")
+
+  final case class VersionConflictException(succeeded: Int, failed: Int)
+      extends ElasticException(
+        s"There are $failed documents failed due to version conflict, but $succeeded documents are updated successfully."
+      )
 }

modules/library/src/test/scala/zio/elasticsearch/ElasticRequestsDSLSpec.scala

@@ -2,6 +2,7 @@ package zio.elasticsearch
 
 import zio.elasticsearch.ElasticAggregation.termsAggregation
 import zio.elasticsearch.ElasticHighlight.highlight
+import zio.elasticsearch.ElasticQuery.term
 import zio.elasticsearch.ElasticRequest._
 import zio.elasticsearch.ElasticSort.sortBy
 import zio.elasticsearch.domain.TestDocument
@@ -197,6 +198,31 @@ object ElasticRequestsDSLSpec extends ZIOSpecDefault {
 
         assert(jsonRequest)(equalTo(expected.toJson))
       },
+      test("successfully encode update by query request with query to JSON") {
+        val jsonRequest = updateByQuery(
+          index = index,
+          query = term(TestDocument.stringField, Some("keyword"), "StringField"),
+          script = Script("ctx._source['intField']++")
+        ) match { case r: UpdateByQuery => r.toJson }
+
+        val expected =
+          """
+            |{
+            |  "script": {
+            |    "source": "ctx._source['intField']++"
+            |  },
+            |  "query": {
+            |    "term": {
+            |      "stringField.keyword": {
+            |        "value": "StringField"
+            |      }
+            |    }
+            |  }
+            |}
+            |""".stripMargin
+
+        assert(jsonRequest)(equalTo(expected.toJson))
+      },
       test("successfully encode update request to JSON with all parameters - script") {
         val jsonRequest = updateByScript(
           index = index,