Support versionType and pipeline parameters in reindex request (#3087)

* Add "externalgt" and "external_gt" cases * Support versionType parameter in reindex request * Support pipeline parameter in reindex request
Philippus · Jun 17, 2024 · a20926f · a20926f
1 parent e8ef111
commit a20926f
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 6 deletions.
diff --git a/...4s-core/src/test/scala/com/sksamuel/elastic4s/requests/reindex/ReindexBuilderFnTest.scala b/...4s-core/src/test/scala/com/sksamuel/elastic4s/requests/reindex/ReindexBuilderFnTest.scala
@@ -0,0 +1,24 @@
+package com.sksamuel.elastic4s.requests.reindex
+
+import com.sksamuel.elastic4s.handlers.reindex.ReindexBuilderFn
+import com.sksamuel.elastic4s.requests.common.VersionType.ExternalGte
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+
+class ReindexBuilderFnTest extends AnyFunSuite with Matchers {
+  import com.sksamuel.elastic4s.ElasticApi._
+
+  test("reindex content builder should support version type") {
+    val req = reindex("source", "target").versionType(ExternalGte)
+
+    ReindexBuilderFn(req).string shouldBe
+      """{"source":{"index":["source"]},"dest":{"index":"target","version_type":"external_gte"}}""".stripMargin
+  }
+
+  test("reindex content builder should support pipeline") {
+    val req = reindex("source", "target").pipeline("my_pipeline")
+
+    ReindexBuilderFn(req).string shouldBe
+      """{"source":{"index":["source"]},"dest":{"index":"target","pipeline":"my_pipeline"}}""".stripMargin
+  }
+}
diff --git a/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/requests/common/VersionType.scala b/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/requests/common/VersionType.scala
@@ -4,9 +4,9 @@ sealed trait VersionType
 object VersionType {
 
   def valueOf(str: String): VersionType = str.toLowerCase match {
-    case "external"                     => VersionType.External
-    case "externalgte" | "external_gte" => VersionType.ExternalGte
-    case _                              => VersionType.Internal
+    case "external" | "externalgt" | "external_gt" => VersionType.External
+    case "externalgte" | "external_gte"            => VersionType.ExternalGte
+    case _                                         => VersionType.Internal
   }
 
   case object External    extends VersionType

diff --git a/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/requests/reindex/ReindexRequest.scala b/elastic4s-domain/src/main/scala/com/sksamuel/elastic4s/requests/reindex/ReindexRequest.scala
@@ -1,11 +1,10 @@
 package com.sksamuel.elastic4s.requests.reindex
 
 import com.sksamuel.elastic4s.ext.OptionImplicits._
-import com.sksamuel.elastic4s.requests.common.{RefreshPolicy, Slice}
+import com.sksamuel.elastic4s.requests.common.{RefreshPolicy, Slice, VersionType}
 import com.sksamuel.elastic4s.requests.script.Script
 import com.sksamuel.elastic4s.requests.searches.queries.Query
 import com.sksamuel.elastic4s.{Index, Indexes}
-
 import scala.concurrent.duration.FiniteDuration
 
 case class ReindexRequest(sourceIndexes: Indexes,
@@ -30,7 +29,9 @@ case class ReindexRequest(sourceIndexes: Indexes,
                           size: Option[Int] = None,
                           createOnly: Option[Boolean] = None,
                           slices: Option[Int] = None,
-                          slice: Option[Slice] = None) {
+                          slice: Option[Slice] = None,
+                          versionType: Option[VersionType] = None,
+                          pipeline: Option[String] = None) {
 
   def remote(uri: String): ReindexRequest = copy(remoteHost = Option(uri))
   def remote(uri: String, user: String, pass: String): ReindexRequest =
@@ -73,4 +74,9 @@ case class ReindexRequest(sourceIndexes: Indexes,
   def createOnly(createOnly: Boolean): ReindexRequest = copy(createOnly = createOnly.some)
   def slice(slice: Slice): ReindexRequest = copy(slice = slice.some)
   def slices(slices: Int): ReindexRequest = copy(slices = slices.some)
+
+  def versionType(versionType: String): ReindexRequest = this.versionType(VersionType.valueOf(versionType))
+  def versionType(versionType: VersionType): ReindexRequest = copy(versionType = versionType.some)
+
+  def pipeline(pipeline: String): ReindexRequest = copy(pipeline = pipeline.some)
 }
diff --git a/...4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/reindex/ReindexBuilderFn.scala b/...4s-handlers/src/main/scala/com/sksamuel/elastic4s/handlers/reindex/ReindexBuilderFn.scala
@@ -49,11 +49,15 @@ object ReindexBuilderFn {
     builder.startObject("dest")
     builder.field("index", request.targetIndex.name)
 
+    request.versionType.foreach(versionType => builder.field("version_type", handlers.VersionTypeHttpString(versionType)))
+
     request.createOnly.foreach {
       case true => builder.field("op_type", "create")
       case false => builder.field("op_type", "index")
     }
 
+    request.pipeline.foreach(builder.field("pipeline", _))
+
     // end dest
     builder.endObject()
   }