From 95b222b1855c2db5197ba259a0a55c4e63327b8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 12 Aug 2026 14:44:59 +0200 Subject: [PATCH 1/2] perf(client): normalize streamed rows in a single pass (#229) ScrollApi.scroll re-normalized every row on the stream with O(cols^2) ListMap scans (fields.map(row.getOrElse) linear gets, a full filterNot second walk with a Set.contains per key, and a double ListMap rebuild) - measured at ~7% of total sidecar CPU on the arrow#160 benchmark. The same shape ran per row in normalizeRow on the window-enrichment stream, the window one-shot loop and the parse path's final rows.map. New ElasticConversion.rowNormalizer(requestedFields), built once per stream/loop and applied per row: - stream-constant work hoisted: field-order array, name -> position java.util.HashMap, EntityContext decision, duplicate-name detection (degenerate duplicate output names fall back to the legacy semantics via normalizeRowOrdered, with the name set still hoisted); - a single walk per row, splitting entries into a positional array and an ordered extras buffer, with one ListMap build; - zero-rebuild passthrough: a row already carrying the requested fields in order (extras may trail, already in final position) is returned as the same instance; an in-order strict prefix under EntityContext too. Output contract unchanged: requested fields first in SQL SELECT order, missing ones null-filled (or skipped under EntityContext), then the row's extra entries in their original order. New RowNormalizerSpec (15 tests) pins the contract, the passthrough identities, the duplicate fallback and .toList equality with the legacy normalizeRow across a shape battery in both contexts, including one normalizer instance reused across a heterogeneous row stream. Guard suites green on real ES 6.8 (rest+jest), 7.17, 8.18, 9.0; core 761; cross-compiled 2.12 + 2.13. Closes #229 Co-Authored-By: Claude Fable 5 --- .../elastic/client/ElasticConversion.scala | 115 +++++++++++-- .../elastic/client/ScrollApi.scala | 19 +-- .../elastic/client/SearchApi.scala | 5 +- .../elastic/client/RowNormalizerSpec.scala | 153 ++++++++++++++++++ 4 files changed, 269 insertions(+), 23 deletions(-) create mode 100644 core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala 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 43e3dae0..6342cceb 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala @@ -23,6 +23,7 @@ import org.json4s.{Extraction, Formats} import java.time.{Instant, LocalDate, LocalDateTime, LocalTime, ZoneId, ZonedDateTime} import java.time.format.DateTimeFormatter import scala.collection.immutable.ListMap +import scala.collection.mutable.ListBuffer import scala.util.Try import scala.jdk.CollectionConverters._ @@ -296,7 +297,8 @@ trait ElasticConversion { // Normalize all rows at the end, after all transformations (flattening, aggregation merging) // Filter out "*" from fields — it is an artifact of COUNT(*) and not a real column val effectiveFields = fields.filterNot(_ == "*") - rows.map(row => normalizeRow(row, effectiveFields)) + if (effectiveFields.isEmpty) rows + else rows.map(rowNormalizer(effectiveFields)) } def findKeyValue(path: String, map: Map[String, Any]): Option[Any] = { @@ -351,23 +353,116 @@ trait ElasticConversion { /** Normalize a row to ensure all requested fields are present in the original SQL SELECT order. * Fields missing from the row are added with null value. Extra fields (such as the internally * carried `_id`) are appended after the requested fields. + * + * Every `row.get` here is a linear scan of the `ListMap` — O(fields × row) per call. Fine for a + * single row; any loop or stream must hoist a [[rowNormalizer]] instead. */ protected def normalizeRow( row: ListMap[String, Any], requestedFields: Seq[String] )(implicit context: ConversionContext): ListMap[String, Any] = { if (requestedFields.isEmpty) row + else normalizeRowOrdered(row, requestedFields, requestedFields.toSet) + } + + /** Legacy normalization body, with the requested-name set supplied by the caller so loops can + * hoist it — shared by [[normalizeRow]] and [[rowNormalizer]] 's duplicate-name fallback. + */ + private def normalizeRowOrdered( + row: ListMap[String, Any], + requestedFields: Seq[String], + requestedSet: Set[String] + )(implicit context: ConversionContext): ListMap[String, Any] = { + // Build ordered entries for requested fields, with null for missing ones + val ordered = + context match { + case EntityContext => requestedFields.flatMap(f => row.get(f).map(v => f -> v)) + case _ => requestedFields.map(f => f -> row.getOrElse(f, null)) + } + // Append any extra fields from the row that aren't in the requested fields list + val extra = row.filterNot { case (k, _) => requestedSet.contains(k) } + ListMap(ordered: _*) ++ extra + } + + /** Build a single-pass normalizer over a fixed list of requested fields, for row loops and + * streams. Same output contract as [[normalizeRow]]: requested fields first, in SQL SELECT order + * — missing ones null-filled, or skipped under [[EntityContext]] — then the row's extra entries + * in their original order. + * + * All stream-constant work (the field order array, the name → position index, the context + * decision) happens once here; the returned function walks each row exactly once. A row that + * already carries the requested fields in order is returned as-is — possibly the SAME instance, + * never a copy — without any rebuild. Degenerate duplicate requested names fall back to the + * legacy per-row scan (with the name set still hoisted), trading speed for the exact + * [[normalizeRow]] semantics on that shape. + */ + protected def rowNormalizer( + requestedFields: Seq[String] + )(implicit context: ConversionContext): ListMap[String, Any] => ListMap[String, Any] = { + if (requestedFields.isEmpty) identity else { - // Build ordered entries for requested fields, with null for missing ones - val ordered = - context match { - case EntityContext => requestedFields.flatMap(f => row.get(f).map(v => f -> v)) - case _ => requestedFields.map(f => f -> row.getOrElse(f, null)) + val fieldArr: Array[String] = requestedFields.toArray + val len = fieldArr.length + val fieldIndex = new java.util.HashMap[String, Integer](len * 2) + var i = 0 + while (i < len) { + fieldIndex.putIfAbsent(fieldArr(i), i) + i += 1 + } + if (fieldIndex.size() != len) { + // Duplicate output names cannot hold distinct positions in a row map — keep the + // legacy per-row semantics for this degenerate shape, name set hoisted per stream + val requestedSet = requestedFields.toSet + row => normalizeRowOrdered(row, requestedFields, requestedSet) + } else { + val nullFillMissing = context match { + case EntityContext => false + case _ => true } - // Append any extra fields from the row that aren't in the requested fields list - val requestedSet = requestedFields.toSet - val extra = row.filterNot { case (k, _) => requestedSet.contains(k) } - ListMap(ordered: _*) ++ extra + row => { + val values = new Array[Any](len) + val seen = new Array[Boolean](len) + var extras: ListBuffer[(String, Any)] = null + var inOrder = true + var passthrough = false + var p = 0 + val it = row.iterator + while (!passthrough && it.hasNext) { + val entry = it.next() + if (inOrder && fieldArr(p) == entry._1) { + values(p) = entry._2 + seen(p) = true + p += 1 + // All requested fields matched in order: whatever the iterator still holds are + // extras already in their final position — the row IS its normalized form + if (p == len) passthrough = true + } else { + inOrder = false + val idx = fieldIndex.get(entry._1) + if (idx ne null) { + values(idx.intValue) = entry._2 + seen(idx.intValue) = true + } else { + if (extras eq null) extras = new ListBuffer[(String, Any)] + extras += entry + } + } + } + // An in-order strict prefix needs no rebuild either when missing fields are skipped + if (passthrough || (inOrder && !nullFillMissing)) row + else { + val builder = ListMap.newBuilder[String, Any] + var j = 0 + while (j < len) { + if (seen(j)) builder += fieldArr(j) -> values(j) + else if (nullFillMissing) builder += fieldArr(j) -> null + j += 1 + } + if (extras ne null) extras.foreach(builder += _) + builder.result() + } + } + } } } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala index 5f377a53..ce705c7e 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala @@ -436,18 +436,10 @@ trait ScrollApi extends ElasticClientHelpers { } // Normalize rows to ensure all requested fields are present (only if context is native) in the original SQL SELECT order - // and flatten inner hits into individual rows + // and flatten inner hits into individual rows. The normalizer is built ONCE per stream — + // each row is walked a single time, and already-shaped rows pass through untouched (#229) val normalized = if (fields.nonEmpty) { - val requestedSet = fields.toSet - source.map { row => - val ordered = - context match { - case EntityContext => fields.flatMap(f => row.get(f).map(v => f -> v)) - case _ => fields.map(f => f -> row.getOrElse(f, null)) - } - val extra = row.filterNot { case (k, _) => requestedSet.contains(k) } - ListMap(ordered: _*) ++ extra - } + source.map(rowNormalizer(fields)) } else { source } @@ -494,6 +486,9 @@ trait ScrollApi extends ElasticClientHelpers { // Determine if we should keep the document ID in the output val shouldKeepDocumentId = keepsDocumentId(outputFields) + // Built once per stream — never per row (#229) + val normalizeOutputRow = rowNormalizer(outputFields) + Source .futureSource( windowCacheFuture.map { @@ -515,7 +510,7 @@ trait ScrollApi extends ElasticClientHelpers { ) .map { case (doc, metrics) => val enrichedDoc = enrichDocumentWithWindowValues(doc, cache, request) - var normalizedDoc = normalizeRow(enrichedDoc, outputFields) + var normalizedDoc = normalizeOutputRow(enrichedDoc) if (!shouldKeepDocumentId) { normalizedDoc = normalizedDoc - ElasticConversion.DocumentIdField } 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 010ff048..c04d61d1 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala @@ -1628,13 +1628,16 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers { // Determine ONCE whether the document ID stays in the output — never per row val shouldKeepDocumentId = keepsDocumentId(outputFields) + // Built once for the whole result set — never per row (#229) + val normalizeOutputRow = rowNormalizer(outputFields) + // Enrich each row with window values, then normalize field order. The base rows carry // their `_id` (see singleSearchInternal with retainDocumentId = true) for the ordinal // lookup — strip it on the way out unless the document-id column is enabled or `_id` // is selected. Only window-enriched rows ever pay this per-row strip. val enrichedRows = baseRows.map { row => val enriched = enrichDocumentWithWindowValues(row, cache, request) - val normalized = normalizeRow(enriched, outputFields) + val normalized = normalizeOutputRow(enriched) if (shouldKeepDocumentId) normalized else normalized - ElasticConversion.DocumentIdField } diff --git a/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala new file mode 100644 index 00000000..c329afc3 --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala @@ -0,0 +1,153 @@ +package app.softnetwork.elastic.client + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import scala.collection.immutable.ListMap + +/** Pins the single-pass [[ElasticConversion.rowNormalizer]] (#229) against the legacy + * [[ElasticConversion.normalizeRow]] contract: requested fields first, in SQL SELECT order — + * missing ones null-filled (or skipped under [[EntityContext]]) — then the row's extra entries + * in their original order. Order is asserted on `.toList` (ListMap equality ignores order). + */ +class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion { + + implicit val context: ConversionContext = NativeContext + + private val fields = Seq("a", "b", "c") + + private def normalize( + row: ListMap[String, Any], + requestedFields: Seq[String] = fields + )(implicit ctx: ConversionContext): ListMap[String, Any] = + rowNormalizer(requestedFields)(ctx)(row) + + private def legacy( + row: ListMap[String, Any], + requestedFields: Seq[String] = fields + )(implicit ctx: ConversionContext): ListMap[String, Any] = + normalizeRow(row, requestedFields)(ctx) + + "rowNormalizer" should "return an already-shaped row as the same instance" in { + val row = ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3) + normalize(row) should be theSameInstanceAs row + } + + it should "return a row with in-order fields followed by extras as the same instance" in { + val row = ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3, "_id" -> "42", "extra" -> true) + normalize(row) should be theSameInstanceAs row + } + + it should "reorder fields to the SQL SELECT order" in { + val row = ListMap[String, Any]("c" -> 3, "a" -> 1, "b" -> 2) + normalize(row).toList shouldBe List("a" -> 1, "b" -> 2, "c" -> 3) + } + + it should "null-fill missing fields in native context" in { + val row = ListMap[String, Any]("b" -> 2) + normalize(row).toList shouldBe List("a" -> null, "b" -> 2, "c" -> null) + } + + it should "append extras after the requested fields, preserving their original order" in { + val row = ListMap[String, Any]("x" -> 0, "c" -> 3, "y" -> 9, "a" -> 1) + normalize(row).toList shouldBe List("a" -> 1, "b" -> null, "c" -> 3, "x" -> 0, "y" -> 9) + } + + it should "resume positional matching after an extra breaks a non-empty in-order prefix" in { + val row = ListMap[String, Any]("a" -> 1, "x" -> 0, "b" -> 2, "c" -> 3) + normalize(row).toList shouldBe List("a" -> 1, "b" -> 2, "c" -> 3, "x" -> 0) + } + + it should "keep a present-but-null value as present" in { + val row = ListMap[String, Any]("a" -> null, "b" -> 2, "c" -> 3) + normalize(row) should be theSameInstanceAs row + val reordered = ListMap[String, Any]("b" -> 2, "a" -> null, "c" -> 3) + normalize(reordered).toList shouldBe List("a" -> null, "b" -> 2, "c" -> 3) + } + + it should "normalize an empty row to all-null fields in native context" in { + normalize(ListMap.empty[String, Any]).toList shouldBe + List("a" -> null, "b" -> null, "c" -> null) + } + + it should "return the row unchanged when no fields are requested" in { + val row = ListMap[String, Any]("z" -> 26, "a" -> 1) + normalize(row, Seq.empty) should be theSameInstanceAs row + } + + it should "skip missing fields in entity context" in { + val row = ListMap[String, Any]("c" -> 3, "a" -> 1) + normalize(row)(EntityContext).toList shouldBe List("a" -> 1, "c" -> 3) + } + + it should "return an in-order strict prefix as the same instance in entity context" in { + val row = ListMap[String, Any]("a" -> 1, "b" -> 2) + normalize(row)(EntityContext) should be theSameInstanceAs row + } + + it should "keep a present-but-null value in entity context" in { + val row = ListMap[String, Any]("b" -> null, "a" -> 1) + normalize(row)(EntityContext).toList shouldBe List("a" -> 1, "b" -> null) + } + + it should "match the legacy normalizeRow output on every shape, in both contexts" in { + val rows = Seq( + ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3), + ListMap[String, Any]("c" -> 3, "b" -> 2, "a" -> 1), + ListMap[String, Any]("b" -> 2), + ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3, "_id" -> "42"), + ListMap[String, Any]("x" -> 0, "c" -> 3, "y" -> 9, "a" -> 1), + ListMap[String, Any]("a" -> 1, "x" -> 0, "b" -> 2, "c" -> 3), + ListMap[String, Any]("a" -> null, "c" -> 3), + ListMap[String, Any]("x" -> 0, "y" -> 9), + ListMap.empty[String, Any] + ) + for (row <- rows) { + normalize(row)(NativeContext).toList shouldBe legacy(row)(NativeContext).toList + normalize(row)(EntityContext).toList shouldBe legacy(row)(EntityContext).toList + normalize(row, Seq.empty)(NativeContext).toList shouldBe + legacy(row, Seq.empty)(NativeContext).toList + } + } + + it should "normalize a heterogeneous stream of rows through ONE normalizer instance" in { + // The production pattern: one closure built per stream, applied to every row — each + // invocation must be independent (no state may leak between rows) + val normalizer = rowNormalizer(fields)(NativeContext) + val shaped = ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3) + val stream = Seq( + shaped, + ListMap[String, Any]("c" -> 30, "a" -> 10), + ListMap[String, Any]("x" -> 0, "b" -> 200), + ListMap.empty[String, Any], + ListMap[String, Any]("a" -> 1000, "b" -> 2000, "c" -> 3000, "_id" -> "42"), + shaped + ) + val normalized = stream.map(normalizer) + normalized.map(_.toList) shouldBe Seq( + List("a" -> 1, "b" -> 2, "c" -> 3), + List("a" -> 10, "b" -> null, "c" -> 30), + List("a" -> null, "b" -> 200, "c" -> null, "x" -> 0), + List("a" -> null, "b" -> null, "c" -> null), + List("a" -> 1000, "b" -> 2000, "c" -> 3000, "_id" -> "42"), + List("a" -> 1, "b" -> 2, "c" -> 3) + ) + normalized.head should be theSameInstanceAs shaped + normalized.last should be theSameInstanceAs shaped + } + + it should "fall back to the legacy semantics when requested fields contain duplicates" in { + val duplicated = Seq("a", "b", "a") + val rows = Seq( + ListMap[String, Any]("a" -> 1, "b" -> 2), + ListMap[String, Any]("b" -> 2, "extra" -> true), + ListMap.empty[String, Any] + ) + for (row <- rows) { + normalize(row, duplicated)(NativeContext).toList shouldBe + legacy(row, duplicated)(NativeContext).toList + normalize(row, duplicated)(EntityContext).toList shouldBe + legacy(row, duplicated)(EntityContext).toList + } + } +} From 46ddd76f35018e69e2d54c286d3e8a1868dbed51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Wed, 12 Aug 2026 14:49:25 +0200 Subject: [PATCH 2/2] perf(tests): improve formatting consistency in RowNormalizerSpec --- .../elastic/client/RowNormalizerSpec.scala | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala index c329afc3..485614d7 100644 --- a/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala +++ b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala @@ -7,8 +7,8 @@ import scala.collection.immutable.ListMap /** Pins the single-pass [[ElasticConversion.rowNormalizer]] (#229) against the legacy * [[ElasticConversion.normalizeRow]] contract: requested fields first, in SQL SELECT order — - * missing ones null-filled (or skipped under [[EntityContext]]) — then the row's extra entries - * in their original order. Order is asserted on `.toList` (ListMap equality ignores order). + * missing ones null-filled (or skipped under [[EntityContext]]) — then the row's extra entries in + * their original order. Order is asserted on `.toList` (ListMap equality ignores order). */ class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion { @@ -92,14 +92,14 @@ class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion it should "match the legacy normalizeRow output on every shape, in both contexts" in { val rows = Seq( - ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3), - ListMap[String, Any]("c" -> 3, "b" -> 2, "a" -> 1), + ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3), + ListMap[String, Any]("c" -> 3, "b" -> 2, "a" -> 1), ListMap[String, Any]("b" -> 2), - ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3, "_id" -> "42"), - ListMap[String, Any]("x" -> 0, "c" -> 3, "y" -> 9, "a" -> 1), - ListMap[String, Any]("a" -> 1, "x" -> 0, "b" -> 2, "c" -> 3), + ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3, "_id" -> "42"), + ListMap[String, Any]("x" -> 0, "c" -> 3, "y" -> 9, "a" -> 1), + ListMap[String, Any]("a" -> 1, "x" -> 0, "b" -> 2, "c" -> 3), ListMap[String, Any]("a" -> null, "c" -> 3), - ListMap[String, Any]("x" -> 0, "y" -> 9), + ListMap[String, Any]("x" -> 0, "y" -> 9), ListMap.empty[String, Any] ) for (row <- rows) { @@ -118,19 +118,19 @@ class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion val stream = Seq( shaped, ListMap[String, Any]("c" -> 30, "a" -> 10), - ListMap[String, Any]("x" -> 0, "b" -> 200), + ListMap[String, Any]("x" -> 0, "b" -> 200), ListMap.empty[String, Any], ListMap[String, Any]("a" -> 1000, "b" -> 2000, "c" -> 3000, "_id" -> "42"), shaped ) val normalized = stream.map(normalizer) normalized.map(_.toList) shouldBe Seq( - List("a" -> 1, "b" -> 2, "c" -> 3), - List("a" -> 10, "b" -> null, "c" -> 30), - List("a" -> null, "b" -> 200, "c" -> null, "x" -> 0), + List("a" -> 1, "b" -> 2, "c" -> 3), + List("a" -> 10, "b" -> null, "c" -> 30), + List("a" -> null, "b" -> 200, "c" -> null, "x" -> 0), List("a" -> null, "b" -> null, "c" -> null), List("a" -> 1000, "b" -> 2000, "c" -> 3000, "_id" -> "42"), - List("a" -> 1, "b" -> 2, "c" -> 3) + List("a" -> 1, "b" -> 2, "c" -> 3) ) normalized.head should be theSameInstanceAs shaped normalized.last should be theSameInstanceAs shaped @@ -139,7 +139,7 @@ class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion it should "fall back to the legacy semantics when requested fields contain duplicates" in { val duplicated = Seq("a", "b", "a") val rows = Seq( - ListMap[String, Any]("a" -> 1, "b" -> 2), + ListMap[String, Any]("a" -> 1, "b" -> 2), ListMap[String, Any]("b" -> 2, "extra" -> true), ListMap.empty[String, Any] )