Skip to content

perf(client): parse each Elasticsearch page once on the scroll hits path (softclient4es-arrow#160) #227

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fupelaqu merged 3 commits into from
Aug 12, 2026
Merged

perf(client): parse each Elasticsearch page once on the scroll hits path (softclient4es-arrow#160) #227

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a57d49e
perf(client): parse each Elasticsearch page once on the scroll hits p…
fupelaqu Aug 12, 2026
bb3dde9
fix(tests): correct formatting in JavaClientConversionSpec documentation
fupelaqu Aug 12, 2026
848ff27
perf(client): parse every Elasticsearch response exactly once across …
fupelaqu Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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 ====================
Expand Down
28 changes: 27 additions & 1 deletion core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Expand Down
9 changes: 5 additions & 4 deletions core/src/main/scala/app/softnetwork/elastic/client/NopeClientApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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)
}

Expand Down
89 changes: 48 additions & 41 deletions core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -251,7 +251,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
)
val aggs = toClientAggregations(aggregations)
ElasticResult.fromTry(
parseResponse(
parseResponseTree(
response,
fieldAliases,
aggs,
Expand Down Expand Up @@ -361,7 +361,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
)
val aggs = toClientAggregations(aggregations)
ElasticResult.fromTry(
parseResponse(
parseResponseTree(
response,
fieldAliases,
aggs,
Expand Down Expand Up @@ -538,7 +538,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
)
val aggs = toClientAggregations(aggregations)
ElasticResult.fromTry(
parseResponse(
parseResponseTree(
response,
fieldAliases,
aggs,
Expand Down Expand Up @@ -657,7 +657,7 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
)
val aggs = toClientAggregations(aggregations)
ElasticResult.fromTry(
parseResponse(
parseResponseTree(
response,
fieldAliases,
aggs,
Expand Down Expand Up @@ -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
Expand All @@ -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 =
Expand Down Expand Up @@ -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}"
Expand All @@ -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 =
Expand All @@ -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
Expand All @@ -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]]
Expand All @@ -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"
)
Expand All @@ -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"
)
Expand All @@ -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
}
Expand Down
61 changes: 61 additions & 0 deletions core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading
Loading