Dependencies: Upgrade Elasticsearch to 8.9.0 (#574)

Co-authored-by: Fabrizio Fortino <[email protected]>

build.sbt

@@ -3,10 +3,10 @@ import ElasticsearchPluginPlugin.autoImport._
 Global / scalaVersion := "2.13.10"
 
 lazy val CirceVersion = "0.14.3"
-lazy val ElasticsearchVersion = "8.8.2"
-lazy val Elastic4sVersion = "8.8.2"
+lazy val ElasticsearchVersion = "8.9.0"
+lazy val Elastic4sVersion = "8.9.0"
 lazy val ElastiknnVersion = IO.read(file("version")).strip()
-lazy val LuceneVersion = "9.6.0"
+lazy val LuceneVersion = "9.7.0"
 
 lazy val ScalacOptions = List("-Xfatal-warnings", "-Ywarn-unused:imports")
 lazy val TestSettings = Seq(

client-python/requirements.txt

@@ -1,4 +1,4 @@
-elasticsearch==8.8.2
+elasticsearch==8.9.0
 dataclasses-json==0.3.7
 tqdm==4.61.1
 scipy==1.10.1

docker/Dockerfile

@@ -1,3 +1,3 @@
-FROM docker.elastic.co/elasticsearch/elasticsearch:8.8.2
+FROM docker.elastic.co/elasticsearch/elasticsearch:8.9.0
 COPY elastiknn-plugin/target/elastiknn*.zip .
 RUN elasticsearch-plugin install -b file:$(ls elastiknn*zip | sort | tail -n1)

docs/pages/installation.md

@@ -42,8 +42,8 @@ Make a Dockerfile like below.
 The image version (`elasticsearch:A.B.C`) must match the plugin's version (e.g. `A.B.C.x/elastiknn-A.B.C.x`).
 
 ```docker
-FROM docker.elastic.co/elasticsearch/elasticsearch:8.8.2
-RUN elasticsearch-plugin install --batch https://github.com/alexklibisz/elastiknn/releases/download/8.8.2.0/elastiknn-8.8.2.0.zip
+FROM docker.elastic.co/elasticsearch/elasticsearch:8.9.0
+RUN elasticsearch-plugin install --batch https://github.com/alexklibisz/elastiknn/releases/download/8.9.0.0/elastiknn-8.9.0.0.zip
 ```
 
 Build and run the Dockerfile. If you have any issues please refer to the [official docs.](https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html)

elastiknn-plugin/src/main/scala/com/klibisz/elastiknn/query/ElasticsearchQueryBuilder.scala

@@ -5,7 +5,6 @@ import com.klibisz.elastiknn.ElastiknnException.ElastiknnRuntimeException
 import com.klibisz.elastiknn.api._
 import com.klibisz.elastiknn.{ELASTIKNN_NAME, api}
 import org.apache.lucene.search.Query
-import org.apache.lucene.util.SetOnce
 import org.elasticsearch.action.ActionListener
 import org.elasticsearch.action.get.{GetAction, GetRequest, GetResponse}
 import org.elasticsearch.client.internal.Client
@@ -15,6 +14,7 @@ import org.elasticsearch.xcontent.{ToXContent, XContentBuilder, XContentParser}
 import org.elasticsearch.{ElasticsearchException, ResourceNotFoundException, TransportVersion}
 
 import java.util.Objects
+import java.util.concurrent.atomic.AtomicReference
 
 object ElasticsearchQueryBuilder {
 
@@ -44,6 +44,43 @@ object ElasticsearchQueryBuilder {
       new ElasticsearchQueryBuilder(query.withVec(sortedVec), elastiknnQueryBuilder)
     }
   }
+
+  // Used by rewriteGetVector to asynchronously provide an ElasticsearchQueryBuilder for an indexed vector.
+  // The doRewrite method will delay indefinitely until until the query builder is provided.
+  private final class IndexedVectorQueryBuilder(indexedVec: Vec.Indexed)
+    extends AbstractQueryBuilder[IndexedVectorQueryBuilder] {
+
+    private val ref = new AtomicReference[ElasticsearchQueryBuilder](null)
+
+    def set(qb: ElasticsearchQueryBuilder): Unit = ref.set(qb)
+
+    override def doRewrite(c: QueryRewriteContext): QueryBuilder = {
+      // If the reference has not been populated, we just return the current instance.
+      // Eventually the reference has been set, and we return real query builder.
+      // Elasticsearch will retry up to a fixed number of times until it gets the real one.
+      val maybe = ref.get()
+      if (maybe == null) this else maybe
+    }
+
+    def doWriteTo(out: StreamOutput): Unit =
+      throw new UnsupportedOperationException("Only supports doRewrite")
+
+    def doXContent(b: XContentBuilder, p: ToXContent.Params): Unit = ()
+
+    def doToQuery(context: SearchExecutionContext): Query =
+      throw new UnsupportedOperationException("Only supports doRewrite")
+
+    def doEquals(other: IndexedVectorQueryBuilder): Boolean =
+      throw new UnsupportedOperationException("Only supports doRewrite")
+
+    def doHashCode(): Int =
+      throw new UnsupportedOperationException("Only supports doRewrite")
+
+    def getWriteableName: String = s"IndexedVectorQueryBuilder(${indexedVec})"
+
+    def getMinimalSupportedVersion: TransportVersion =
+      throw new UnsupportedOperationException("Only supports doRewrite")
+  }
 }
 
 final class ElasticsearchQueryBuilder(val query: NearestNeighborsQuery, elastiknnQueryBuilder: ElastiknnQueryBuilder)
@@ -82,10 +119,11 @@ final class ElasticsearchQueryBuilder(val query: NearestNeighborsQuery, elastikn
     def unexpected(e: Exception): ElastiknnRuntimeException =
       new ElastiknnRuntimeException(s"Failed to retrieve vector at index [${ixv.index}] id [${ixv.id}] field [${ixv.field}]", e)
 
-    // This is basically an semaphore containing the constructed query.
-    val supplier = new SetOnce[ElasticsearchQueryBuilder]()
+    // Instantiate a placeholder query builder which will be provided with the real query builder after
+    // the indexed vector has been asynchronously fetched.
+    val queryBuilder = new ElasticsearchQueryBuilder.IndexedVectorQueryBuilder(ixv)
 
-    // Request the actual document in order to construct the query.
+    // Request the actual document in order to construct the query
     c.registerAsyncAction((client: Client, listener: ActionListener[_]) => {
       client.execute(
         GetAction.INSTANCE,
@@ -102,11 +140,11 @@ final class ElasticsearchQueryBuilder(val query: NearestNeighborsQuery, elastikn
               field match {
                 case map: java.util.Map[String @unchecked, Object @unchecked] if map.isInstanceOf[java.util.Map[String, Object]] =>
                   val vec = XContentCodec.decodeUnsafeFromMap[Vec](map)
-                  supplier.set(new ElasticsearchQueryBuilder(query.withVec(vec), elastiknnQueryBuilder))
+                  queryBuilder.set(new ElasticsearchQueryBuilder(query.withVec(vec), elastiknnQueryBuilder))
                   listener.asInstanceOf[ActionListener[Any]].onResponse(null)
                 case lst: java.util.List[Object @unchecked] if lst.isInstanceOf[java.util.List[Object]] =>
                   val vec = XContentCodec.decodeUnsafeFromList[Vec](lst)
-                  supplier.set(new ElasticsearchQueryBuilder(query.withVec(vec), elastiknnQueryBuilder))
+                  queryBuilder.set(new ElasticsearchQueryBuilder(query.withVec(vec), elastiknnQueryBuilder))
                   listener.asInstanceOf[ActionListener[Any]].onResponse(null)
                 case _ => listener.onFailure(unexpectedFieldType)
               }
@@ -120,7 +158,7 @@ final class ElasticsearchQueryBuilder(val query: NearestNeighborsQuery, elastikn
       )
     })
 
-    RewriteQueryBuilder(_ => supplier.get())
+    queryBuilder
   }
 
   override def getMinimalSupportedVersion: TransportVersion = TransportVersion.ZERO

version

@@ -1 +1 @@
-8.8.2.0
+8.9.0.0