From a57d49ed652ecc28372a31391f7f4bc6a14c7c51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 12 Aug 2026 09:49:10 +0200 Subject: [PATCH 1/3] perf(client): parse each Elasticsearch page once on the scroll hits path (softclient4es-arrow#160) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scroll / PIT+search_after paths parsed every page three times: the typed client parsed the HTTP response, convertToJson re-serialized the whole response to a JSON String, and core parseResponse re-parsed that string — ~19% of Flight sidecar CPU during JOIN leg extraction, scaling with the number and width of selected columns. This made a post-join aggregate's extra keyword column cost +9.8s over the bare join on the arrow#160 benchmark while the DuckDB aggregation itself cost nothing. Paging searches now run with document type ObjectNode so each _source is materialized once as the Jackson tree the row parser consumes; hitsToResponseNode re-parents those trees into a minimal envelope (_id + _source; whole-hit fallback via the new TokenBuffer-based convertToTree when inner_hits/fields are present), and callers use the node-level parseSingleSearchResponse. Aggregation-bearing responses go through convertToTree instead of StringWriter + readTree. One-shot search and msearch paths are unchanged. Measured on the benchmark corpus (overlay image, same machine/day): J0 40.0->28.3s, J2 45.0->31.1s; 10M-leg extraction 256.7k->364.5k docs/s (J0) and 227.1k->332.0k docs/s (J2); J2-J0 +5.0->+2.7s. Row oracles exact. es8+es9 suites 299/299 green on real ES 8.18.3/9.0.3; new JavaClientConversionSpec pins the envelope contract. Co-Authored-By: Claude Fable 5 --- .../elastic/client/java/JavaClientApi.scala | 37 +++-- .../client/java/JavaClientConversion.scala | 65 +++++++- .../java/JavaClientConversionSpec.scala | 148 ++++++++++++++++++ .../elastic/client/java/JavaClientApi.scala | 37 +++-- .../client/java/JavaClientConversion.scala | 65 +++++++- .../java/JavaClientConversionSpec.scala | 148 ++++++++++++++++++ 6 files changed, 464 insertions(+), 36 deletions(-) create mode 100644 es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala create mode 100644 es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index 95adb33e..bba5642a 100644 --- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -117,6 +117,7 @@ import co.elastic.clients.elasticsearch.watcher.{ WatchStatus } import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode import com.google.gson.JsonParser import _root_.java.io.{IOException, StringReader} @@ -1393,7 +1394,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { .size(config.scrollSize) .build() - val response = apply().search(searchRequest, classOf[JMap[String, Object]]) + val response = apply().search(searchRequest, classOf[ObjectNode]) if ( response.shards() != null && response @@ -1435,7 +1436,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { .scroll(Time.of(t => t.time(config.keepAlive))) .build() - val response = apply().scroll(scrollRequest, classOf[JMap[String, Object]]) + val response = apply().scroll(scrollRequest, classOf[ObjectNode]) if ( response.shards() != null && response @@ -1614,7 +1615,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { val response = apply().search( requestBuilder.build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) // Check errors @@ -1743,19 +1744,21 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { * BY, COUNT, AVG, etc.) */ private def extractAllResults( - response: Either[SearchResponse[JMap[String, Object]], ScrollResponse[JMap[String, Object]]], + response: Either[SearchResponse[ObjectNode], ScrollResponse[ObjectNode]], fieldAliases: ListMap[String, String], aggregations: ListMap[String, SQLAggregation], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - val jsonString = - response match { - case Left(l) => convertToJson(l) - case Right(r) => convertToJson(r) - } - - parseResponse( - jsonString, + // Single parse (softclient4es-arrow#160): hits-only pages — the scroll hot path — reuse the + // `_source` trees the transport already parsed; only aggregation-bearing responses (at most + // one per query) serialize the whole envelope, and even then at token level, never a string. + val aggs = response.fold(_.aggregations(), _.aggregations()) + val jsonNode: JsonNode = + if (aggs != null && !aggs.isEmpty) response.fold(convertToTree(_), convertToTree(_)) + else hitsToResponseNode(response.fold(_.hits().hits(), _.hits().hits())) + + parseSingleSearchResponse( + jsonNode, fieldAliases, aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)), retainDocumentId = retainDocumentId @@ -1772,14 +1775,14 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { /** Extract ONLY hits (for search_after optimization) Ignores aggregations for better performance */ private def extractHitsOnly( - response: SearchResponse[JMap[String, Object]], + response: SearchResponse[ObjectNode], fieldAliases: ListMap[String, String], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - val jsonString = convertToJson(response) - - parseResponse( - jsonString, + // Single parse (softclient4es-arrow#160): the `_source` trees were parsed once by the + // transport; re-parent them into the envelope instead of serializing and re-parsing. + parseSingleSearchResponse( + hitsToResponseNode(response.hits().hits()), fieldAliases, ListMap.empty, retainDocumentId = retainDocumentId diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala index c0197bcc..40b88d2a 100644 --- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala +++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala @@ -17,8 +17,12 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonpSerializable -import co.elastic.clients.json.jackson.JacksonJsonpMapper +import co.elastic.clients.json.jackson.{JacksonJsonpGenerator, JacksonJsonpMapper} +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode +import com.fasterxml.jackson.databind.util.TokenBuffer import java.io.{IOException, StringWriter} import scala.util.Try @@ -44,4 +48,63 @@ trait JavaClientConversion { _: JavaClientCompanion => } } } + + /** Convert any Elasticsearch response to a Jackson tree without going through a JSON string. + * + * The response is serialized into a [[TokenBuffer]] — a token-level copy with no character + * writing, no string escaping and no re-parsing — and the tree is read back from those tokens. + * On the scroll hot path the string round-trip (serialize to `StringWriter`, then + * `mapper.readTree`) was measured at ~19% of the sidecar CPU during JOIN leg extraction + * (softclient4es-arrow#160), dominated by per-character string writing and re-parsing costs that + * both scale with the number and width of the selected columns. + */ + protected def convertToTree[T <: JsonpSerializable](response: T): JsonNode = { + val buffer = new TokenBuffer(JacksonConfig.objectMapper, false) + val generator = new JacksonJsonpGenerator(buffer) + try { + response.serialize(generator, jsonpMapper) + generator.flush() + val parser = buffer.asParser() + try { + val tree: JsonNode = JacksonConfig.objectMapper.readTree(parser) + tree + } finally { + Try(parser.close()).failed.foreach { ex => + logger.warn(s"Failed to close token-buffer parser: ${ex.getMessage}") + } + } + } catch { + case ex: Exception => + logger.error(s"Failed to convert response to a Jackson tree: ${ex.getMessage}", ex) + throw new IOException("Failed to convert Elasticsearch response to a Jackson tree", ex) + } finally { + Try(generator.close()).failed.foreach { ex => + logger.warn(s"Failed to close JSON generator: ${ex.getMessage}") + } + } + } + + /** Build the minimal response-envelope tree the row parser consumes, from hits whose `_source` + * was already parsed as a Jackson tree by the transport (document type [[ObjectNode]]). + * + * This is the single-parse hits path (softclient4es-arrow#160): each `_source` node is + * re-parented into the envelope untouched — no serialization, no re-parse. The row parser reads + * exactly `_id`, `_source`, `inner_hits` and `fields` per hit, so hits carrying `inner_hits` or + * `fields` (UNNEST legs, script fields) fall back to a whole-hit [[convertToTree]] to keep full + * shape fidelity — still token-level, never a string. + */ + protected def hitsToResponseNode(hits: _root_.java.util.List[Hit[ObjectNode]]): ObjectNode = { + val root = JacksonConfig.objectMapper.createObjectNode() + val hitsArray = root.putObject("hits").putArray("hits") + hits.forEach { hit => + if (hit.innerHits().isEmpty && hit.fields().isEmpty) { + val hitNode = hitsArray.addObject() + Option(hit.id()).foreach(id => hitNode.put("_id", id)) + Option(hit.source()).foreach(source => hitNode.set[JsonNode]("_source", source)) + } else { + hitsArray.add(convertToTree(hit)) + } + } + root + } } diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala new file mode 100644 index 00000000..5beeef99 --- /dev/null +++ b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -0,0 +1,148 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * 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 app.softnetwork.elastic.client.java + +import app.softnetwork.elastic.client.ElasticConfig +import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.search.Hit +import co.elastic.clients.json.JsonData +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode +import com.typesafe.config.ConfigFactory +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import _root_.java.util.{Collections, List => JList} + +/** Pins the envelope contract of the single-parse hits path (softclient4es-arrow#160). + * + * The row parser (`ElasticConversion.parseSimpleHits`) reads exactly `_id`, `_source`, + * `inner_hits` and `fields` per hit — `hitsToResponseNode` must provide those and may omit + * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity + * node, and `convertToTree` must produce the same tree `convertToJson` round-trips to. + */ +class JavaClientConversionSpec extends AnyWordSpec with Matchers { + + private object Companion extends JavaClientCompanion with JavaClientConversion { + override def elasticConfig: ElasticConfig = ElasticConfig(ConfigFactory.load()) + def envelope(hits: JList[Hit[ObjectNode]]): ObjectNode = hitsToResponseNode(hits) + def tree(hit: Hit[ObjectNode]): JsonNode = convertToTree(hit) + def json(hit: Hit[ObjectNode]): String = convertToJson(hit) + } + + private val mapper = JacksonConfig.objectMapper + + private def sourceNode(): ObjectNode = { + val node = mapper.createObjectNode() + node.put("name", "a") + node.put("amount", 1.5) + node + } + + private def hitOf( + id: Option[String], + source: Option[ObjectNode], + fields: Map[String, JsonData] = Map.empty + ): Hit[ObjectNode] = + Hit.of[ObjectNode] { builder => + builder.index("idx") + id.foreach(builder.id) + source.foreach(builder.source) + fields.foreach { case (name, value) => + builder.fields(Collections.singletonMap(name, value)) + } + builder + } + + "hitsToResponseNode" should { + + "build the minimal envelope and re-parent the _source node untouched" in { + val src = sourceNode() + val envelope = Companion.envelope(Collections.singletonList(hitOf(Some("1"), Some(src)))) + + val hits = envelope.path("hits").path("hits") + hits.isArray shouldBe true + hits.size() shouldBe 1 + val hitNode = hits.get(0) + hitNode.path("_id").asText() shouldBe "1" + hitNode.get("_source") should be theSameInstanceAs src + hitNode.has("_index") shouldBe false + } + + "omit _id when the hit has none" in { + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(None, Some(sourceNode())))) + .path("hits") + .path("hits") + .get(0) + hitNode.has("_id") shouldBe false + hitNode.has("_source") shouldBe true + } + + "omit _source when the hit has none" in { + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(Some("2"), None))) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_id").asText() shouldBe "2" + hitNode.has("_source") shouldBe false + } + + "produce an empty hits array for no hits" in { + val hits = Companion + .envelope(Collections.emptyList[Hit[ObjectNode]]()) + .path("hits") + .path("hits") + hits.isArray shouldBe true + hits.size() shouldBe 0 + } + + "preserve explicit nulls in _source" in { + val src = mapper.createObjectNode() + src.putNull("maybe") + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(Some("3"), Some(src)))) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_source").get("maybe").isNull shouldBe true + } + + "fall back to a full-fidelity hit node when fields are present" in { + val src = sourceNode() + val hit = hitOf(Some("4"), Some(src), fields = Map("f" -> JsonData.of("v"))) + val hitNode = Companion + .envelope(Collections.singletonList(hit)) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_id").asText() shouldBe "4" + hitNode.path("_index").asText() shouldBe "idx" + hitNode.path("fields").path("f").asText() shouldBe "v" + hitNode.path("_source") shouldBe src + hitNode.path("_source") shouldNot be theSameInstanceAs src + } + } + + "convertToTree" should { + "produce the same tree convertToJson round-trips to" in { + val hit = hitOf(Some("5"), Some(sourceNode()), fields = Map("f" -> JsonData.of("v"))) + Companion.tree(hit) shouldBe mapper.readTree(Companion.json(hit)) + } + } +} diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index ae6bab25..6de60d3e 100644 --- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -116,6 +116,7 @@ import co.elastic.clients.elasticsearch.watcher.{ WatchStatus } import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode import com.google.gson.JsonParser import _root_.java.io.{IOException, StringReader} @@ -1393,7 +1394,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { .size(config.scrollSize) .build() - val response = apply().search(searchRequest, classOf[JMap[String, Object]]) + val response = apply().search(searchRequest, classOf[ObjectNode]) if ( response.shards() != null && response @@ -1435,7 +1436,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { .scroll(Time.of(t => t.time(config.keepAlive))) .build() - val response = apply().scroll(scrollRequest, classOf[JMap[String, Object]]) + val response = apply().scroll(scrollRequest, classOf[ObjectNode]) if ( response.shards() != null && response @@ -1614,7 +1615,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { val response = apply().search( requestBuilder.build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) // Check errors @@ -1743,19 +1744,21 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { * BY, COUNT, AVG, etc.) */ private def extractAllResults( - response: Either[SearchResponse[JMap[String, Object]], ScrollResponse[JMap[String, Object]]], + response: Either[SearchResponse[ObjectNode], ScrollResponse[ObjectNode]], fieldAliases: ListMap[String, String], aggregations: ListMap[String, SQLAggregation], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - val jsonString = - response match { - case Left(l) => convertToJson(l) - case Right(r) => convertToJson(r) - } - - parseResponse( - jsonString, + // Single parse (softclient4es-arrow#160): hits-only pages — the scroll hot path — reuse the + // `_source` trees the transport already parsed; only aggregation-bearing responses (at most + // one per query) serialize the whole envelope, and even then at token level, never a string. + val aggs = response.fold(_.aggregations(), _.aggregations()) + val jsonNode: JsonNode = + if (aggs != null && !aggs.isEmpty) response.fold(convertToTree(_), convertToTree(_)) + else hitsToResponseNode(response.fold(_.hits().hits(), _.hits().hits())) + + parseSingleSearchResponse( + jsonNode, fieldAliases, aggregations.map(kv => kv._1 -> kv._2), retainDocumentId = retainDocumentId @@ -1772,14 +1775,14 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { /** Extract ONLY hits (for search_after optimization) Ignores aggregations for better performance */ private def extractHitsOnly( - response: SearchResponse[JMap[String, Object]], + response: SearchResponse[ObjectNode], fieldAliases: ListMap[String, String], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - val jsonString = convertToJson(response) - - parseResponse( - jsonString, + // Single parse (softclient4es-arrow#160): the `_source` trees were parsed once by the + // transport; re-parent them into the envelope instead of serializing and re-parsing. + parseSingleSearchResponse( + hitsToResponseNode(response.hits().hits()), fieldAliases, ListMap.empty, retainDocumentId = retainDocumentId diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala index c0197bcc..40b88d2a 100644 --- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala +++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala @@ -17,8 +17,12 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonpSerializable -import co.elastic.clients.json.jackson.JacksonJsonpMapper +import co.elastic.clients.json.jackson.{JacksonJsonpGenerator, JacksonJsonpMapper} +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode +import com.fasterxml.jackson.databind.util.TokenBuffer import java.io.{IOException, StringWriter} import scala.util.Try @@ -44,4 +48,63 @@ trait JavaClientConversion { _: JavaClientCompanion => } } } + + /** Convert any Elasticsearch response to a Jackson tree without going through a JSON string. + * + * The response is serialized into a [[TokenBuffer]] — a token-level copy with no character + * writing, no string escaping and no re-parsing — and the tree is read back from those tokens. + * On the scroll hot path the string round-trip (serialize to `StringWriter`, then + * `mapper.readTree`) was measured at ~19% of the sidecar CPU during JOIN leg extraction + * (softclient4es-arrow#160), dominated by per-character string writing and re-parsing costs that + * both scale with the number and width of the selected columns. + */ + protected def convertToTree[T <: JsonpSerializable](response: T): JsonNode = { + val buffer = new TokenBuffer(JacksonConfig.objectMapper, false) + val generator = new JacksonJsonpGenerator(buffer) + try { + response.serialize(generator, jsonpMapper) + generator.flush() + val parser = buffer.asParser() + try { + val tree: JsonNode = JacksonConfig.objectMapper.readTree(parser) + tree + } finally { + Try(parser.close()).failed.foreach { ex => + logger.warn(s"Failed to close token-buffer parser: ${ex.getMessage}") + } + } + } catch { + case ex: Exception => + logger.error(s"Failed to convert response to a Jackson tree: ${ex.getMessage}", ex) + throw new IOException("Failed to convert Elasticsearch response to a Jackson tree", ex) + } finally { + Try(generator.close()).failed.foreach { ex => + logger.warn(s"Failed to close JSON generator: ${ex.getMessage}") + } + } + } + + /** Build the minimal response-envelope tree the row parser consumes, from hits whose `_source` + * was already parsed as a Jackson tree by the transport (document type [[ObjectNode]]). + * + * This is the single-parse hits path (softclient4es-arrow#160): each `_source` node is + * re-parented into the envelope untouched — no serialization, no re-parse. The row parser reads + * exactly `_id`, `_source`, `inner_hits` and `fields` per hit, so hits carrying `inner_hits` or + * `fields` (UNNEST legs, script fields) fall back to a whole-hit [[convertToTree]] to keep full + * shape fidelity — still token-level, never a string. + */ + protected def hitsToResponseNode(hits: _root_.java.util.List[Hit[ObjectNode]]): ObjectNode = { + val root = JacksonConfig.objectMapper.createObjectNode() + val hitsArray = root.putObject("hits").putArray("hits") + hits.forEach { hit => + if (hit.innerHits().isEmpty && hit.fields().isEmpty) { + val hitNode = hitsArray.addObject() + Option(hit.id()).foreach(id => hitNode.put("_id", id)) + Option(hit.source()).foreach(source => hitNode.set[JsonNode]("_source", source)) + } else { + hitsArray.add(convertToTree(hit)) + } + } + root + } } diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala new file mode 100644 index 00000000..5beeef99 --- /dev/null +++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -0,0 +1,148 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * 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 app.softnetwork.elastic.client.java + +import app.softnetwork.elastic.client.ElasticConfig +import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.search.Hit +import co.elastic.clients.json.JsonData +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.ObjectNode +import com.typesafe.config.ConfigFactory +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import _root_.java.util.{Collections, List => JList} + +/** Pins the envelope contract of the single-parse hits path (softclient4es-arrow#160). + * + * The row parser (`ElasticConversion.parseSimpleHits`) reads exactly `_id`, `_source`, + * `inner_hits` and `fields` per hit — `hitsToResponseNode` must provide those and may omit + * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity + * node, and `convertToTree` must produce the same tree `convertToJson` round-trips to. + */ +class JavaClientConversionSpec extends AnyWordSpec with Matchers { + + private object Companion extends JavaClientCompanion with JavaClientConversion { + override def elasticConfig: ElasticConfig = ElasticConfig(ConfigFactory.load()) + def envelope(hits: JList[Hit[ObjectNode]]): ObjectNode = hitsToResponseNode(hits) + def tree(hit: Hit[ObjectNode]): JsonNode = convertToTree(hit) + def json(hit: Hit[ObjectNode]): String = convertToJson(hit) + } + + private val mapper = JacksonConfig.objectMapper + + private def sourceNode(): ObjectNode = { + val node = mapper.createObjectNode() + node.put("name", "a") + node.put("amount", 1.5) + node + } + + private def hitOf( + id: Option[String], + source: Option[ObjectNode], + fields: Map[String, JsonData] = Map.empty + ): Hit[ObjectNode] = + Hit.of[ObjectNode] { builder => + builder.index("idx") + id.foreach(builder.id) + source.foreach(builder.source) + fields.foreach { case (name, value) => + builder.fields(Collections.singletonMap(name, value)) + } + builder + } + + "hitsToResponseNode" should { + + "build the minimal envelope and re-parent the _source node untouched" in { + val src = sourceNode() + val envelope = Companion.envelope(Collections.singletonList(hitOf(Some("1"), Some(src)))) + + val hits = envelope.path("hits").path("hits") + hits.isArray shouldBe true + hits.size() shouldBe 1 + val hitNode = hits.get(0) + hitNode.path("_id").asText() shouldBe "1" + hitNode.get("_source") should be theSameInstanceAs src + hitNode.has("_index") shouldBe false + } + + "omit _id when the hit has none" in { + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(None, Some(sourceNode())))) + .path("hits") + .path("hits") + .get(0) + hitNode.has("_id") shouldBe false + hitNode.has("_source") shouldBe true + } + + "omit _source when the hit has none" in { + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(Some("2"), None))) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_id").asText() shouldBe "2" + hitNode.has("_source") shouldBe false + } + + "produce an empty hits array for no hits" in { + val hits = Companion + .envelope(Collections.emptyList[Hit[ObjectNode]]()) + .path("hits") + .path("hits") + hits.isArray shouldBe true + hits.size() shouldBe 0 + } + + "preserve explicit nulls in _source" in { + val src = mapper.createObjectNode() + src.putNull("maybe") + val hitNode = Companion + .envelope(Collections.singletonList(hitOf(Some("3"), Some(src)))) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_source").get("maybe").isNull shouldBe true + } + + "fall back to a full-fidelity hit node when fields are present" in { + val src = sourceNode() + val hit = hitOf(Some("4"), Some(src), fields = Map("f" -> JsonData.of("v"))) + val hitNode = Companion + .envelope(Collections.singletonList(hit)) + .path("hits") + .path("hits") + .get(0) + hitNode.path("_id").asText() shouldBe "4" + hitNode.path("_index").asText() shouldBe "idx" + hitNode.path("fields").path("f").asText() shouldBe "v" + hitNode.path("_source") shouldBe src + hitNode.path("_source") shouldNot be theSameInstanceAs src + } + } + + "convertToTree" should { + "produce the same tree convertToJson round-trips to" in { + val hit = hitOf(Some("5"), Some(sourceNode()), fields = Map("f" -> JsonData.of("v"))) + Companion.tree(hit) shouldBe mapper.readTree(Companion.json(hit)) + } + } +} From bb3dde9e5f4f0c66d7e63decd12fe916bc41bad5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 12 Aug 2026 09:51:21 +0200 Subject: [PATCH 2/3] fix(tests): correct formatting in JavaClientConversionSpec documentation --- .../elastic/client/java/JavaClientConversionSpec.scala | 4 ++-- .../elastic/client/java/JavaClientConversionSpec.scala | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala index 5beeef99..844866c3 100644 --- a/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala +++ b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -32,8 +32,8 @@ import _root_.java.util.{Collections, List => JList} * * The row parser (`ElasticConversion.parseSimpleHits`) reads exactly `_id`, `_source`, * `inner_hits` and `fields` per hit — `hitsToResponseNode` must provide those and may omit - * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity - * node, and `convertToTree` must produce the same tree `convertToJson` round-trips to. + * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity node, + * and `convertToTree` must produce the same tree `convertToJson` round-trips to. */ class JavaClientConversionSpec extends AnyWordSpec with Matchers { diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala index 5beeef99..844866c3 100644 --- a/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala +++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -32,8 +32,8 @@ import _root_.java.util.{Collections, List => JList} * * The row parser (`ElasticConversion.parseSimpleHits`) reads exactly `_id`, `_source`, * `inner_hits` and `fields` per hit — `hitsToResponseNode` must provide those and may omit - * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity - * node, and `convertToTree` must produce the same tree `convertToJson` round-trips to. + * everything else. A hit carrying `inner_hits` or `fields` must fall back to a full-fidelity node, + * and `convertToTree` must produce the same tree `convertToJson` round-trips to. */ class JavaClientConversionSpec extends AnyWordSpec with Matchers { From 848ff277d8f6a6b0fd40064ba54d2cd9eed93590 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 12 Aug 2026 11:54:15 +0200 Subject: [PATCH 3/3] perf(client): parse every Elasticsearch response exactly once across all clients (#228) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The es6/es7 clients still processed every scroll page three times (typed parse, XContent/Gson re-serialization, Jackson re-parse), and every client double-parsed one-shot single and multi searches: the executor contract returned a JSON String that core parseResponse re-parsed. The same per-column extraction CPU that #227 removed from the es8/es9 scroll path (~19% of Flight sidecar CPU, softclient4es-arrow#160) was being paid on the REPL/JDBC 6.x/7.x paths and on every one-shot query. The executor contract now hands core an already-parsed Jackson tree (executeSingle/MultiSearch sync+async return Option[JsonNode]; core gains the node-level parseResponseTree dispatch, and parseInnerHits moved from Gson to Jackson): - es6/es7 REST: one-shot search, msearch and all scroll/search_after/ PIT paging now go through the low-level RestClient and the raw response entity bytes are Jackson-parsed once — one pass total. The typed builders still build request bodies; endpoints are percent-encoded like RequestConverters did, msearch sends UTF-8 bytes as bare application/json (ES 6.8 answers 406 to "application/x-ndjson; charset=UTF-8"), scroll ids / shard failures / search_after cursors are read from the tree, and pages with failed shards fail loudly like es8/es9 instead of silently losing rows. - es6 Jest: scroll pages and searches parse the retained raw body (getJsonString) — the Gson re-serialization pass is gone. - es8/es9: one-shot search/msearch use document type ObjectNode with the new searchResponseToTree/msearchResponseToTree (re-parented _source trees; token-level whole-envelope fallback for aggregation-bearing responses and msearch failure items). Review hardening (adversarial review, all fixes suite-validated): permanent 4xx on paging fails fast instead of burning retries (ResponseException is an IOException; 408/429/5xx stay retriable); a first page with failed shards releases the scroll context it created; a hit page without sort values aborts paging instead of silently restarting from page one; and paging streams now FAIL on error instead of ending quietly — a mid-scroll failure used to surface a silently truncated result set as a SUCCESSFUL result (same defect class as #209/#224), including pre-existing swallows on es8/es9. Suites on real ES, all green on the final code: core 746 unit tests, es6rest 303, es6jest 290, es7rest 305, es8java 316, es9java 316; 2.12+2.13 cross-compile; new parseResponseTree cases in ElasticConversionSpec and searchResponseToTree/msearchResponseToTree contracts in JavaClientConversionSpec (es8+es9). Closes #228 Co-Authored-By: Claude Fable 5 --- .../client/ElasticClientDelegator.scala | 9 +- .../elastic/client/ElasticConversion.scala | 28 +- .../elastic/client/NopeClientApi.scala | 9 +- .../elastic/client/SearchApi.scala | 89 ++-- .../client/ElasticConversionSpec.scala | 61 +++ .../elastic/client/jest/JestScrollApi.scala | 23 +- .../elastic/client/jest/JestSearchApi.scala | 37 +- .../client/rest/RestHighLevelClientApi.scala | 396 ++++++++++------ .../rest/RestHighLevelClientHelpers.scala | 102 ++++ .../client/rest/RestHighLevelClientApi.scala | 440 +++++++++++------- .../rest/RestHighLevelClientHelpers.scala | 102 ++++ .../elastic/client/java/JavaClientApi.scala | 47 +- .../client/java/JavaClientConversion.scala | 36 ++ .../java/JavaClientConversionSpec.scala | 86 ++++ .../elastic/client/java/JavaClientApi.scala | 47 +- .../client/java/JavaClientConversion.scala | 36 ++ .../java/JavaClientConversionSpec.scala | 86 ++++ .../elastic/client/MockElasticClientApi.scala | 21 +- 18 files changed, 1236 insertions(+), 419 deletions(-) diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala index b1adc145..48070d41 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala @@ -43,6 +43,7 @@ import app.softnetwork.elastic.sql.transform.{ } import app.softnetwork.elastic.sql.watcher.{Watcher, WatcherStatus} import com.typesafe.config.Config +import com.fasterxml.jackson.databind.JsonNode import org.apache.hadoop.conf.Configuration import org.json4s.Formats import org.slf4j.{Logger, LoggerFactory} @@ -1449,22 +1450,22 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes { override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = delegate.executeSingleSearch(elasticQuery) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = delegate.executeMultiSearch(elasticQueries) override private[client] def executeSingleSearchAsync(elasticQuery: ElasticQuery)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = + ): Future[ElasticResult[Option[JsonNode]]] = delegate.executeSingleSearchAsync(elasticQuery) override private[client] def executeMultiSearchAsync(elasticQueries: ElasticQueries)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = + ): Future[ElasticResult[Option[JsonNode]]] = delegate.executeMultiSearchAsync(elasticQueries) // ==================== ScrollApi ==================== diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala index 900f4991..43e3dae0 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala @@ -89,8 +89,34 @@ trait ElasticConversion { nestedHits: Map[String, Seq[(String, String)]] = Map.empty, explodeNested: Boolean = true, retainDocumentId: Boolean = false + )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] = + Try(mapper.readTree(results)).flatMap { json => + parseResponseTree( + json, + fieldAliases, + aggregations, + fields, + nestedHits, + explodeNested, + retainDocumentId + ) + } + + /** Node-level twin of [[parseResponse]] (#228): same single/multi dispatch, but the caller hands + * over an already-parsed Jackson tree. This is the entry the client executors use so each + * Elasticsearch response is parsed exactly once — never serialized back to a String for core to + * re-parse. + */ + def parseResponseTree( + results: JsonNode, + fieldAliases: ListMap[String, String], + aggregations: ListMap[String, ClientAggregation], + fields: Seq[String] = Seq.empty, + nestedHits: Map[String, Seq[(String, String)]] = Map.empty, + explodeNested: Boolean = true, + retainDocumentId: Boolean = false )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] = { - var json = mapper.readTree(results) + var json = results if (json.has("responses")) { json = json.get("responses") } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/NopeClientApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/NopeClientApi.scala index fa151763..4c729207 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/NopeClientApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/NopeClientApi.scala @@ -34,6 +34,7 @@ import app.softnetwork.elastic.sql.policy.{EnrichPolicy, EnrichPolicyTask, Enric import app.softnetwork.elastic.sql.schema.TableAlias import app.softnetwork.elastic.sql.transform.{TransformConfig, TransformStats} import app.softnetwork.elastic.sql.watcher.{Watcher, WatcherStatus} +import com.fasterxml.jackson.databind.JsonNode import scala.collection.immutable.ListMap import scala.concurrent.{ExecutionContext, Future} @@ -202,21 +203,21 @@ trait NopeClientApi extends ElasticClientApi { override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = ElasticResult.success(None) + ): ElasticResult[Option[JsonNode]] = ElasticResult.success(None) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = ElasticResult.success(None) + ): ElasticResult[Option[JsonNode]] = ElasticResult.success(None) override private[client] def executeSingleSearchAsync(elasticQuery: ElasticQuery)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = Future { + ): Future[ElasticResult[Option[JsonNode]]] = Future { ElasticResult.success(None) } override private[client] def executeMultiSearchAsync(elasticQueries: ElasticQueries)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = Future { + ): Future[ElasticResult[Option[JsonNode]]] = Future { ElasticResult.success(None) } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala index f58dbd4f..010ff048 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala @@ -36,7 +36,7 @@ import app.softnetwork.elastic.sql.query.{ SelectStatement, SingleSearch } -import com.google.gson.{Gson, JsonElement, JsonObject, JsonParser} +import com.fasterxml.jackson.databind.JsonNode import com.typesafe.config.ConfigFactory import org.json4s.Formats @@ -251,7 +251,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { ) val aggs = toClientAggregations(aggregations) ElasticResult.fromTry( - parseResponse( + parseResponseTree( response, fieldAliases, aggs, @@ -361,7 +361,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { ) val aggs = toClientAggregations(aggregations) ElasticResult.fromTry( - parseResponse( + parseResponseTree( response, fieldAliases, aggs, @@ -538,7 +538,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { ) val aggs = toClientAggregations(aggregations) ElasticResult.fromTry( - parseResponse( + parseResponseTree( response, fieldAliases, aggs, @@ -657,7 +657,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { ) val aggs = toClientAggregations(aggregations) ElasticResult.fromTry( - parseResponse( + parseResponseTree( response, fieldAliases, aggs, @@ -1065,9 +1065,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { logger.info( s"✅ Successfully executed search with inner hits in indices '${elasticQuery.indices.mkString(",")}'" ) - ElasticResult.attempt { - JsonParser.parseString(response).getAsJsonObject - } match { + ElasticResult.attempt(parseInnerHits[U, I](response, innerField)) match { case ElasticFailure(error) => logger.error( s"❌ Failed to parse Elasticsearch response for search with inner hits in indices '${elasticQuery.indices @@ -1079,8 +1077,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { index = Some(elasticQuery.indices.mkString(",")) ) ) - case ElasticSuccess(parsedResponse) => - ElasticResult.attempt(parseInnerHits[U, I](parsedResponse, innerField)) + case success => success } case ElasticSuccess(_) => val error = @@ -1154,9 +1151,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { logger.info( s"✅ Successfully executed multi-search inner hits with ${elasticQueries.queries.size} queries" ) - ElasticResult.attempt { - JsonParser.parseString(response).getAsJsonObject - } match { + ElasticResult.attempt(parseInnerHits[U, I](response, innerField)) match { case ElasticFailure(error) => logger.error( s"❌ Failed to parse Elasticsearch response for multi-search inner hits with ${elasticQueries.queries.size} queries: ${error.message}" @@ -1166,8 +1161,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { operation = Some("multisearchWithInnerHits") ) ) - case ElasticSuccess(parsedResponse) => - ElasticResult.attempt(parseInnerHits[U, I](parsedResponse, innerField)) + case success => success } case ElasticSuccess(_) => val error = @@ -1194,25 +1188,35 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { // METHODS TO IMPLEMENT // ======================================================================== + /** Execute the search and hand back the response as an already-parsed Jackson tree (#228). + * + * The tree contract kills the historical double parse: implementations must materialize the + * Elasticsearch response as the Jackson tree core consumes — parsing the raw response bytes + * exactly once, or re-parenting `_source` trees the transport already parsed — never by + * serializing a typed response back to a JSON String for core to re-parse. + */ private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] + ): ElasticResult[Option[JsonNode]] + /** @see [[executeSingleSearch]] for the single-parse tree contract (#228). */ private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] + ): ElasticResult[Option[JsonNode]] + /** @see [[executeSingleSearch]] for the single-parse tree contract (#228). */ private[client] def executeSingleSearchAsync( elasticQuery: ElasticQuery )(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] + ): Future[ElasticResult[Option[JsonNode]]] + /** @see [[executeSingleSearch]] for the single-parse tree contract (#228). */ private[client] def executeMultiSearchAsync( elasticQueries: ElasticQueries )(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] + ): Future[ElasticResult[Option[JsonNode]]] // ================================================================================ // IMPLICIT CONVERSIONS @@ -1231,7 +1235,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { ): String private def parseInnerHits[M: Manifest: ClassTag, I: Manifest: ClassTag]( - searchResult: JsonObject, + searchResult: JsonNode, innerField: String )(implicit formats: Formats): Seq[(M, Seq[I])] = { val mManifest = implicitly[Manifest[M]] @@ -1243,28 +1247,31 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { s"🔍 Processing inner hits with types: M=${mClass.getSimpleName}, I=${iClass.getSimpleName}" ) - def innerHits(result: JsonElement) = { - result.getAsJsonObject - .get("inner_hits") - .getAsJsonObject - .get(innerField) - .getAsJsonObject - .get("hits") - .getAsJsonObject - .get("hits") - .getAsJsonArray - .iterator() + def innerHits(result: JsonNode): Iterator[JsonNode] = { + val hits = result + .path("inner_hits") + .path(innerField) + .path("hits") + .path("hits") + if (!hits.isArray) { + throw new IllegalStateException( + s"No inner hits found for field '$innerField' in search response" + ) + } + hits.elements().asScala } - val gson = new Gson() - val results = searchResult.get("hits").getAsJsonObject.get("hits").getAsJsonArray.iterator() + val hits = searchResult.path("hits").path("hits") + if (!hits.isArray) { + throw new IllegalStateException("No hits found in search response") + } - (for (result <- results.asScala) + (for (result <- hits.elements().asScala) yield ( result match { - case obj: JsonObject => + case obj if obj.isObject => Try { - val source = gson.toJson(obj.get("_source")) + val source = mapper.writeValueAsString(obj.get("_source")) logger.debug( s"Deserializing main entity ${mClass.getSimpleName} from source: $source" ) @@ -1275,12 +1282,12 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { logger.error(s"❌ Failed to deserialize main entity: ${f.getMessage}", f) throw f } - case _ => serialization.read[M](result.getAsString)(formats, mManifest) + case other => serialization.read[M](other.asText())(formats, mManifest) }, - (for (innerHit <- innerHits(result).asScala) yield innerHit match { - case obj: JsonObject => + (for (innerHit <- innerHits(result)) yield innerHit match { + case obj if obj.isObject => Try { - val source = gson.toJson(obj.get("_source")) + val source = mapper.writeValueAsString(obj.get("_source")) logger.debug( s"Deserializing inner hit entity ${iClass.getSimpleName} from source: $source" ) @@ -1291,7 +1298,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { logger.error(s"❌ Failed to deserialize inner hit entity: ${f.getMessage}") throw f } - case _ => serialization.read[I](innerHit.getAsString)(formats, iManifest) + case other => serialization.read[I](other.asText())(formats, iManifest) }).toList )).toList } diff --git a/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala index c1234bfe..9d873acb 100644 --- a/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala +++ b/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala @@ -1752,6 +1752,67 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver innerHitKeys(this) shouldBe Set("total") innerHitKeys(EnabledDocumentIdConversion) shouldBe Set("total", "_id") } + + it should "dispatch an already-parsed single-search tree through parseResponseTree (#228)" in { + val tree = mapper.readTree( + """{ + | "took": 1, + | "hits": { + | "hits": [ + | { "_id": "1", "_source": { "name": "Laptop", "price": 999.99 } } + | ] + | } + |}""".stripMargin + ) + + parseResponseTree(tree, ListMap.empty, ListMap.empty) match { + case Success(rows) => + rows should have size 1 + rows.head("name") shouldBe "Laptop" + case Failure(error) => throw error + } + } + + it should "dispatch an already-parsed multi-search tree through parseResponseTree (#228)" in { + val tree = mapper.readTree( + """{ + | "responses": [ + | { "hits": { "hits": [ { "_id": "1", "_source": { "name": "Laptop" } } ] } }, + | { "hits": { "hits": [ { "_id": "2", "_source": { "name": "Mouse" } } ] } } + | ] + |}""".stripMargin + ) + + parseResponseTree(tree, ListMap.empty, ListMap.empty) match { + case Success(rows) => + rows.map(_("name")) should contain theSameElementsAs Seq("Laptop", "Mouse") + case Failure(error) => throw error + } + } + + it should "surface a multi-search item error from parseResponseTree (#228)" in { + val tree = mapper.readTree( + """{ + | "responses": [ + | { "hits": { "hits": [] } }, + | { "error": { "type": "search_phase_execution_exception", "reason": "boom" }, "status": 500 } + | ] + |}""".stripMargin + ) + + parseResponseTree(tree, ListMap.empty, ListMap.empty) match { + case Success(rows) => fail(s"Expected a failure, got $rows") + case Failure(error) => + error.getMessage should include("boom") + } + } + + it should "return a Failure on a malformed response string instead of throwing (#228)" in { + parseResponse("{ not json", ListMap.empty, ListMap.empty) match { + case Success(rows) => fail(s"Expected a failure, got $rows") + case Failure(_) => succeed + } + } } case class Products(category: String, top_products: List[Product], avg_price: Double) diff --git a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala index 311b386e..e40d7881 100644 --- a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala +++ b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala @@ -79,9 +79,11 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers { val scrollId = result.getJsonObject.get("_scroll_id").getAsString // Extract ALL results (hits + aggregations) + // Single parse for core (#228): Jackson reads the raw response body Jest + // retained — the Gson tree is only consulted for the scroll cursor. val results = extractAllResults( - result.getJsonObject.toString, + result.getJsonString, fieldAliases, aggregations, config.retainDocumentId @@ -110,7 +112,7 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers { val newScrollId = result.getJsonObject.get("_scroll_id").getAsString val results = extractAllResults( - result.getJsonObject.toString, + result.getJsonString, fieldAliases, aggregations, config.retainDocumentId @@ -126,10 +128,13 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers { } } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Scroll failed after retries: ${ex.getMessage}", ex) scrollIdOpt.foreach(clearScroll) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -214,8 +219,9 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers { throw new IOException(s"Search after failed: ${result.getErrorMessage}") } // Extract ONLY hits (no aggregations) + // Single parse for core (#228): raw body, not the Gson tree re-serialized val hits = - extractHitsOnly(result.getJsonObject.toString, fieldAliases, config.retainDocumentId) + extractHitsOnly(result.getJsonString, fieldAliases, config.retainDocumentId) if (hits.isEmpty) { None @@ -255,9 +261,12 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers { Some((nextSearchAfter, hits)) } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Search after failed after retries: ${ex.getMessage}", ex) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) diff --git a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestSearchApi.scala b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestSearchApi.scala index 7dad5f17..5a34f282 100644 --- a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestSearchApi.scala +++ b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestSearchApi.scala @@ -16,7 +16,14 @@ package app.softnetwork.elastic.client.jest -import app.softnetwork.elastic.client.{ElasticQueries, ElasticQuery, SearchApi, SerializationApi} +import app.softnetwork.elastic.client.{ + mapper, + ElasticQueries, + ElasticQuery, + SearchApi, + SerializationApi +} +import com.fasterxml.jackson.databind.JsonNode import app.softnetwork.elastic.client.result.ElasticResult import app.softnetwork.elastic.sql.PainlessContextType import app.softnetwork.elastic.sql.bridge.ElasticSearchRequest @@ -66,7 +73,7 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { override def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJestAction( operation = "executeSingleSearch", index = Some(elasticQuery.indices.mkString(",")), @@ -75,7 +82,10 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { elasticQuery.search._1 }(result => if (result.isSucceeded) { - Some(result.getJsonString) + // Single parse for core (#228): Jackson reads the raw response body Jest retained — + // never the Gson tree re-serialized to a String. (Jest's own Gson parse is forced by + // its transport and cannot be bypassed.) + Some(mapper.readTree(result.getJsonString)) } else { None } @@ -83,7 +93,7 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJestAction( operation = "executeMultiSearch", index = Some( @@ -99,7 +109,10 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { ).build() }(result => if (result.isSucceeded) { - Some(result.getJsonString) + // Single parse for core (#228): Jackson reads the raw response body Jest retained — + // never the Gson tree re-serialized to a String. (Jest's own Gson parse is forced by + // its transport and cannot be bypassed.) + Some(mapper.readTree(result.getJsonString)) } else { None } @@ -107,7 +120,7 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { override def executeSingleSearchAsync( elasticQuery: ElasticQuery - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = executeAsyncJestAction( operation = "executeSingleSearchAsync", index = Some(elasticQuery.indices.mkString(",")), @@ -116,7 +129,10 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { elasticQuery.search._1 }(result => if (result.isSucceeded) { - Some(result.getJsonString) + // Single parse for core (#228): Jackson reads the raw response body Jest retained — + // never the Gson tree re-serialized to a String. (Jest's own Gson parse is forced by + // its transport and cannot be bypassed.) + Some(mapper.readTree(result.getJsonString)) } else { None } @@ -126,7 +142,7 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { elasticQueries: ElasticQueries )(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = + ): Future[ElasticResult[Option[JsonNode]]] = executeAsyncJestAction( operation = "executeMultiSearchAsync", index = Some( @@ -142,7 +158,10 @@ trait JestSearchApi extends SearchApi with JestClientHelpers { ).build() }(result => if (result.isSucceeded) { - Some(result.getJsonString) + // Single parse for core (#228): Jackson reads the raw response body Jest retained — + // never the Gson tree re-serialized to a String. (Jest's own Gson parse is forced by + // its transport and cannot be bypassed.) + Some(mapper.readTree(result.getJsonString)) } else { None } diff --git a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala index 3324df10..fcb731c6 100644 --- a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala +++ b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala @@ -47,6 +47,8 @@ import app.softnetwork.elastic.utils.CronIntervalCalculator import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.node.ObjectNode import com.google.gson.JsonParser +import org.apache.http.entity.ContentType +import org.apache.http.nio.entity.NByteArrayEntity import org.apache.http.util.EntityUtils import org.elasticsearch.action.admin.indices.alias.{Alias, IndicesAliasesRequest} import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions @@ -69,14 +71,7 @@ import org.elasticsearch.action.ingest.{ GetPipelineResponse, PutPipelineRequest } -import org.elasticsearch.action.search.{ - ClearScrollRequest, - MultiSearchRequest, - MultiSearchResponse, - SearchRequest, - SearchResponse, - SearchScrollRequest -} +import org.elasticsearch.action.search.ClearScrollRequest import org.elasticsearch.action.support.{IndicesOptions, WriteRequest} import org.elasticsearch.action.support.master.AcknowledgedResponse import org.elasticsearch.action.update.{UpdateRequest, UpdateResponse} @@ -123,6 +118,7 @@ import org.json4s.DefaultFormats import org.json4s.jackson.JsonMethods import java.io.IOException +import java.nio.charset.StandardCharsets import java.time.ZoneId import java.util.Date import scala.collection.immutable.ListMap @@ -1053,40 +1049,102 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ): String = implicitly[ElasticSearchRequest](sqlSearch).query + // ========================================================================== + // Single-parse search execution (#228) + // + // The typed RestHighLevelClient path parsed every response three times: the + // client parsed the HTTP response into its typed form (pass 1), the typed + // form was re-serialized to a JSON String (pass 2) and core re-parsed that + // String into the Jackson tree the row parser consumes (pass 3). Searches now + // go through the low-level RestClient and the raw response entity bytes are + // Jackson-parsed exactly once into the tree core reads — one pass total. + // ========================================================================== + + /** `/{indices}/{types}/_search` — empty segments are skipped and each segment is percent-encoded, + * mirroring the high-level client's endpoint building (mapping types are still first-class on + * 6.x). Commas and wildcards stay raw; date-math characters are encoded. + */ + private[rest] def searchEndpoint(indices: Seq[String], types: Seq[String]): String = { + val parts = + Seq(indices, types) + .map(_.filter(_.nonEmpty)) + .filter(_.nonEmpty) + .map(seq => encodePathPart(seq.mkString(","))) + (parts :+ "_search").mkString("/", "/", "") + } + + /** Percent-encode one path segment exactly like the high-level client's RequestConverters: the + * segment is made absolute so a leading `-` or `:` cannot be misread, and any slash a segment + * carries (date-math rounding, e.g. ``) is encoded manually since URI treats it as + * a separator. + */ + private def encodePathPart(part: String): String = + new java.net.URI(null, null, null, -1, "/" + part, null, null).getRawPath + .substring(1) + .replaceAll("/", "%2F") + + /** Jackson-parse the raw response entity bytes — the single parse of the response. */ + private[rest] def readResponseTree(response: Response): JsonNode = + mapper.readTree(response.getEntity.getContent) + + /** NDJSON body of a `_msearch` request. Each query body is compacted through a Jackson round-trip + * so a formatted query cannot break the line-delimited protocol; the header carries the indices + * (and legacy types when present). + */ + private[rest] def msearchBody(queries: Seq[ElasticQuery]): String = { + val body = new StringBuilder + queries.foreach { q => + val header = mapper.createObjectNode() + val indicesArray = header.putArray("index") + q.indices.foreach(indicesArray.add) + if (q.types.nonEmpty) { + val typesArray = header.putArray("type") + q.types.foreach(typesArray.add) + } + body.append(mapper.writeValueAsString(header)).append('\n') + body.append(mapper.writeValueAsString(mapper.readTree(q.query))).append('\n') + } + body.toString + } + + private[rest] def msearchRequest(elasticQueries: ElasticQueries): Request = { + val req = new Request("POST", "/_msearch") + // Exactly what the typed client sent: UTF-8 bytes with a bare `application/json` content + // type. ES 6.x answers 406 to "application/x-ndjson; charset=UTF-8" (the charset parameter is + // not accepted on the NDJSON mime), and the NDJSON endpoints accept the JSON content type for + // line-delimited bodies on every version. + req.setEntity( + new NByteArrayEntity( + msearchBody(elasticQueries.queries).getBytes(StandardCharsets.UTF_8), + ContentType.create("application/json") + ) + ) + req + } + + private def searchRequest(elasticQuery: ElasticQuery): Request = { + val req = new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + req.setJsonEntity(elasticQuery.query) + req + } + override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = - executeRestAction[SearchRequest, SearchResponse, Option[String]]( + ): ElasticResult[Option[JsonNode]] = + executeRestLowLevelAction[Option[JsonNode]]( operation = "singleSearch", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( - request = { - val req = new SearchRequest(elasticQuery.indices: _*).types(elasticQuery.types: _*) - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - elasticQuery.query - ) - req.source(SearchSourceBuilder.fromXContent(xContentParser)) - req - } + request = searchRequest(elasticQuery) )( - executor = req => apply().search(req, RequestOptions.DEFAULT) - )(response => { - if (response.status() == RestStatus.OK) { - Some(Strings.toString(response)) - } else { - None - } - }) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = - executeRestAction[MultiSearchRequest, MultiSearchResponse, Option[String]]( + ): ElasticResult[Option[JsonNode]] = + executeRestLowLevelAction[Option[JsonNode]]( operation = "multiSearch", index = Some( elasticQueries.queries @@ -1096,63 +1154,28 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ), retryable = true )( - request = { - val req = new MultiSearchRequest() - for (query <- elasticQueries.queries) { - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - query.query - ) - val searchSourceBuilder = SearchSourceBuilder.fromXContent(xContentParser) - req.add( - new SearchRequest(query.indices: _*) - .types(query.types: _*) - .source(searchSourceBuilder) - ) - } - req - } + request = msearchRequest(elasticQueries) )( - executor = req => apply().msearch(req, RequestOptions.DEFAULT) - )(response => Some(Strings.toString(response))) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeSingleSearchAsync( elasticQuery: ElasticQuery - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = - executeAsyncRestAction[SearchRequest, SearchResponse, Option[String]]( + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = + executeAsyncRestLowLevelAction[Option[JsonNode]]( operation = "executeSingleSearchAsync", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( - request = { - val req = new SearchRequest(elasticQuery.indices: _*).types(elasticQuery.types: _*) - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - elasticQuery.query - ) - req.source(SearchSourceBuilder.fromXContent(xContentParser)) - req - } + request = searchRequest(elasticQuery) )( - executor = (req, listener) => apply().searchAsync(req, RequestOptions.DEFAULT, listener) - )(response => { - if (response.status() == RestStatus.OK) { - Some(Strings.toString(response)) - } else { - None - } - }) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeMultiSearchAsync( elasticQueries: ElasticQueries - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = - executeAsyncRestAction[MultiSearchRequest, MultiSearchResponse, Option[String]]( + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = + executeAsyncRestLowLevelAction[Option[JsonNode]]( operation = "executeMultiSearchAsync", index = Some( elasticQueries.queries @@ -1162,28 +1185,10 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ), retryable = true )( - request = { - val req = new MultiSearchRequest() - for (query <- elasticQueries.queries) { - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - query.query - ) - val searchSourceBuilder = SearchSourceBuilder.fromXContent(xContentParser) - req.add( - new SearchRequest(query.indices: _*) - .types(query.types: _*) - .source(searchSourceBuilder) - ) - } - req - } + request = msearchRequest(elasticQueries) )( - executor = (req, listener) => apply().msearchAsync(req, RequestOptions.DEFAULT, listener) - )(response => Some(Strings.toString(response))) + transformer = response => Some(readResponseTree(response)) + ) } @@ -1432,6 +1437,96 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel with RestHighLevelClientSearchApi with RestHighLevelClientCompanion => + // ========================================================================== + // Single-parse paging (#228) + // + // Every page used to be processed three times: the RestHighLevelClient + // parsed the HTTP response into its typed form, `response.toString` + // re-serialized that form via XContent, and core `parseResponse` re-parsed + // the string into the Jackson tree the row parser consumes. Pages now go + // through the low-level RestClient and the raw response entity bytes are + // Jackson-parsed exactly once — the typed request builders still build the + // request body, only the response side changed. + // ========================================================================== + + /** Execute a paging request through the low-level client and Jackson-parse the raw response + * entity bytes once. + * + * Error statuses surface as [[org.elasticsearch.client.ResponseException]] — an `IOException`, + * which `retryWithBackoff` would retry. The typed path never retried an error status + * (`ElasticsearchStatusException` is not retriable), so permanent client errors (4xx except + * 408/429) are rethrown as non-retriable to keep failing fast; retrying 408/429/5xx is a + * deliberate resilience gain on transient cluster conditions. + */ + private def executeSearchPage(request: Request): JsonNode = + try { + readResponseTree(apply().getLowLevelClient.performRequest(request)) + } catch { + case ex: org.elasticsearch.client.ResponseException => + val status = Try(ex.getResponse.getStatusLine.getStatusCode).getOrElse(0) + if (status >= 400 && status < 500 && status != 408 && status != 429) { + throw new IllegalStateException(ex.getMessage, ex) + } + throw ex + } + + /** Reasons of any failed shards on this page, if some shards failed. A page with failed shards is + * silent row loss on a paging path — callers must fail loudly, mirroring the es8/es9 typed shard + * check. + */ + private def shardFailures(tree: JsonNode): Option[String] = { + val shards = tree.path("_shards") + if (shards.path("failed").asInt(0) > 0) { + val failures = shards.path("failures") + val reasons = + if (failures.isArray && failures.size() > 0) { + failures + .elements() + .asScala + .map { failure => + val reason = failure.path("reason") + val message = reason.path("reason") + if (message.isTextual) message.asText() else reason.toString + } + .mkString("; ") + } else { + "Unknown shard failure" + } + Some(reasons) + } else { + None + } + } + + /** Convert a hit's `sort` array into the values fed back as `search_after` on the next page. */ + private def sortValuesOf(sortNode: JsonNode): Array[AnyRef] = + sortNode + .elements() + .asScala + .map { value => + val converted: AnyRef = + if (value.isTextual) value.textValue() + else if (value.isNumber) value.numberValue() + else if (value.isBoolean) java.lang.Boolean.valueOf(value.booleanValue()) + else if (value.isNull) null + else + // sort values are scalars by contract; a container would corrupt the cursor + throw new IllegalStateException( + s"Unsupported search_after sort value of type ${value.getNodeType}: $value" + ) + converted + } + .toArray + + private def scrollContinuationRequest(scrollId: String, keepAlive: String): Request = { + val body = mapper.createObjectNode() + body.put("scroll", keepAlive) + body.put("scroll_id", scrollId) + val request = new Request("POST", "/_search/scroll") + request.setJsonEntity(mapper.writeValueAsString(body)) + request + } + /** Classic scroll (works for both hits and aggregations) */ override private[client] def scrollClassic( @@ -1464,25 +1559,26 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel DeprecationHandler.THROW_UNSUPPORTED_OPERATION, query ) - // Execute the search - val searchRequest = - new SearchRequest(elasticQuery.indices: _*) - .types(elasticQuery.types: _*) - .source( - SearchSourceBuilder.fromXContent(xContentParser).size(config.scrollSize) - ) + val sourceBuilder = + SearchSourceBuilder.fromXContent(xContentParser).size(config.scrollSize) - searchRequest.scroll( - TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - ) + // Validate the keep-alive eagerly, exactly as the typed request builder did + TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - val response = apply().search(searchRequest, RequestOptions.DEFAULT) + val request = + new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + request.addParameter("scroll", config.keepAlive) + request.setJsonEntity(Strings.toString(sourceBuilder)) - if (response.status() != RestStatus.OK) { - throw new IOException(s"Initial scroll failed with status: ${response.status()}") - } + val tree = executeSearchPage(request) - val scrollId = response.getScrollId + val scrollId = tree.path("_scroll_id").textValue() + + shardFailures(tree).foreach { reasons => + // the failed request still created a server-side scroll context — release it + Option(scrollId).foreach(clearScroll) + throw new IOException(s"Initial scroll failed: $reasons") + } if (scrollId == null) { throw new IllegalStateException("Scroll ID is null in response") @@ -1491,7 +1587,7 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel // Extract both hits AND aggregations val results = extractAllResults( - response.toString, + tree, fieldAliases, aggregations, config.retainDocumentId @@ -1509,24 +1605,20 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel // Subsequent scroll requests logger.debug(s"Fetching next scroll batch (scrollId: $scrollId)") - val scrollRequest = new SearchScrollRequest(scrollId) - scrollRequest.scroll( - TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - ) - - val result = apply().scroll(scrollRequest, RequestOptions.DEFAULT) + val tree = + executeSearchPage(scrollContinuationRequest(scrollId, config.keepAlive)) - if (result.status() != RestStatus.OK) { + shardFailures(tree).foreach { reasons => + // the cursor is spent: a retry against a cleared context can only 404, and + // re-polling a scroll cursor skips rows — fail without retrying clearScroll(scrollId) - throw new IOException( - s"Scroll continuation failed with status: ${result.status()}" - ) + throw new IllegalStateException(s"Scroll continuation failed: $reasons") } - val newScrollId = result.getScrollId + val newScrollId = tree.path("_scroll_id").textValue() val results = extractAllResults( - result.toString, + tree, fieldAliases, aggregations, config.retainDocumentId @@ -1542,10 +1634,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel } } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Scroll failed after retries: ${ex.getMessage}", ex) scrollIdOpt.foreach(clearScroll) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -1621,29 +1716,34 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel sourceBuilder.searchAfter(searchAfter) } - // Execute the search - val searchRequest = - new SearchRequest(elasticQuery.indices: _*) - .types(elasticQuery.types: _*) - .source( - sourceBuilder - ) + // Execute the search (single parse #228: the raw response bytes are Jackson-parsed + // once; a non-OK status surfaces as a ResponseException) + val request = + new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + request.setJsonEntity(Strings.toString(sourceBuilder)) - val response = apply().search(searchRequest, RequestOptions.DEFAULT) + val tree = executeSearchPage(request) - if (response.status() != RestStatus.OK) { - throw new IOException(s"Search after failed with status: ${response.status()}") + shardFailures(tree).foreach { reasons => + throw new IOException(s"Search after failed: $reasons") } // Extract ONLY hits (no aggregations for search_after) - val hits = extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId) + val hits = extractHitsOnly(tree, fieldAliases, config.retainDocumentId) if (hits.isEmpty) { None } else { - val searchHits = response.getHits.getHits - val lastHit = searchHits.last - val nextSearchAfter = Option(lastHit.getSortValues) + val hitsArray = tree.path("hits").path("hits") + val lastHit = hitsArray.get(hitsArray.size() - 1) + val lastSort = lastHit.path("sort") + if (!lastSort.isArray || lastSort.size() == 0) { + // paging on without a cursor would refetch the same page forever + throw new IllegalStateException( + "search_after page returned hits without sort values — cannot continue paging" + ) + } + val nextSearchAfter = Some(sortValuesOf(lastSort)) logger.debug( s"Retrieved ${hits.size} hits, next search_after: ${nextSearchAfter @@ -1679,13 +1779,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel /** Extract ALL results: hits + aggregations This is crucial for queries with aggregations */ private def extractAllResults( - jsonString: String, + json: JsonNode, fieldAliases: ListMap[String, String], aggregations: ListMap[String, SQLAggregation], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - parseResponse( - jsonString, + parseSingleSearchResponse( + json, fieldAliases, aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)), retainDocumentId = retainDocumentId @@ -1702,12 +1802,12 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel /** Extract ONLY hits (for search_after optimization) */ private def extractHitsOnly( - jsonString: String, + json: JsonNode, fieldAliases: ListMap[String, String], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - parseResponse( - jsonString, + parseSingleSearchResponse( + json, fieldAliases, ListMap.empty, retainDocumentId = retainDocumentId diff --git a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala index 31f25554..ab40bbf0 100644 --- a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala +++ b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala @@ -296,6 +296,108 @@ trait RestHighLevelClientHelpers extends ElasticClientHelpers { _: RestHighLevel } } + /** Asynchronous variant of [[executeRestLowLevelAction]] (#228): same error mapping and status + * handling, driven by the low-level client's [[org.elasticsearch.client.ResponseListener]]. + */ + private[client] def executeAsyncRestLowLevelAction[T]( + operation: String, + index: Option[String] = None, + retryable: Boolean = true + )( + request: => org.elasticsearch.client.Request + )( + transformer: org.elasticsearch.client.Response => T + )(implicit ec: scala.concurrent.ExecutionContext): scala.concurrent.Future[ElasticResult[T]] = { + val indexStr = index.map(i => s" on index '$i'").getOrElse("") + logger.debug(s"Executing low-level operation '$operation'$indexStr asynchronously") + + val promise: Promise[ElasticResult[T]] = Promise() + + def transform(result: org.elasticsearch.client.Response): ElasticResult[T] = { + val statusCode = result.getStatusLine.getStatusCode + if (statusCode >= 200 && statusCode < 300) { + Try(transformer(result)) match { + case Success(transformed) => + logger.debug(s"Operation '$operation'$indexStr succeeded with status $statusCode") + ElasticResult.success(transformed) + case Failure(ex) => + logger.error(s"Transformation failed for operation '$operation'$indexStr", ex) + ElasticResult.failure( + ElasticError( + message = s"Failed to transform result: ${ex.getMessage}", + cause = Some(ex), + statusCode = Some(500), + operation = Some(operation) + ) + ) + } + } else { + val errorMessage = Option(result.getStatusLine.getReasonPhrase) + .filter(_.nonEmpty) + .getOrElse("Unknown error") + + val error = ElasticError( + message = errorMessage, + cause = None, + statusCode = Some(statusCode), + operation = Some(operation) + ) + + logError(operation, indexStr, error) + ElasticResult.failure(error) + } + } + + try { + val listener = new org.elasticsearch.client.ResponseListener { + override def onSuccess(response: org.elasticsearch.client.Response): Unit = + promise.success(transform(response)) + + override def onFailure(ex: Exception): Unit = { + val (message, statusCode) = ex match { + case respEx: org.elasticsearch.client.ResponseException => + ( + s"HTTP error during $operation: ${respEx.getMessage}", + Try(respEx.getResponse.getStatusLine.getStatusCode).toOption + ) + case _ => + (s"Exception during $operation: ${ex.getMessage}", None) + } + + logger.warn(s"Exception during operation '$operation'$indexStr: ${ex.getMessage}") + + promise.success( + ElasticResult.failure( + ElasticError( + message = message, + cause = Some(ex), + statusCode = statusCode, + operation = Some(operation) + ) + ) + ) + } + } + + apply().getLowLevelClient.performRequestAsync(request, listener) + } catch { + case ex: Exception => + logger.error(s"Failed to initiate async operation '$operation'$indexStr", ex) + promise.success( + ElasticResult.failure( + ElasticError( + message = s"Failed to initiate $operation: ${ex.getMessage}", + cause = Some(ex), + statusCode = None, + operation = Some(operation) + ) + ) + ) + } + + promise.future + } + //format:off /** Asynchronous variant to execute a Rest High Level Client action. * diff --git a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala index d58920ab..d25449df 100644 --- a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala +++ b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala @@ -59,6 +59,8 @@ import app.softnetwork.elastic.utils.CronIntervalCalculator import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode} import com.google.gson.JsonParser +import org.apache.http.entity.ContentType +import org.apache.http.nio.entity.NByteArrayEntity import org.apache.http.util.EntityUtils import org.elasticsearch.action.admin.indices.alias.Alias import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest @@ -84,12 +86,7 @@ import org.elasticsearch.action.ingest.{ import org.elasticsearch.action.search.{ ClearScrollRequest, ClosePointInTimeRequest, - MultiSearchRequest, - MultiSearchResponse, - OpenPointInTimeRequest, - SearchRequest, - SearchResponse, - SearchScrollRequest + OpenPointInTimeRequest } import org.elasticsearch.action.support.{IndicesOptions, WriteRequest} import org.elasticsearch.action.support.master.AcknowledgedResponse @@ -180,6 +177,7 @@ import org.json4s.jackson.JsonMethods import org.json4s.DefaultFormats import java.io.IOException +import java.nio.charset.StandardCharsets import java.time.ZonedDateTime import scala.collection.immutable.ListMap import scala.jdk.CollectionConverters._ @@ -1090,40 +1088,102 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ): String = implicitly[ElasticSearchRequest](singleSearch).query + // ========================================================================== + // Single-parse search execution (#228) + // + // The typed RestHighLevelClient path parsed every response three times: the + // client parsed the HTTP response into its typed form (pass 1), the typed + // form was re-serialized to a JSON String (pass 2) and core re-parsed that + // String into the Jackson tree the row parser consumes (pass 3). Searches now + // go through the low-level RestClient and the raw response entity bytes are + // Jackson-parsed exactly once into the tree core reads — one pass total. + // ========================================================================== + + /** `/{indices}/{types}/_search` — empty segments are skipped and each segment is percent-encoded, + * mirroring the high-level client's endpoint building (types are legacy 6.x mapping types, + * usually absent on 7.x). Commas and wildcards stay raw; date-math characters are encoded. + */ + private[rest] def searchEndpoint(indices: Seq[String], types: Seq[String]): String = { + val parts = + Seq(indices, types) + .map(_.filter(_.nonEmpty)) + .filter(_.nonEmpty) + .map(seq => encodePathPart(seq.mkString(","))) + (parts :+ "_search").mkString("/", "/", "") + } + + /** Percent-encode one path segment exactly like the high-level client's RequestConverters: the + * segment is made absolute so a leading `-` or `:` cannot be misread, and any slash a segment + * carries (date-math rounding, e.g. ``) is encoded manually since URI treats it as + * a separator. + */ + private def encodePathPart(part: String): String = + new java.net.URI(null, null, null, -1, "/" + part, null, null).getRawPath + .substring(1) + .replaceAll("/", "%2F") + + /** Jackson-parse the raw response entity bytes — the single parse of the response. */ + private[rest] def readResponseTree(response: Response): JsonNode = + mapper.readTree(response.getEntity.getContent) + + /** NDJSON body of a `_msearch` request. Each query body is compacted through a Jackson round-trip + * so a formatted query cannot break the line-delimited protocol; the header carries the indices + * (and legacy types when present). + */ + private[rest] def msearchBody(queries: Seq[ElasticQuery]): String = { + val body = new StringBuilder + queries.foreach { q => + val header = mapper.createObjectNode() + val indicesArray = header.putArray("index") + q.indices.foreach(indicesArray.add) + if (q.types.nonEmpty) { + val typesArray = header.putArray("type") + q.types.foreach(typesArray.add) + } + body.append(mapper.writeValueAsString(header)).append('\n') + body.append(mapper.writeValueAsString(mapper.readTree(q.query))).append('\n') + } + body.toString + } + + private[rest] def msearchRequest(elasticQueries: ElasticQueries): Request = { + val req = new Request("POST", "/_msearch") + // Exactly what the typed client sent: UTF-8 bytes with a bare `application/json` content + // type. ES 6.x answers 406 to "application/x-ndjson; charset=UTF-8" (the charset parameter is + // not accepted on the NDJSON mime), and the NDJSON endpoints accept the JSON content type for + // line-delimited bodies on every version. + req.setEntity( + new NByteArrayEntity( + msearchBody(elasticQueries.queries).getBytes(StandardCharsets.UTF_8), + ContentType.create("application/json") + ) + ) + req + } + + private def searchRequest(elasticQuery: ElasticQuery): Request = { + val req = new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + req.setJsonEntity(elasticQuery.query) + req + } + override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = - executeRestAction[SearchRequest, SearchResponse, Option[String]]( + ): ElasticResult[Option[JsonNode]] = + executeRestLowLevelAction[Option[JsonNode]]( operation = "singleSearch", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( - request = { - val req = new SearchRequest(elasticQuery.indices: _*).types(elasticQuery.types: _*) - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - elasticQuery.query - ) - req.source(SearchSourceBuilder.fromXContent(xContentParser)) - req - } + request = searchRequest(elasticQuery) )( - executor = req => apply().search(req, RequestOptions.DEFAULT) - )(response => { - if (response.status() == RestStatus.OK) { - Some(Strings.toString(response)) - } else { - None - } - }) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = - executeRestAction[MultiSearchRequest, MultiSearchResponse, Option[String]]( + ): ElasticResult[Option[JsonNode]] = + executeRestLowLevelAction[Option[JsonNode]]( operation = "multiSearch", index = Some( elasticQueries.queries @@ -1133,63 +1193,28 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ), retryable = true )( - request = { - val req = new MultiSearchRequest() - for (query <- elasticQueries.queries) { - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - query.query - ) - val searchSourceBuilder = SearchSourceBuilder.fromXContent(xContentParser) - req.add( - new SearchRequest(query.indices: _*) - .types(query.types: _*) - .source(searchSourceBuilder) - ) - } - req - } + request = msearchRequest(elasticQueries) )( - executor = req => apply().msearch(req, RequestOptions.DEFAULT) - )(response => Some(Strings.toString(response))) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeSingleSearchAsync( elasticQuery: ElasticQuery - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = - executeAsyncRestAction[SearchRequest, SearchResponse, Option[String]]( + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = + executeAsyncRestLowLevelAction[Option[JsonNode]]( operation = "executeSingleSearchAsync", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( - request = { - val req = new SearchRequest(elasticQuery.indices: _*).types(elasticQuery.types: _*) - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - elasticQuery.query - ) - req.source(SearchSourceBuilder.fromXContent(xContentParser)) - req - } + request = searchRequest(elasticQuery) )( - executor = (req, listener) => apply().searchAsync(req, RequestOptions.DEFAULT, listener) - )(response => { - if (response.status() == RestStatus.OK) { - Some(Strings.toString(response)) - } else { - None - } - }) + transformer = response => Some(readResponseTree(response)) + ) override private[client] def executeMultiSearchAsync( elasticQueries: ElasticQueries - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = - executeAsyncRestAction[MultiSearchRequest, MultiSearchResponse, Option[String]]( + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = + executeAsyncRestLowLevelAction[Option[JsonNode]]( operation = "executeMultiSearchAsync", index = Some( elasticQueries.queries @@ -1199,28 +1224,10 @@ trait RestHighLevelClientSearchApi extends SearchApi with RestHighLevelClientHel ), retryable = true )( - request = { - val req = new MultiSearchRequest() - for (query <- elasticQueries.queries) { - val xContentParser = XContentType.JSON - .xContent() - .createParser( - namedXContentRegistry, - DeprecationHandler.THROW_UNSUPPORTED_OPERATION, - query.query - ) - val searchSourceBuilder = SearchSourceBuilder.fromXContent(xContentParser) - req.add( - new SearchRequest(query.indices: _*) - .types(query.types: _*) - .source(searchSourceBuilder) - ) - } - req - } + request = msearchRequest(elasticQueries) )( - executor = (req, listener) => apply().msearchAsync(req, RequestOptions.DEFAULT, listener) - )(response => Some(Strings.toString(response))) + transformer = response => Some(readResponseTree(response)) + ) } @@ -1462,6 +1469,96 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel with RestHighLevelClientVersionApi with RestHighLevelClientCompanion => + // ========================================================================== + // Single-parse paging (#228) + // + // Every page used to be processed three times: the RestHighLevelClient + // parsed the HTTP response into its typed form, `response.toString` + // re-serialized that form via XContent, and core `parseResponse` re-parsed + // the string into the Jackson tree the row parser consumes. Pages now go + // through the low-level RestClient and the raw response entity bytes are + // Jackson-parsed exactly once — the typed request builders still build the + // request body, only the response side changed. + // ========================================================================== + + /** Execute a paging request through the low-level client and Jackson-parse the raw response + * entity bytes once. + * + * Error statuses surface as [[org.elasticsearch.client.ResponseException]] — an `IOException`, + * which `retryWithBackoff` would retry. The typed path never retried an error status + * (`ElasticsearchStatusException` is not retriable), so permanent client errors (4xx except + * 408/429) are rethrown as non-retriable to keep failing fast; retrying 408/429/5xx is a + * deliberate resilience gain on transient cluster conditions. + */ + private def executeSearchPage(request: Request): JsonNode = + try { + readResponseTree(apply().getLowLevelClient.performRequest(request)) + } catch { + case ex: org.elasticsearch.client.ResponseException => + val status = Try(ex.getResponse.getStatusLine.getStatusCode).getOrElse(0) + if (status >= 400 && status < 500 && status != 408 && status != 429) { + throw new IllegalStateException(ex.getMessage, ex) + } + throw ex + } + + /** Reasons of any failed shards on this page, if some shards failed. A page with failed shards is + * silent row loss on a paging path — callers must fail loudly, mirroring the es8/es9 typed shard + * check. + */ + private def shardFailures(tree: JsonNode): Option[String] = { + val shards = tree.path("_shards") + if (shards.path("failed").asInt(0) > 0) { + val failures = shards.path("failures") + val reasons = + if (failures.isArray && failures.size() > 0) { + failures + .elements() + .asScala + .map { failure => + val reason = failure.path("reason") + val message = reason.path("reason") + if (message.isTextual) message.asText() else reason.toString + } + .mkString("; ") + } else { + "Unknown shard failure" + } + Some(reasons) + } else { + None + } + } + + /** Convert a hit's `sort` array into the values fed back as `search_after` on the next page. */ + private def sortValuesOf(sortNode: JsonNode): Array[AnyRef] = + sortNode + .elements() + .asScala + .map { value => + val converted: AnyRef = + if (value.isTextual) value.textValue() + else if (value.isNumber) value.numberValue() + else if (value.isBoolean) java.lang.Boolean.valueOf(value.booleanValue()) + else if (value.isNull) null + else + // sort values are scalars by contract; a container would corrupt the cursor + throw new IllegalStateException( + s"Unsupported search_after sort value of type ${value.getNodeType}: $value" + ) + converted + } + .toArray + + private def scrollContinuationRequest(scrollId: String, keepAlive: String): Request = { + val body = mapper.createObjectNode() + body.put("scroll", keepAlive) + body.put("scroll_id", scrollId) + val request = new Request("POST", "/_search/scroll") + request.setJsonEntity(mapper.writeValueAsString(body)) + request + } + /** Classic scroll (works for both hits and aggregations) */ override private[client] def scrollClassic( @@ -1494,25 +1591,26 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel DeprecationHandler.THROW_UNSUPPORTED_OPERATION, query ) - // Execute the search - val searchRequest = - new SearchRequest(elasticQuery.indices: _*) - .types(elasticQuery.types: _*) - .source( - SearchSourceBuilder.fromXContent(xContentParser).size(config.scrollSize) - ) + val sourceBuilder = + SearchSourceBuilder.fromXContent(xContentParser).size(config.scrollSize) - searchRequest.scroll( - TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - ) + // Validate the keep-alive eagerly, exactly as the typed request builder did + TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - val response = apply().search(searchRequest, RequestOptions.DEFAULT) + val request = + new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + request.addParameter("scroll", config.keepAlive) + request.setJsonEntity(Strings.toString(sourceBuilder)) - if (response.status() != RestStatus.OK) { - throw new IOException(s"Initial scroll failed with status: ${response.status()}") - } + val tree = executeSearchPage(request) - val scrollId = response.getScrollId + val scrollId = tree.path("_scroll_id").textValue() + + shardFailures(tree).foreach { reasons => + // the failed request still created a server-side scroll context — release it + Option(scrollId).foreach(clearScroll) + throw new IOException(s"Initial scroll failed: $reasons") + } if (scrollId == null) { throw new IllegalStateException("Scroll ID is null in response") @@ -1521,7 +1619,7 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel // Extract both hits AND aggregations val results = extractAllResults( - response.toString, + tree, fieldAliases, aggregations, config.retainDocumentId @@ -1539,24 +1637,20 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel // Subsequent scroll requests logger.debug(s"Fetching next scroll batch (scrollId: $scrollId)") - val scrollRequest = new SearchScrollRequest(scrollId) - scrollRequest.scroll( - TimeValue.parseTimeValue(config.keepAlive, "scroll_timeout") - ) + val tree = + executeSearchPage(scrollContinuationRequest(scrollId, config.keepAlive)) - val result = apply().scroll(scrollRequest, RequestOptions.DEFAULT) - - if (result.status() != RestStatus.OK) { + shardFailures(tree).foreach { reasons => + // the cursor is spent: a retry against a cleared context can only 404, and + // re-polling a scroll cursor skips rows — fail without retrying clearScroll(scrollId) - throw new IOException( - s"Scroll continuation failed with status: ${result.status()}" - ) + throw new IllegalStateException(s"Scroll continuation failed: $reasons") } - val newScrollId = result.getScrollId + val newScrollId = tree.path("_scroll_id").textValue() val results = extractAllResults( - result.toString, + tree, fieldAliases, aggregations, config.retainDocumentId @@ -1572,10 +1666,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel } } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Scroll failed after retries: ${ex.getMessage}", ex) scrollIdOpt.foreach(clearScroll) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -1654,29 +1751,34 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel sourceBuilder.searchAfter(searchAfter) } - // Execute the search - val searchRequest = - new SearchRequest(elasticQuery.indices: _*) - .types(elasticQuery.types: _*) - .source( - sourceBuilder - ) + // Execute the search (single parse #228: the raw response bytes are Jackson-parsed + // once; a non-OK status surfaces as a ResponseException) + val request = + new Request("POST", searchEndpoint(elasticQuery.indices, elasticQuery.types)) + request.setJsonEntity(Strings.toString(sourceBuilder)) - val response = apply().search(searchRequest, RequestOptions.DEFAULT) + val tree = executeSearchPage(request) - if (response.status() != RestStatus.OK) { - throw new IOException(s"Search after failed with status: ${response.status()}") + shardFailures(tree).foreach { reasons => + throw new IOException(s"Search after failed: $reasons") } // Extract ONLY hits (no aggregations for search_after) - val hits = extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId) + val hits = extractHitsOnly(tree, fieldAliases, config.retainDocumentId) if (hits.isEmpty) { None } else { - val searchHits = response.getHits.getHits - val lastHit = searchHits.last - val nextSearchAfter = Option(lastHit.getSortValues) + val hitsArray = tree.path("hits").path("hits") + val lastHit = hitsArray.get(hitsArray.size() - 1) + val lastSort = lastHit.path("sort") + if (!lastSort.isArray || lastSort.size() == 0) { + // paging on without a cursor would refetch the same page forever + throw new IllegalStateException( + "search_after page returned hits without sort values — cannot continue paging" + ) + } + val nextSearchAfter = Some(sortValuesOf(lastSort)) logger.debug( s"Retrieved ${hits.size} hits, next search_after: ${nextSearchAfter @@ -1690,9 +1792,12 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel Some((nextSearchAfter, hits)) } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Search after failed after retries: ${ex.getMessage}", ex) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -1793,36 +1898,45 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel ) sourceBuilder.pointInTimeBuilder(pitBuilder) - // Build request with PIT - val searchRequest = new SearchRequest() - .source(sourceBuilder) - .requestCache(false) // Disable cache for PIT + // Build request with PIT — no index in the path, the PIT owns the target + // (single parse #228: raw response bytes Jackson-parsed once) + val request = new Request("POST", "/_search") + request.addParameter("request_cache", "false") // Disable cache for PIT + request.setJsonEntity(Strings.toString(sourceBuilder)) - val response = apply().search(searchRequest, RequestOptions.DEFAULT) + val tree = executeSearchPage(request) - if (response.status() != RestStatus.OK) { - throw new IOException( - s"PIT search_after failed with status: ${response.status()}" - ) + shardFailures(tree).foreach { reasons => + throw new IOException(s"PIT search_after failed: $reasons") } val hits = - extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId) + extractHitsOnly(tree, fieldAliases, config.retainDocumentId) if (hits.isEmpty) { None // end of stream — watchTermination owns the single PIT close (#202) } else { - val searchHits = response.getHits.getHits - val lastHit = searchHits.last - val nextSearchAfter = Option(lastHit.getSortValues) + val hitsArray = tree.path("hits").path("hits") + val lastHit = hitsArray.get(hitsArray.size() - 1) + val lastSort = lastHit.path("sort") + if (!lastSort.isArray || lastSort.size() == 0) { + // paging on without a cursor would refetch the same page forever + throw new IllegalStateException( + "search_after page returned hits without sort values — cannot continue paging" + ) + } + val nextSearchAfter = Some(sortValuesOf(lastSort)) logger.debug(s"Retrieved ${hits.size} hits, continuing with PIT") Some((nextSearchAfter, hits)) } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"PIT search_after failed after retries: ${ex.getMessage}", ex) - None // ends the stream — watchTermination owns the single PIT close (#202) + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class + // as #209/#224) — watchTermination still owns the single PIT close (#202) + Future.failed(ex) } } .watchTermination() { (_, done) => @@ -1897,13 +2011,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel /** Extract ALL results: hits + aggregations This is crucial for queries with aggregations */ private def extractAllResults( - jsonString: String, + json: JsonNode, fieldAliases: ListMap[String, String], aggregations: ListMap[String, SQLAggregation], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - parseResponse( - jsonString, + parseSingleSearchResponse( + json, fieldAliases, aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)), retainDocumentId = retainDocumentId @@ -1920,12 +2034,12 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel /** Extract ONLY hits (for search_after optimization) */ private def extractHitsOnly( - jsonString: String, + json: JsonNode, fieldAliases: ListMap[String, String], retainDocumentId: Boolean )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = { - parseResponse( - jsonString, + parseSingleSearchResponse( + json, fieldAliases, ListMap.empty, retainDocumentId = retainDocumentId diff --git a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala index 31f25554..ab40bbf0 100644 --- a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala +++ b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientHelpers.scala @@ -296,6 +296,108 @@ trait RestHighLevelClientHelpers extends ElasticClientHelpers { _: RestHighLevel } } + /** Asynchronous variant of [[executeRestLowLevelAction]] (#228): same error mapping and status + * handling, driven by the low-level client's [[org.elasticsearch.client.ResponseListener]]. + */ + private[client] def executeAsyncRestLowLevelAction[T]( + operation: String, + index: Option[String] = None, + retryable: Boolean = true + )( + request: => org.elasticsearch.client.Request + )( + transformer: org.elasticsearch.client.Response => T + )(implicit ec: scala.concurrent.ExecutionContext): scala.concurrent.Future[ElasticResult[T]] = { + val indexStr = index.map(i => s" on index '$i'").getOrElse("") + logger.debug(s"Executing low-level operation '$operation'$indexStr asynchronously") + + val promise: Promise[ElasticResult[T]] = Promise() + + def transform(result: org.elasticsearch.client.Response): ElasticResult[T] = { + val statusCode = result.getStatusLine.getStatusCode + if (statusCode >= 200 && statusCode < 300) { + Try(transformer(result)) match { + case Success(transformed) => + logger.debug(s"Operation '$operation'$indexStr succeeded with status $statusCode") + ElasticResult.success(transformed) + case Failure(ex) => + logger.error(s"Transformation failed for operation '$operation'$indexStr", ex) + ElasticResult.failure( + ElasticError( + message = s"Failed to transform result: ${ex.getMessage}", + cause = Some(ex), + statusCode = Some(500), + operation = Some(operation) + ) + ) + } + } else { + val errorMessage = Option(result.getStatusLine.getReasonPhrase) + .filter(_.nonEmpty) + .getOrElse("Unknown error") + + val error = ElasticError( + message = errorMessage, + cause = None, + statusCode = Some(statusCode), + operation = Some(operation) + ) + + logError(operation, indexStr, error) + ElasticResult.failure(error) + } + } + + try { + val listener = new org.elasticsearch.client.ResponseListener { + override def onSuccess(response: org.elasticsearch.client.Response): Unit = + promise.success(transform(response)) + + override def onFailure(ex: Exception): Unit = { + val (message, statusCode) = ex match { + case respEx: org.elasticsearch.client.ResponseException => + ( + s"HTTP error during $operation: ${respEx.getMessage}", + Try(respEx.getResponse.getStatusLine.getStatusCode).toOption + ) + case _ => + (s"Exception during $operation: ${ex.getMessage}", None) + } + + logger.warn(s"Exception during operation '$operation'$indexStr: ${ex.getMessage}") + + promise.success( + ElasticResult.failure( + ElasticError( + message = message, + cause = Some(ex), + statusCode = statusCode, + operation = Some(operation) + ) + ) + ) + } + } + + apply().getLowLevelClient.performRequestAsync(request, listener) + } catch { + case ex: Exception => + logger.error(s"Failed to initiate async operation '$operation'$indexStr", ex) + promise.success( + ElasticResult.failure( + ElasticError( + message = s"Failed to initiate $operation: ${ex.getMessage}", + cause = Some(ex), + statusCode = None, + operation = Some(operation) + ) + ) + ) + } + + promise.future + } + //format:off /** Asynchronous variant to execute a Rest High Level Client action. * diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index bba5642a..f7537bc7 100644 --- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -1022,12 +1022,14 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJavaAction( operation = "singleSearch", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( + // Single parse (#228): document type ObjectNode materializes each _source once as the + // Jackson tree core consumes — no typed-response-to-String round trip. apply() .search( new SearchRequest.Builder() @@ -1036,13 +1038,13 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { new StringReader(elasticQuery.query) ) .build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) - )(resp => Some(convertToJson(resp))) + )(resp => Some(searchResponseToTree(resp))) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJavaAction( operation = "multiSearch", index = Some(elasticQueries.queries.flatMap(_.indices).distinct.mkString(",")), @@ -1056,30 +1058,32 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { } val request = new MsearchRequest.Builder().searches(items.asJava).build() - apply().msearch(request, classOf[JMap[String, Object]]) - }(resp => Some(convertToJson(resp))) + // Single parse (#228) — see executeSingleSearch + apply().msearch(request, classOf[ObjectNode]) + }(resp => Some(msearchResponseToTree(resp))) override private[client] def executeSingleSearchAsync( elasticQuery: ElasticQuery - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = fromCompletableFuture( + // Single parse (#228) — see executeSingleSearch async() .search( new SearchRequest.Builder() .index(elasticQuery.indices.asJava) .withJson(new StringReader(elasticQuery.query)) .build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) ).map { response => - ElasticSuccess(Some(convertToJson(response))): ElasticResult[Option[String]] + ElasticSuccess(Some(searchResponseToTree(response))): ElasticResult[Option[JsonNode]] }.recover( asyncElasticFailure("singleSearch", Some(elasticQuery.indices.mkString(","))) ) override private[client] def executeMultiSearchAsync( elasticQueries: ElasticQueries - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = fromCompletableFuture { val items = elasticQueries.queries.map { q => new RequestItem.Builder() @@ -1089,10 +1093,11 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { } val request = new MsearchRequest.Builder().searches(items.asJava).build() - async().msearch(request, classOf[JMap[String, Object]]) + // Single parse (#228) — see executeSingleSearch + async().msearch(request, classOf[ObjectNode]) } .map { response => - ElasticSuccess(Some(convertToJson(response))): ElasticResult[Option[String]] + ElasticSuccess(Some(msearchResponseToTree(response))): ElasticResult[Option[JsonNode]] } .recover( asyncElasticFailure( @@ -1450,7 +1455,9 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { } else { "Unknown shard failure" } - throw new IOException(s"Scroll continuation failed: $errorMsg") + // the cursor is spent: a retry against a cleared context can only 404, and + // re-polling a scroll cursor skips rows — fail without retrying + throw new IllegalStateException(s"Scroll continuation failed: $errorMsg") } val newScrollId = response.scrollId() @@ -1470,10 +1477,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { } } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Scroll failed after retries: ${ex.getMessage}", ex) scrollIdOpt.foreach(clearScroll) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -1659,9 +1669,12 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { Some((nextSearchAfter, hits)) } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"PIT search_after failed after retries: ${ex.getMessage}", ex) - None // ends the stream — watchTermination owns the single PIT close (#202) + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class + // as #209/#224) — watchTermination still owns the single PIT close (#202) + Future.failed(ex) } } .watchTermination() { (_, done) => diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala index 40b88d2a..7468e8a9 100644 --- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala +++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala @@ -17,6 +17,7 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.{MsearchResponse, SearchResponse} import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonpSerializable import co.elastic.clients.json.jackson.{JacksonJsonpGenerator, JacksonJsonpMapper} @@ -107,4 +108,39 @@ trait JavaClientConversion { _: JavaClientCompanion => } root } + + /** Response tree of a one-shot search (#228), single-parse. + * + * Hits-only responses — the common row-shaped case — re-parent the `_source` trees the transport + * already parsed (see [[hitsToResponseNode]]); aggregation-bearing responses serialize the whole + * envelope once at token level via [[convertToTree]], never through a String. + */ + protected def searchResponseToTree(response: SearchResponse[ObjectNode]): JsonNode = + if (response.aggregations() != null && !response.aggregations().isEmpty) + convertToTree(response) + else hitsToResponseNode(response.hits().hits()) + + /** Response tree of a multi search (#228), single-parse. + * + * The `responses` array is rebuilt item by item with the same policy as + * [[searchResponseToTree]]; a failed item keeps full fidelity so core still sees its `error` + * object. + */ + protected def msearchResponseToTree(response: MsearchResponse[ObjectNode]): JsonNode = { + val root = JacksonConfig.objectMapper.createObjectNode() + val responses = root.putArray("responses") + response.responses().forEach { item => + if (item.isResult) { + val result = item.result() + if (result.aggregations() != null && !result.aggregations().isEmpty) { + responses.add(convertToTree(result)) + } else { + responses.add(hitsToResponseNode(result.hits().hits())) + } + } else { + responses.add(convertToTree(item)) + } + } + root + } } diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala index 844866c3..5ed77c63 100644 --- a/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala +++ b/es8/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -18,6 +18,9 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.client.ElasticConfig import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch._types.aggregations.Aggregate +import co.elastic.clients.elasticsearch.core.{MsearchResponse, SearchResponse} +import co.elastic.clients.elasticsearch.core.msearch.MultiSearchResponseItem import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonData import com.fasterxml.jackson.databind.JsonNode @@ -42,6 +45,10 @@ class JavaClientConversionSpec extends AnyWordSpec with Matchers { def envelope(hits: JList[Hit[ObjectNode]]): ObjectNode = hitsToResponseNode(hits) def tree(hit: Hit[ObjectNode]): JsonNode = convertToTree(hit) def json(hit: Hit[ObjectNode]): String = convertToJson(hit) + def searchTree(response: SearchResponse[ObjectNode]): JsonNode = searchResponseToTree(response) + def msearchTree(response: MsearchResponse[ObjectNode]): JsonNode = + msearchResponseToTree(response) + def searchJson(response: SearchResponse[ObjectNode]): String = convertToJson(response) } private val mapper = JacksonConfig.objectMapper @@ -145,4 +152,83 @@ class JavaClientConversionSpec extends AnyWordSpec with Matchers { Companion.tree(hit) shouldBe mapper.readTree(Companion.json(hit)) } } + + private def searchResponseOf( + hits: JList[Hit[ObjectNode]], + aggregations: Map[String, Aggregate] = Map.empty + ): SearchResponse[ObjectNode] = + SearchResponse.of[ObjectNode] { builder => + builder + .took(1) + .timedOut(false) + .shards(s => s.total(1).successful(1).failed(0)) + .hits(h => h.hits(hits)) + aggregations.foreach { case (name, aggregate) => builder.aggregations(name, aggregate) } + builder + } + + "searchResponseToTree" should { + + "re-parent hits through the minimal envelope when the response has no aggregations" in { + val src = sourceNode() + val tree = Companion.searchTree( + searchResponseOf(Collections.singletonList(hitOf(Some("1"), Some(src)))) + ) + val hitNode = tree.path("hits").path("hits").get(0) + hitNode.path("_id").asText() shouldBe "1" + hitNode.get("_source") should be theSameInstanceAs src + tree.has("took") shouldBe false + } + + "keep the exact convertToJson shape when aggregations are present" in { + val response = searchResponseOf( + Collections.singletonList(hitOf(Some("1"), Some(sourceNode()))), + aggregations = Map("avg_price" -> Aggregate.of(a => a.avg(v => v.value(2.5)))) + ) + // normalize both sides through the same parse: the token-level tree may carry + // LongNode/IntNode artifacts that are value-equal but not node-class-equal + mapper.readTree(Companion.searchTree(response).toString) shouldBe + mapper.readTree(Companion.searchJson(response)) + } + } + + "msearchResponseToTree" should { + + "rebuild the responses array with per-item envelopes and full-fidelity failures" in { + val src = sourceNode() + val response = MsearchResponse.of[ObjectNode] { builder => + builder + .took(1) + .responses( + MultiSearchResponseItem.of[ObjectNode](item => + item.result(r => + r.took(1) + .timedOut(false) + .shards(s => s.total(1).successful(1).failed(0)) + .hits(h => h.hits(Collections.singletonList(hitOf(Some("1"), Some(src))))) + .status(200) + ) + ), + MultiSearchResponseItem.of[ObjectNode](item => + item.failure(f => + f.error(e => e.`type`("search_phase_execution_exception").reason("boom")) + .status(500) + ) + ) + ) + } + + val tree = Companion.msearchTree(response) + val responses = tree.path("responses") + responses.isArray shouldBe true + responses.size() shouldBe 2 + + val first = responses.get(0) + first.path("hits").path("hits").get(0).get("_source") should be theSameInstanceAs src + + val second = responses.get(1) + second.path("error").path("reason").asText() shouldBe "boom" + second.path("status").asInt() shouldBe 500 + } + } } diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala index 6de60d3e..49268597 100644 --- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala +++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala @@ -1022,12 +1022,14 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJavaAction( operation = "singleSearch", index = Some(elasticQuery.indices.mkString(",")), retryable = true )( + // Single parse (#228): document type ObjectNode materializes each _source once as the + // Jackson tree core consumes — no typed-response-to-String round trip. apply() .search( new SearchRequest.Builder() @@ -1036,13 +1038,13 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { new StringReader(elasticQuery.query) ) .build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) - )(resp => Some(convertToJson(resp))) + )(resp => Some(searchResponseToTree(resp))) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = executeJavaAction( operation = "multiSearch", index = Some(elasticQueries.queries.flatMap(_.indices).distinct.mkString(",")), @@ -1056,30 +1058,32 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { } val request = new MsearchRequest.Builder().searches(items.asJava).build() - apply().msearch(request, classOf[JMap[String, Object]]) - }(resp => Some(convertToJson(resp))) + // Single parse (#228) — see executeSingleSearch + apply().msearch(request, classOf[ObjectNode]) + }(resp => Some(msearchResponseToTree(resp))) override private[client] def executeSingleSearchAsync( elasticQuery: ElasticQuery - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = fromCompletableFuture( + // Single parse (#228) — see executeSingleSearch async() .search( new SearchRequest.Builder() .index(elasticQuery.indices.asJava) .withJson(new StringReader(elasticQuery.query)) .build(), - classOf[JMap[String, Object]] + classOf[ObjectNode] ) ).map { response => - ElasticSuccess(Some(convertToJson(response))): ElasticResult[Option[String]] + ElasticSuccess(Some(searchResponseToTree(response))): ElasticResult[Option[JsonNode]] }.recover( asyncElasticFailure("singleSearch", Some(elasticQuery.indices.mkString(","))) ) override private[client] def executeMultiSearchAsync( elasticQueries: ElasticQueries - )(implicit ec: ExecutionContext): Future[ElasticResult[Option[String]]] = + )(implicit ec: ExecutionContext): Future[ElasticResult[Option[JsonNode]]] = fromCompletableFuture { val items = elasticQueries.queries.map { q => new RequestItem.Builder() @@ -1089,10 +1093,11 @@ trait JavaClientSearchApi extends SearchApi with JavaClientHelpers { } val request = new MsearchRequest.Builder().searches(items.asJava).build() - async().msearch(request, classOf[JMap[String, Object]]) + // Single parse (#228) — see executeSingleSearch + async().msearch(request, classOf[ObjectNode]) } .map { response => - ElasticSuccess(Some(convertToJson(response))): ElasticResult[Option[String]] + ElasticSuccess(Some(msearchResponseToTree(response))): ElasticResult[Option[JsonNode]] } .recover( asyncElasticFailure( @@ -1450,7 +1455,9 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { } else { "Unknown shard failure" } - throw new IOException(s"Scroll continuation failed: $errorMsg") + // the cursor is spent: a retry against a cleared context can only 404, and + // re-polling a scroll cursor skips rows — fail without retrying + throw new IllegalStateException(s"Scroll continuation failed: $errorMsg") } val newScrollId = response.scrollId() @@ -1470,10 +1477,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { } } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"Scroll failed after retries: ${ex.getMessage}", ex) scrollIdOpt.foreach(clearScroll) - None + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class as + // #209/#224) + Future.failed(ex) } } .mapConcat(identity) @@ -1659,9 +1669,12 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers { Some((nextSearchAfter, hits)) } } - }(system, logger).recover { case ex: Exception => + }(system, logger).recoverWith { case ex: Exception => logger.error(s"PIT search_after failed after retries: ${ex.getMessage}", ex) - None // ends the stream — watchTermination owns the single PIT close (#202) + // fail the stream instead of ending it: ending here would surface a silently + // truncated result set as a SUCCESSFUL result (#228 review; same defect class + // as #209/#224) — watchTermination still owns the single PIT close (#202) + Future.failed(ex) } } .watchTermination() { (_, done) => diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala index 40b88d2a..7468e8a9 100644 --- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala +++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientConversion.scala @@ -17,6 +17,7 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch.core.{MsearchResponse, SearchResponse} import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonpSerializable import co.elastic.clients.json.jackson.{JacksonJsonpGenerator, JacksonJsonpMapper} @@ -107,4 +108,39 @@ trait JavaClientConversion { _: JavaClientCompanion => } root } + + /** Response tree of a one-shot search (#228), single-parse. + * + * Hits-only responses — the common row-shaped case — re-parent the `_source` trees the transport + * already parsed (see [[hitsToResponseNode]]); aggregation-bearing responses serialize the whole + * envelope once at token level via [[convertToTree]], never through a String. + */ + protected def searchResponseToTree(response: SearchResponse[ObjectNode]): JsonNode = + if (response.aggregations() != null && !response.aggregations().isEmpty) + convertToTree(response) + else hitsToResponseNode(response.hits().hits()) + + /** Response tree of a multi search (#228), single-parse. + * + * The `responses` array is rebuilt item by item with the same policy as + * [[searchResponseToTree]]; a failed item keeps full fidelity so core still sees its `error` + * object. + */ + protected def msearchResponseToTree(response: MsearchResponse[ObjectNode]): JsonNode = { + val root = JacksonConfig.objectMapper.createObjectNode() + val responses = root.putArray("responses") + response.responses().forEach { item => + if (item.isResult) { + val result = item.result() + if (result.aggregations() != null && !result.aggregations().isEmpty) { + responses.add(convertToTree(result)) + } else { + responses.add(hitsToResponseNode(result.hits().hits())) + } + } else { + responses.add(convertToTree(item)) + } + } + root + } } diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala index 844866c3..5ed77c63 100644 --- a/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala +++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/java/JavaClientConversionSpec.scala @@ -18,6 +18,9 @@ package app.softnetwork.elastic.client.java import app.softnetwork.elastic.client.ElasticConfig import app.softnetwork.elastic.sql.serialization.JacksonConfig +import co.elastic.clients.elasticsearch._types.aggregations.Aggregate +import co.elastic.clients.elasticsearch.core.{MsearchResponse, SearchResponse} +import co.elastic.clients.elasticsearch.core.msearch.MultiSearchResponseItem import co.elastic.clients.elasticsearch.core.search.Hit import co.elastic.clients.json.JsonData import com.fasterxml.jackson.databind.JsonNode @@ -42,6 +45,10 @@ class JavaClientConversionSpec extends AnyWordSpec with Matchers { def envelope(hits: JList[Hit[ObjectNode]]): ObjectNode = hitsToResponseNode(hits) def tree(hit: Hit[ObjectNode]): JsonNode = convertToTree(hit) def json(hit: Hit[ObjectNode]): String = convertToJson(hit) + def searchTree(response: SearchResponse[ObjectNode]): JsonNode = searchResponseToTree(response) + def msearchTree(response: MsearchResponse[ObjectNode]): JsonNode = + msearchResponseToTree(response) + def searchJson(response: SearchResponse[ObjectNode]): String = convertToJson(response) } private val mapper = JacksonConfig.objectMapper @@ -145,4 +152,83 @@ class JavaClientConversionSpec extends AnyWordSpec with Matchers { Companion.tree(hit) shouldBe mapper.readTree(Companion.json(hit)) } } + + private def searchResponseOf( + hits: JList[Hit[ObjectNode]], + aggregations: Map[String, Aggregate] = Map.empty + ): SearchResponse[ObjectNode] = + SearchResponse.of[ObjectNode] { builder => + builder + .took(1) + .timedOut(false) + .shards(s => s.total(1).successful(1).failed(0)) + .hits(h => h.hits(hits)) + aggregations.foreach { case (name, aggregate) => builder.aggregations(name, aggregate) } + builder + } + + "searchResponseToTree" should { + + "re-parent hits through the minimal envelope when the response has no aggregations" in { + val src = sourceNode() + val tree = Companion.searchTree( + searchResponseOf(Collections.singletonList(hitOf(Some("1"), Some(src)))) + ) + val hitNode = tree.path("hits").path("hits").get(0) + hitNode.path("_id").asText() shouldBe "1" + hitNode.get("_source") should be theSameInstanceAs src + tree.has("took") shouldBe false + } + + "keep the exact convertToJson shape when aggregations are present" in { + val response = searchResponseOf( + Collections.singletonList(hitOf(Some("1"), Some(sourceNode()))), + aggregations = Map("avg_price" -> Aggregate.of(a => a.avg(v => v.value(2.5)))) + ) + // normalize both sides through the same parse: the token-level tree may carry + // LongNode/IntNode artifacts that are value-equal but not node-class-equal + mapper.readTree(Companion.searchTree(response).toString) shouldBe + mapper.readTree(Companion.searchJson(response)) + } + } + + "msearchResponseToTree" should { + + "rebuild the responses array with per-item envelopes and full-fidelity failures" in { + val src = sourceNode() + val response = MsearchResponse.of[ObjectNode] { builder => + builder + .took(1) + .responses( + MultiSearchResponseItem.of[ObjectNode](item => + item.result(r => + r.took(1) + .timedOut(false) + .shards(s => s.total(1).successful(1).failed(0)) + .hits(h => h.hits(Collections.singletonList(hitOf(Some("1"), Some(src))))) + .status(200) + ) + ), + MultiSearchResponseItem.of[ObjectNode](item => + item.failure(f => + f.error(e => e.`type`("search_phase_execution_exception").reason("boom")) + .status(500) + ) + ) + ) + } + + val tree = Companion.msearchTree(response) + val responses = tree.path("responses") + responses.isArray shouldBe true + responses.size() shouldBe 2 + + val first = responses.get(0) + first.path("hits").path("hits").get(0).get("_source") should be theSameInstanceAs src + + val second = responses.get(1) + second.path("error").path("reason").asText() shouldBe "boom" + second.path("status").asInt() shouldBe 500 + } + } } diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/MockElasticClientApi.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/MockElasticClientApi.scala index 60213b9d..036bc144 100644 --- a/testkit/src/main/scala/app/softnetwork/elastic/client/MockElasticClientApi.scala +++ b/testkit/src/main/scala/app/softnetwork/elastic/client/MockElasticClientApi.scala @@ -25,6 +25,7 @@ import app.softnetwork.elastic.client.scroll._ import app.softnetwork.elastic.sql.PainlessContextType import app.softnetwork.elastic.sql.query.{SQLAggregation, SingleSearch} import app.softnetwork.elastic.sql.schema.TableAlias +import com.fasterxml.jackson.databind.JsonNode import org.slf4j.{Logger, LoggerFactory} import scala.collection.immutable.ListMap @@ -308,36 +309,40 @@ trait MockElasticClientApi extends NopeClientApi { override private[client] def executeSingleSearch( elasticQuery: ElasticQuery - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = ElasticResult.success( Some( - allDocumentsAsHits( - elasticQuery.indices.headOption.getOrElse("default_index") + mapper.readTree( + allDocumentsAsHits( + elasticQuery.indices.headOption.getOrElse("default_index") + ) ) ) ) override private[client] def executeMultiSearch( elasticQueries: ElasticQueries - ): ElasticResult[Option[String]] = + ): ElasticResult[Option[JsonNode]] = ElasticResult.success( Some( - allDocumentsAsHits( - elasticQueries.queries.head.indices.headOption.getOrElse("default_index") + mapper.readTree( + allDocumentsAsHits( + elasticQueries.queries.head.indices.headOption.getOrElse("default_index") + ) ) ) ) override private[client] def executeSingleSearchAsync(elasticQuery: ElasticQuery)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = + ): Future[ElasticResult[Option[JsonNode]]] = Future { executeSingleSearch(elasticQuery) } override private[client] def executeMultiSearchAsync(elasticQueries: ElasticQueries)(implicit ec: ExecutionContext - ): Future[ElasticResult[Option[String]]] = + ): Future[ElasticResult[Option[JsonNode]]] = Future { executeMultiSearch(elasticQueries) }