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..485614d7 --- /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 + } + } +}