perf(client): normalize streamed rows in a single pass (#229)

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()
+          }
+        }
+      }
     }
   }
 

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
                 }

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
     }

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
+    }
+  }
+}