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 8cd2351f..47110a48 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
@@ -349,6 +349,11 @@ trait ElasticConversion {
     // major (a bare number on ES 6, `{value, relation}` on ES 7+).
     val rootDocCount: Option[Long] = docCountOf(json.path("hits"), "total")
 
+    // SCHEMA-12 — the rows below are `_source` documents (Case 1 and Case 4) exactly when the
+    // response carries hits; the aggregation arms (Cases 2 and 3) build bucket rows, which never
+    // hold a nested `_source` shape, so they keep the projector alone.
+    val fromHits = hitsNode.exists(_.nonEmpty)
+
     val rows = (hitsNode, aggsNode) match {
       case (Some(hits), None) if hits.nonEmpty =>
         // Case 1 : only hits
@@ -399,17 +404,154 @@ trait ElasticConversion {
         fields.indices.filterNot(i => fields(i) == "*").map(i => converters(i)).toArray
       else null
     if (effective.isEmpty) rows
-    else rows.map(rowProjector(effective.map(_._1), effective.map(_._2), convs))
+    else {
+      val project = rowProjector(effective.map(_._1), effective.map(_._2), convs)
+      // SCHEMA-12 — an un-aliased dotted column reads its value through the nested path before
+      // the projection places it. The SOURCE names are handed over before the "*" filter: a star
+      // keeps every parent (D5 c). Decided once per response; `project` itself when nothing
+      // qualifies, so every other statement runs exactly the projection above.
+      val shape =
+        if (fromHits) withNestedPaths(fields, fieldAliases, nestedHits)(project) else project
+      rows.map(shape)
+    }
   }
 
-  def findKeyValue(path: String, map: Map[String, Any]): Option[Any] = {
-    val keys = path.split("\\.")
-    keys.foldLeft(Option(map): Option[Any]) {
-      case (Some(m: Map[_, _]), key) =>
-        m.asInstanceOf[Map[String, Any]].get(key)
-      case _ => None
+  def findKeyValue(path: String, map: Map[String, Any]): Option[Any] =
+    findKeyValue(path.split("\\."), map)
+
+  /** [[findKeyValue]] over a path split once (SCHEMA-12: per stream, never per row). Same rule:
+    * each segment is looked up in the map the previous one reached; anything else than a map before
+    * the last segment, or an absent key, is `None`.
+    */
+  private[client] def findKeyValue(segments: Array[String], map: Map[String, Any]): Option[Any] =
+    if (map eq null) None
+    else {
+      var current: Any = map
+      var i = 0
+      var found = true
+      while (found && i < segments.length) {
+        current match {
+          case m: Map[_, _] =>
+            m.asInstanceOf[Map[String, Any]].get(segments(i)) match {
+              case Some(v) => current = v
+              case None    => found = false
+            }
+          case _ => found = false
+        }
+        i += 1
+      }
+      if (found) Some(current) else None
+    }
+
+  /** The value a row carries under a FLAT key, when it holds one (SCHEMA-12, A-3). A flat `null` is
+    * no value: Elasticsearch indexes none for it (a document `{"obj.a": null, "obj": {"a": 4}}`
+    * holds 4 in doc values, the `fields` API, SUM and COUNT), so the nested spelling is read
+    * instead. The one precedence rule of the row path and of the ranking-window sort key
+    * (`SearchApi.resolveSortKey`).
+    */
+  private[client] def flatValue(map: Map[String, Any], path: String): Option[Any] =
+    map.get(path) match {
+      case found @ Some(v) if v != null => found
+      case _                            => None
+    }
+
+  /** SCHEMA-12 — the un-aliased dotted columns of ONE projection, and the parents the reader may
+    * drop, decided once per stream (per response on the one-shot routes) from the names both row
+    * sites already hold (D2, D5):
+    *
+    *   - a column qualifies when its name contains `.`, no `*`, is not an alias (a value of
+    *     `fieldAliases`, which includes the computed `__cN`) and is not an UNNEST output (a value
+    *     of `nestedHits`): its output name IS its source path;
+    *   - its top-level parent `P` is droppable in NATIVE rows only when (a) the projection does not
+    *     request `P` as a source or an output name, (c) it has no `*` / `x.*`, (d) `P` is not an
+    *     inner-hits key, and (f) no aliased source lies under `P` and no segment of a qualifying
+    *     column under `P` is an alias key or value -- `jsonNodeToMap` renames an alias source at
+    *     every depth, so such a parent may be a value's only carrier (A-5). (b) and (e) are per row
+    *     (see [[ElasticConversion.NestedPathColumns]]);
+    *   - a qualifying path that is also an aliased source reads its alias's value (A-6).
+    *
+    * `None` when nothing qualifies, and under [[DocumentContext]]: an INSERT ... SELECT writes
+    * documents, not SQL rows, and keeps main's document shape (D13).
+    */
+  private[client] def nestedPathColumns(
+    sourceFields: Seq[String],
+    fieldAliases: ListMap[String, String],
+    nestedHits: Map[String, Seq[(String, String)]]
+  )(implicit context: ConversionContext): Option[ElasticConversion.NestedPathColumns] =
+    context match {
+      case DocumentContext => None
+      case _ =>
+        val aliasValues: Set[String] = fieldAliases.values.toSet
+        val unnestOutputs: Set[String] = nestedHits.valuesIterator.flatMap(_.map(_._2)).toSet
+        val paths: Seq[String] = sourceFields
+          .filter(f =>
+            f.indexOf('.') >= 0 && f.indexOf('*') < 0 && !aliasValues.contains(f) &&
+            !unnestOutputs.contains(f)
+          )
+          .distinct
+        if (paths.isEmpty) None
+        else {
+          val segments: Array[Array[String]] = paths.map(_.split("\\.")).toArray
+          val aliasOf: Array[String] = paths.map(p => fieldAliases.getOrElse(p, null)).toArray
+          val parentOf: Array[Int] = Array.fill(paths.size)(-1)
+          val parents: Array[String] = context match {
+            case EntityContext =>
+              // entity rows feed case-class extraction, which reads the document's shape
+              Array.empty[String]
+            case _ =>
+              val star = sourceFields.exists(f => f == "*" || f.endsWith(".*"))
+              val requested: Set[String] = sourceFields.toSet ++ fieldAliases.keySet
+              val aliasNames: Set[String] = aliasValues ++ fieldAliases.keySet
+              def keeps(parent: String): Boolean =
+                star || // (c)
+                requested.contains(parent) || // (a)
+                nestedHits.contains(parent) || // (d)
+                fieldAliases.keysIterator.exists(source =>
+                  source == parent || source.startsWith(parent + ".")
+                ) || // (f) an aliased source under the parent
+                segments.exists(s => s(0) == parent && s.exists(aliasNames.contains)) // (f)
+              val droppable = segments
+                .map(_(0))
+                .distinct
+                .filterNot(keeps)
+                // one bit per parent in the reader's per-row mask: a parent past the 64th is kept
+                .take(64)
+              var i = 0
+              while (i < segments.length) {
+                parentOf(i) = droppable.indexOf(segments(i)(0))
+                i += 1
+              }
+              droppable
+          }
+          Some(
+            new ElasticConversion.NestedPathColumns(
+              paths.toArray,
+              segments,
+              aliasOf,
+              parentOf,
+              parents
+            )
+          )
+        }
+    }
+
+  /** SCHEMA-12 — `project`, preceded by the reader of the projection's un-aliased dotted columns
+    * ([[nestedPathColumns]]). Returns the `project` INSTANCE itself when nothing qualifies, or
+    * under [[DocumentContext]], so a statement without such a column runs the projection it ran
+    * before, and no per-row work is added (D7). The only way core builds a row from hits against a
+    * projection: `jsonToRows`' tail (hit rows) and `ScrollApi.scroll`'s normalizer.
+    */
+  private[client] def withNestedPaths(
+    sourceFields: Seq[String],
+    fieldAliases: ListMap[String, String],
+    nestedHits: Map[String, Seq[(String, String)]]
+  )(
+    project: ListMap[String, Any] => ListMap[String, Any]
+  )(implicit context: ConversionContext): ListMap[String, Any] => ListMap[String, Any] =
+    nestedPathColumns(sourceFields, fieldAliases, nestedHits) match {
+      case Some(read) => read.andThen(project)
+      case None       => project
     }
-  }
 
   /** Parse simple hits (without aggregations)
     *
@@ -1536,4 +1678,102 @@ object ElasticConversion extends ElasticConversion {
     * `elastic.include-document-id` is enabled.
     */
   val DocumentIdField: String = "_id"
+
+  /** SCHEMA-12 — the reader of ONE projection's un-aliased dotted columns, built once per stream by
+    * `nestedPathColumns` and composed BEFORE the projector, so `SELECT obj.a FROM t` carries the
+    * value of `obj.a` under the output name `obj.a`, as `SELECT obj.a AS k` carries it under `k`.
+    *
+    * For each column `i`, in order:
+    *   - an alias twin (`aliasOf(i)` set, A-6) takes its alias's value; absent stays absent;
+    *   - a non-null FLAT key (a flat `_source` spelling, a `fields` / docvalue entry) wins and the
+    *     row is left as it is (A-3, `flatValue`): the walk did not read the parent;
+    *   - else the nested path is walked (`findKeyValue`) and its value placed under the output
+    *     name, replacing a flat `null`;
+    *   - on a miss, a second walk -- on the miss path only -- classifies the stop: an absent key in
+    *     objects holding no dotted key means the document has no such value (the parent was read);
+    *     a list, a scalar or an object holding a dotted key before the leaf may carry it (not
+    *     read).
+    *
+    * Then each droppable parent (`parents`, decided once per stream) every column under which was
+    * READ in this row is removed when its value is an object (D5 b, e). Walks run before drops, on
+    * the original parent. Per-row state is one `Long` mask -- bit `j` set = `parents(j)` not read
+    * -- never a per-row split or set.
+    */
+  private[client] final class NestedPathColumns(
+    val paths: Array[String],
+    val segments: Array[Array[String]],
+    val aliasOf: Array[String],
+    val parentOf: Array[Int],
+    val parents: Array[String]
+  ) extends (ListMap[String, Any] => ListMap[String, Any]) {
+
+    private[this] val len: Int = paths.length
+
+    override def apply(in: ListMap[String, Any]): ListMap[String, Any] = {
+      var row = in
+      var notRead = 0L
+      var i = 0
+      while (i < len) {
+        val alias = aliasOf(i)
+        if (alias ne null) {
+          row.get(alias) match {
+            case Some(v) => row = row + (paths(i) -> v)
+            case None    =>
+          }
+        } else if (flatValue(row, paths(i)).isDefined) {
+          if (parentOf(i) >= 0) notRead |= 1L << parentOf(i)
+        } else {
+          findKeyValue(segments(i), row) match {
+            case Some(v) => row = row + (paths(i) -> v)
+            case None =>
+              if (parentOf(i) >= 0 && !absentFrom(segments(i), row))
+                notRead |= 1L << parentOf(i)
+          }
+        }
+        i += 1
+      }
+      var j = 0
+      while (j < parents.length) {
+        if ((notRead & (1L << j)) == 0L) {
+          row.get(parents(j)) match {
+            case Some(_: Map[_, _]) => row = row - parents(j)
+            case _                  =>
+          }
+        }
+        j += 1
+      }
+      row
+    }
+
+    /** On a miss: `true` when the walk stopped on a key absent from objects holding no dotted key
+      * (nothing under the parent can be this column's value), `false` when it met anything that may
+      * carry it -- a list or a scalar before the leaf, or an object holding a dotted key (`obj:
+      * {"x.y": 10}` read as `obj.x.y`). A parent absent from the row has nothing to lose.
+      */
+    private[this] def absentFrom(path: Array[String], row: ListMap[String, Any]): Boolean =
+      row.get(path(0)) match {
+        case None => true
+        case Some(top) =>
+          var current: Any = top
+          var k = 1
+          // 0 = still walking, 1 = absent (read), 2 = may carry the value (not read)
+          var verdict = 0
+          while (verdict == 0 && k < path.length) {
+            current match {
+              case m: Map[_, _] =>
+                val map = m.asInstanceOf[Map[String, Any]]
+                if (map.keysIterator.exists(_.indexOf('.') >= 0)) verdict = 2
+                else
+                  map.get(path(k)) match {
+                    case Some(v) => current = v
+                    case None    => verdict = 1
+                  }
+              case _ => verdict = 2
+            }
+            k += 1
+          }
+          // a walk that reached the leaf found it: unreachable on the miss path, and harmless
+          verdict != 2
+      }
+  }
 }
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala
index c629bd28..88574105 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala
@@ -1633,7 +1633,8 @@ trait IndicesApi extends ElasticClientHelpers with SchemaCacheTtlApi {
 
         // INSERT … AS SELECT
         case Left(single: SingleSearch) =>
-          implicit val context: ConversionContext = NativeContext
+          // SCHEMA-12 (D13): the rows become DOCUMENTS, written in main's shape
+          implicit val context: ConversionContext = DocumentContext
           Right(
             scroll(single).map { case (row, _) =>
               // Even with `elastic.include-document-id` enabled, the id column must never be
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 cd1334e5..e3230204 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala
@@ -726,9 +726,11 @@ trait ScrollApi extends ElasticClientHelpers with SchemaCacheTtlApi {
 
     // 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. The normalizer is built ONCE per stream —
-    // each row is walked a single time, and already-shaped rows pass through untouched (#229)
+    // each row is walked a single time, and already-shaped rows pass through untouched (#229).
+    // SCHEMA-12: an un-aliased dotted column reads its value through the nested path first
+    // (`withNestedPaths`, decided once per stream; the normalizer itself when nothing qualifies).
     val normalized = if (fields.nonEmpty) {
-      source.map(rowNormalizer(fields))
+      source.map(withNestedPaths(fields, fieldAliases, nestedHits)(rowNormalizer(fields)))
     } else {
       source
     }
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 624c0dc9..8137e04a 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
@@ -1988,22 +1988,14 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
     */
   /** Resolve an OVER ORDER BY column value from a top_hits inner-source map. The inner-source map
     * keeps nested objects un-flattened, so a dotted column (e.g. `address.salary`) is not a
-    * top-level key. Fall back to walking the dotted path into nested maps. This is purely additive
-    * — a flat column hits the direct lookup and behaves exactly as before.
+    * top-level key. Fall back to walking the dotted path into nested maps.
+    *
+    * SCHEMA-12 (A-3): the row path's one precedence rule, so the key follows the value the row
+    * shows — a NON-NULL flat key wins (`flatValue`), and a flat `null` beside the nested value
+    * reads the nested one, as Elasticsearch does.
     */
   private def resolveSortKey(h: Map[String, Any], col: String): Any =
-    h.get(col) match {
-      case Some(v) => v
-      case None =>
-        col
-          .split('.')
-          .foldLeft(Option[Any](h)) {
-            case (Some(m: Map[_, _]), part) =>
-              m.asInstanceOf[Map[String, Any]].get(part)
-            case _ => None
-          }
-          .orNull
-    }
+    flatValue(h, col).orElse(findKeyValue(col, h)).orNull
 
   private def extractRankings(
     row: ListMap[String, Any],
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/package.scala b/core/src/main/scala/app/softnetwork/elastic/client/package.scala
index 6e178179..02e31e1b 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/package.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/package.scala
@@ -578,4 +578,12 @@ package object client extends SerializationApi {
   case object NativeContext extends ConversionContext
 
   case object EntityContext extends ConversionContext
+
+  /** SCHEMA-12 (D13) — the rows an INSERT ... SELECT writes as DOCUMENTS. Native everywhere (every
+    * `context match` reads it as `NativeContext`), but an un-aliased dotted column is not read
+    * through its nested path: a document is not a SQL row, and core's ingest scripts read a dotted
+    * column as `ctx.obj?.a`, so a document keeps the shape it was read in: a nested source stays
+    * nested (`{"obj": {"a": 1}}`; the row's null `obj.a` cell is not written).
+    */
+  private[client] case object DocumentContext extends ConversionContext
 }
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/NestedPathColumnsSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/NestedPathColumnsSpec.scala
new file mode 100644
index 00000000..9e253a43
--- /dev/null
+++ b/core/src/test/scala/app/softnetwork/elastic/client/NestedPathColumnsSpec.scala
@@ -0,0 +1,544 @@
+/*
+ * 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
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.stream.scaladsl.{Sink, Source}
+import app.softnetwork.elastic.client.result.{ElasticFailure, ElasticResult, ElasticSuccess}
+import app.softnetwork.elastic.client.scroll.ScrollConfig
+import app.softnetwork.elastic.sql.parser.Parser
+import app.softnetwork.elastic.sql.query.{
+  MultiSearch,
+  SQLAggregation,
+  SelectStatement,
+  SingleSearch
+}
+import com.fasterxml.jackson.databind.JsonNode
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.collection.immutable.ListMap
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+/** An un-aliased dotted column (`SELECT obj.a FROM t`) carries the document's value under its
+  * output name in every row core builds from hits, as its aliased form (`obj.a AS k`) does -- and
+  * its parent object stops leaking into the row once the reader has read it.
+  *
+  * No Docker: a [[NopeClientApi]] double answers every request with the canned page of the index it
+  * is asked for, parsed as a client parses a page (no projection), so each route -- the one-shot
+  * search (sync and async), the scroll / PIT / search_after stream, the one-shot `_msearch` and the
+  * per-leg `UNION ALL` -- builds its rows with core's own derivation of the projection. The pages
+  * are what Elasticsearch 8.18.3 answers for a `_source` restricted to the column (MEASURED, T0-2):
+  * a flat spelling stays flat, a flat `null` beside the nested value is returned beside it, an
+  * array of objects is returned whole.
+  */
+class NestedPathColumnsSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with ElasticConversion {
+
+  implicit val system: ActorSystem = ActorSystem("nested-path-columns-spec")
+
+  override def afterAll(): Unit = {
+    system.terminate()
+    super.afterAll()
+  }
+
+  private def hit(id: String, source: String): String = s"""{"_id":"$id","_source":$source}"""
+
+  private def page(hits: Seq[String]): String =
+    s"""{"took":1,"hits":{"total":{"value":${hits.size}},"hits":${hits.mkString("[", ",", "]")}}}"""
+
+  /** `t1`: what Elasticsearch returns for `_source: ["obj.a"]`, by document (D shapes). */
+  private val dotted: Seq[(String, String)] = Seq(
+    "d1"  -> """{"obj":{"a":1}}""", // D-n
+    "d2"  -> """{"obj.a":2}""", // D-f
+    "d3"  -> """{"obj.a":3,"obj":{"a":30}}""", // D-b
+    "d4"  -> """{"obj.a":null,"obj":{"a":4}}""", // D-z
+    "d4n" -> """{"obj":{"a":4}}""", // D-n4
+    "d5"  -> """{"obj":[{"a":5},{"a":50}]}""", // D-l
+    "d6"  -> """{}""", // D-s (the sibling is filtered out by the includes)
+    "d6w" -> """{"obj":{"b":6}}""", // D-s read with the whole `_source` (window base rows)
+    "d7"  -> """{}""" // D-m
+  )
+
+  /** `t3`: the same for `_source: ["obj.x.y"]`. */
+  private val deep: Seq[(String, String)] = Seq(
+    "d9"  -> """{"obj":{"x":{"y":9}}}""", // D-x
+    "d10" -> """{"obj":{"x.y":10}}""" // D-mx
+  )
+
+  /** `t2`: `_source: ["n"]`. */
+  private val plain: Seq[(String, String)] = Seq(
+    "n1" -> """{"n":-2}""",
+    "n2" -> """{"n":10}""",
+    "n3" -> """{}"""
+  )
+
+  private val pages: Map[String, String] = Map(
+    "t1" -> page(dotted.map((hit _).tupled)),
+    "t2" -> page(plain.map((hit _).tupled)),
+    "t3" -> page(deep.map((hit _).tupled))
+  )
+
+  /** Every route ends in this double: a canned page per index, parsed with no projection, exactly
+    * as the clients parse their pages (`parseResponseTree(json, fieldAliases, ListMap.empty)`).
+    */
+  private class PagedClient extends NopeClientApi {
+    override protected def logger: Logger = LoggerFactory.getLogger("NestedPathColumnsSpec")
+    override def version: ElasticResult[String] = ElasticSuccess("8.18.3")
+
+    /** The projection `search` / `scroll` hand to the row sites, as they derive it. */
+    def projection(
+      single: SingleSearch
+    ): (Seq[String], ListMap[String, String], Map[String, Seq[(String, String)]]) =
+      (extractOutputFieldNames(single), single.fieldAliases, single.nestedHitsMappings)
+
+    private def tree(q: ElasticQuery): JsonNode = mapper.readTree(pages(q.indices.head))
+
+    private def stream(
+      q: ElasticQuery,
+      aliases: ListMap[String, String],
+      ctx: ConversionContext
+    ): Source[ListMap[String, Any], NotUsed] =
+      Source(parseResponseTree(tree(q), aliases, ListMap.empty)(ctx).get.toList)
+
+    override private[client] def pitSearchAfter(
+      elasticQuery: ElasticQuery,
+      fieldAliases: ListMap[String, String],
+      config: ScrollConfig,
+      hasSorts: Boolean
+    )(implicit
+      system: ActorSystem,
+      context: ConversionContext
+    ): Source[ListMap[String, Any], NotUsed] = stream(elasticQuery, fieldAliases, context)
+
+    override private[client] def searchAfter(
+      elasticQuery: ElasticQuery,
+      fieldAliases: ListMap[String, String],
+      config: ScrollConfig,
+      hasSorts: Boolean
+    )(implicit
+      system: ActorSystem,
+      context: ConversionContext
+    ): Source[ListMap[String, Any], NotUsed] = stream(elasticQuery, fieldAliases, context)
+
+    override private[client] def scrollClassic(
+      elasticQuery: ElasticQuery,
+      fieldAliases: ListMap[String, String],
+      aggregations: ListMap[String, SQLAggregation],
+      config: ScrollConfig
+    )(implicit
+      system: ActorSystem,
+      context: ConversionContext
+    ): Source[ListMap[String, Any], NotUsed] = stream(elasticQuery, fieldAliases, context)
+
+    override private[client] def executeSingleSearch(
+      elasticQuery: ElasticQuery
+    ): ElasticResult[Option[JsonNode]] = ElasticSuccess(Some(tree(elasticQuery)))
+
+    override private[client] def executeSingleSearchAsync(elasticQuery: ElasticQuery)(implicit
+      ec: scala.concurrent.ExecutionContext
+    ): scala.concurrent.Future[ElasticResult[Option[JsonNode]]] =
+      scala.concurrent.Future.successful(ElasticSuccess(Some(tree(elasticQuery))))
+
+    override private[client] def executeMultiSearch(
+      elasticQueries: ElasticQueries
+    ): ElasticResult[Option[JsonNode]] =
+      ElasticSuccess(
+        Some(
+          mapper.readTree(
+            elasticQueries.queries
+              .map(q => pages(q.indices.head))
+              .mkString(
+                """{"responses":[""",
+                ",",
+                "]}"
+              )
+          )
+        )
+      )
+  }
+
+  private lazy val client = new PagedClient
+
+  /** The routes a row-shaped statement takes (spec §4 R): P1 sync / async (LIMIT 10 => one-shot
+    * search), P2 (no LIMIT => scroll / PIT; ORDER BY => search_after; the stream itself).
+    */
+  private val routes: Seq[String] = Seq("P1", "P1-async", "P2", "P2-ordered", "P2-stream")
+
+  private def rows(
+    sql: String,
+    route: String,
+    ctx: ConversionContext = NativeContext
+  ): Seq[ListMap[String, Any]] = {
+    implicit val ec: scala.concurrent.ExecutionContext = system.dispatcher
+    def ok(r: ElasticResult[ElasticResponse]): Seq[ListMap[String, Any]] = r match {
+      case ElasticSuccess(response) => response.results
+      case ElasticFailure(error)    => fail(s"[$route] $sql: ${error.message}")
+    }
+    route match {
+      case "P1" => ok(client.search(SelectStatement(s"$sql LIMIT 10"))(ctx))
+      case "P1-async" =>
+        ok(Await.result(client.searchAsync(SelectStatement(s"$sql LIMIT 10"))(ec, ctx), 30.seconds))
+      case "P2"         => ok(client.search(SelectStatement(sql))(ctx))
+      case "P2-ordered" => ok(client.search(SelectStatement(s"$sql ORDER BY id"))(ctx))
+      case "P2-stream" =>
+        Await.result(
+          client.scroll(SelectStatement(sql))(system, ctx).map(_._1).runWith(Sink.seq),
+          30.seconds
+        )
+    }
+  }
+
+  private def cls(v: Any): String = if (v == null) "null" else v.getClass.getName
+
+  // ---------------------------------------------------------------------------------------------
+
+  "An un-aliased dotted column" should "carry the value its aliased form carries, on every route" in {
+    // D-z is the one exception (A-3): a flat null holds no value in Elasticsearch, so the reader
+    // walks to 4, where the alias loop keeps the flat null
+    val expected = Seq[Any](1, 2, 3, 4, 4, null, null, null, null)
+    routes.foreach { route =>
+      val s1 = rows("SELECT obj.a FROM t1", route)
+      val c1 = rows("SELECT obj.a AS k FROM t1", route)
+      withClue(s"[$route] ") {
+        s1.map(_.get("obj.a").orNull) shouldBe expected
+        s1.zip(c1).zip(dotted.map(_._1)).foreach { case ((s, c), id) =>
+          withClue(s"$id: ") {
+            if (id == "d4") c("k") shouldBe (null: Any)
+            else {
+              s("obj.a") shouldBe c("k")
+              cls(s("obj.a")) shouldBe cls(c("k"))
+            }
+          }
+        }
+      }
+    }
+  }
+
+  it should "drop the parent the reader read, and keep every parent that may carry a value" in {
+    routes.foreach { route =>
+      val s1 = rows("SELECT obj.a FROM t1", route)
+      withClue(s"[$route] ") {
+        s1.map(_.toList) shouldBe Seq(
+          List("obj.a" -> 1),
+          List("obj.a" -> 2),
+          // a non-null flat spelling won: the walk did not read `obj`, which stays as on main
+          List("obj.a" -> 3, "obj" -> Map("a" -> 30)),
+          List("obj.a" -> 4),
+          List("obj.a" -> 4),
+          // an array of objects: the walk met a list, the parent is the value's only carrier
+          List("obj.a" -> null, "obj" -> List(Map("a" -> 5), Map("a" -> 50))),
+          List("obj.a" -> null),
+          // a sibling missing from the document loses nothing: the parent goes
+          List("obj.a" -> null),
+          List("obj.a" -> null)
+        )
+      }
+    }
+  }
+
+  it should "walk a deeper path, and keep the parent of a mixed spelling" in {
+    routes.foreach { route =>
+      withClue(s"[$route] ") {
+        rows("SELECT obj.x.y FROM t3", route).map(_.toList) shouldBe Seq(
+          List("obj.x.y" -> 9),
+          List("obj.x.y" -> null, "obj" -> Map("x.y" -> 10))
+        )
+      }
+    }
+  }
+
+  it should "place itself in SELECT order beside a plain column" in {
+    routes.foreach { route =>
+      withClue(s"[$route] ") {
+        rows("SELECT id, obj.a FROM t1", route).take(4).map(_.keys.toList) shouldBe Seq(
+          List("id", "obj.a"),
+          List("id", "obj.a"),
+          List("id", "obj.a", "obj"),
+          List("id", "obj.a")
+        )
+      }
+    }
+  }
+
+  it should "keep the parent requested, aliased under it, renamed by an alias or under a star" in {
+    val d1 = """{"obj":{"a":1,"b":6},"n":3,"a":7}"""
+    def one(sql: String): ListMap[String, Any] = {
+      val c = new PagedClient {
+        override private[client] def executeSingleSearch(
+          q: ElasticQuery
+        ): ElasticResult[Option[JsonNode]] =
+          ElasticSuccess(Some(mapper.readTree(page(Seq(hit("d1", d1))))))
+      }
+      c.search(SelectStatement(s"$sql LIMIT 10"))(NativeContext) match {
+        case ElasticSuccess(r) => r.results.head
+        case ElasticFailure(e) => fail(s"$sql: ${e.message}")
+      }
+    }
+    // S5: `obj` is requested
+    one("SELECT obj, obj.a FROM t1").toList shouldBe
+    List("obj" -> Map("a" -> 1, "b" -> 6), "obj.a" -> 1, "n" -> 3, "a" -> 7)
+    // S4: an aliased source lies under `obj` (A-5)
+    val s4 = one("SELECT obj.a, obj.b AS k FROM t1")
+    s4("obj.a") shouldBe 1
+    s4("k") shouldBe 6
+    s4 should contain key "obj"
+    // S6: the alias twin reads its alias (A-6), and the parent stays
+    val s6 = one("SELECT obj.a, obj.a AS k FROM t1")
+    s6("obj.a") shouldBe 1
+    s6("k") shouldBe 1
+    s6 should contain key "obj"
+    // U5: a segment of the column is an alias key -- the rename reaches `obj.a` (A-5)
+    one("SELECT a AS z, obj.a FROM t1") should contain key "obj"
+    // U3: a star keeps every parent (D5 c)
+    val u3 = one("SELECT *, obj.a FROM t1")
+    u3("obj.a") shouldBe 1
+    u3 should contain key "obj"
+    // the plain form drops it
+    one("SELECT obj.a FROM t1").toList shouldBe List("obj.a" -> 1, "n" -> 3, "a" -> 7)
+  }
+
+  it should "read the value in the entity context, and keep the document's shape there" in {
+    val entity = rows("SELECT obj.a FROM t1", "P1", EntityContext)
+    entity.map(_.get("obj.a").orNull) shouldBe Seq[Any](1, 2, 3, 4, 4, null, null, null, null)
+    // an absent value stays ABSENT under EntityContext (SubqueryResolver's contract)
+    entity(5) should not contain key("obj.a")
+    entity(6) should not contain key("obj.a")
+    entity.head should contain key "obj"
+    entity(3) should contain key "obj"
+  }
+
+  "A UNION ALL" should "carry an un-aliased dotted leg's values, one-shot and leg by leg" in {
+    Seq("" -> "P4 (per leg, scroll)", " LIMIT 10" -> "P3 (one-shot _msearch)").foreach {
+      case (limit, route) =>
+        withClue(s"[$route] ") {
+          val s8 = client.search(
+            SelectStatement(s"SELECT obj.a FROM t1$limit UNION ALL SELECT n FROM t2$limit")
+          )(NativeContext) match {
+            case ElasticSuccess(r) => r.results
+            case ElasticFailure(e) => fail(e.message)
+          }
+          s8.map(_.get("obj.a").orNull) shouldBe Seq[Any](1, 2, 3, 4, 4, null, null, null, null, -2,
+            10, null)
+          s8.take(2).map(_.keys.toList) shouldBe Seq(List("obj.a"), List("obj.a"))
+          val s9 = client.search(
+            SelectStatement(s"SELECT n FROM t2$limit UNION ALL SELECT obj.a FROM t1$limit")
+          )(NativeContext) match {
+            case ElasticSuccess(r) => r.results
+            case ElasticFailure(e) => fail(e.message)
+          }
+          s9.map(_.get("n").orNull) shouldBe Seq[Any](-2, 10, null, 1, 2, 3, 4, 4, null, null, null,
+            null)
+        }
+    }
+  }
+
+  it should "convert a dotted leg's value to the result's type in the one rebuild (S10)" in {
+    val legs = mapper.readTree(s"""{"responses":[${pages("t1")},${pages("t2")}]}""")
+    val toDouble: Any => Any = {
+      case null                => null
+      case n: java.lang.Number => n.doubleValue()
+      case other               => other
+    }
+    val parsed = parseResponseTree(
+      legs,
+      ListMap.empty,
+      ListMap.empty,
+      Seq("obj.a"),
+      legProjections = Seq(
+        LegProjection(ListMap.empty, Seq("obj.a"), Map.empty, ListMap.empty, Some(Array(toDouble))),
+        LegProjection(ListMap.empty, Seq("n"), Map.empty, ListMap.empty, Some(Array(toDouble)))
+      )
+    )(NativeContext).get
+    parsed.map(_.get("obj.a").orNull) shouldBe Seq[Any](1.0, 2.0, 3.0, 4.0, 4.0, null, null, null,
+      null, -2.0, 10.0, null)
+    parsed.head("obj.a") shouldBe a[java.lang.Double]
+  }
+
+  // ---- the plan, decided once per stream (D2, D5, D7) -----------------------------------------
+
+  private def legs(sql: String): Seq[SingleSearch] = Parser(sql) match {
+    case Right(single: SingleSearch) => Seq(single)
+    case Right(multi: MultiSearch)   => multi.requests
+    case other                       => fail(s"$sql: $other")
+  }
+
+  /** Per leg: (walked paths, droppable parents, alias twins), or None when nothing qualifies. */
+  private def plan(
+    sql: String,
+    ctx: ConversionContext = NativeContext
+  ): Seq[Option[(List[String], List[String], List[String])]] =
+    legs(sql).map { leg =>
+      val (fields, aliases, nested) = client.projection(leg)
+      nestedPathColumns(fields, aliases, nested)(ctx).map(p =>
+        (p.paths.toList, p.parents.toList, p.aliasOf.toList.filter(_ ne null))
+      )
+    }
+
+  private val t = "dotted_cols"
+
+  "The nested-path plan" should "select exactly the un-aliased dotted columns, and the parents it may drop" in {
+    def walks(paths: String*)(parents: String*) = Some((paths.toList, parents.toList, Nil))
+    val expected: Seq[(String, Seq[Option[(List[String], List[String], List[String])]])] = Seq(
+      s"SELECT obj.a FROM $t"     -> Seq(walks("obj.a")("obj")),
+      s"SELECT id, obj.a FROM $t" -> Seq(walks("obj.a")("obj")),
+      s"SELECT obj.x.y FROM $t"   -> Seq(walks("obj.x.y")("obj")),
+      // S4: an aliased source under `obj` keeps it (A-5)
+      s"SELECT obj.a, obj.b AS k FROM $t" -> Seq(walks("obj.a")()),
+      // S5: `obj` is requested
+      s"SELECT obj, obj.a FROM $t" -> Seq(walks("obj.a")()),
+      // S6: the alias twin reads `k` (A-6), and keeps `obj`
+      s"SELECT obj.a, obj.a AS k FROM $t" -> Seq(Some((List("obj.a"), Nil, List("k")))),
+      s"""SELECT "obj.a" FROM $t"""       -> Seq(walks("obj.a")("obj")),
+      s"SELECT obj.a FROM $t UNION ALL SELECT n FROM $t"   -> Seq(walks("obj.a")("obj"), None),
+      s"SELECT n FROM $t UNION ALL SELECT obj.a FROM $t"   -> Seq(None, walks("obj.a")("obj")),
+      s"SELECT obj.a FROM $t UNION ALL SELECT dbl FROM $t" -> Seq(walks("obj.a")("obj"), None),
+      // S11 / U4-U6: an output named `obj`, an alias key among the segments; `obj.a` an alias
+      s"SELECT n AS obj, obj.a FROM $t"         -> Seq(walks("obj.a")()),
+      s"SELECT a AS z, obj.a FROM $t"           -> Seq(walks("obj.a")()),
+      s"""SELECT n AS "obj.a", obj.b FROM $t""" -> Seq(walks("obj.b")("obj")),
+      // U1 / U2: the qualifier is stripped at parse time; U3: a star keeps every parent
+      s"SELECT t.obj.a FROM $t AS t" -> Seq(walks("obj.a")("obj")),
+      s"SELECT $t.obj.a FROM $t"     -> Seq(walks("obj.a")("obj")),
+      s"SELECT *, obj.a FROM $t"     -> Seq(walks("obj.a")()),
+      // C6 and C7b name `obj.a` (site 1 never reads a bucket row; C7b is an S row, D5 d)
+      s"SELECT obj.a, COUNT(*) AS c FROM $t GROUP BY obj.a" -> Seq(walks("obj.a")("obj")),
+      s"SELECT t.id, t.obj.a, c.name FROM $t t JOIN UNNEST(t.children) AS c" ->
+      Seq(walks("obj.a")("obj")),
+      s"SELECT id, obj.a, ROW_NUMBER() OVER (PARTITION BY obj.a ORDER BY id) AS rn FROM $t" ->
+      Seq(walks("obj.a")("obj"))
+    ) ++ controls.map(c => c -> legs(c).map(_ => None))
+    expected.foreach { case (sql, want) => withClue(s"[$sql] ") { plan(sql) shouldBe want } }
+  }
+
+  /** C1-C5, C7a, C7c, C8: no un-aliased dotted column. */
+  private lazy val controls: Seq[String] = Seq(
+    s"SELECT obj.a AS k FROM $t",
+    s"""SELECT obj.a AS "obj.a" FROM $t""",
+    s"SELECT n FROM $t",
+    s"SELECT n AS m FROM $t",
+    s"SELECT n + 1 FROM $t",
+    s"SELECT ABS(obj.a) FROM $t",
+    s"SELECT obj.a::BIGINT FROM $t",
+    s"SELECT obj.a + 1 FROM $t",
+    s"SELECT obj FROM $t",
+    s"SELECT t.id, c.name FROM $t t JOIN UNNEST(t.children) AS c",
+    s"SELECT t.id, c.addr.city FROM $t t JOIN UNNEST(t.children) AS c",
+    s"SELECT * FROM $t",
+    s"SELECT obj.a AS k FROM $t UNION ALL SELECT n FROM $t"
+  )
+
+  it should "drop no parent in the entity context, and read nothing under DocumentContext" in {
+    plan(s"SELECT obj.a FROM $t", EntityContext) shouldBe Seq(Some((List("obj.a"), Nil, Nil)))
+    plan(s"SELECT obj.a FROM $t", DocumentContext) shouldBe Seq(None)
+  }
+
+  it should "hand back the projector INSTANCE when nothing qualifies (D7)" in {
+    val project: ListMap[String, Any] => ListMap[String, Any] = row => row
+    (controls.flatMap(legs) ++ legs(s"SELECT obj.a FROM $t")).foreach { leg =>
+      val (fields, aliases, nested) = client.projection(leg)
+      withClue(s"[${leg.sql}] ") {
+        withNestedPaths(fields, aliases, nested)(project)(
+          DocumentContext
+        ) should be theSameInstanceAs project
+      }
+    }
+    controls.flatMap(legs).foreach { leg =>
+      val (fields, aliases, nested) = client.projection(leg)
+      withClue(s"[${leg.sql}] ") {
+        withNestedPaths(fields, aliases, nested)(project)(
+          NativeContext
+        ) should be theSameInstanceAs project
+        withNestedPaths(fields, aliases, nested)(project)(
+          EntityContext
+        ) should be theSameInstanceAs project
+      }
+    }
+  }
+
+  it should "keep a parent past the 64th, which the per-row mask cannot track" in {
+    val paths = (0 to 64).map(i => s"p$i.a")
+    val p = nestedPathColumns(paths, ListMap.empty, Map.empty)(NativeContext).get
+    p.parents.length shouldBe 64
+    p.parentOf(63) shouldBe 63
+    p.parentOf(64) shouldBe -1
+    val row =
+      ListMap[String, Any](paths.map(_.takeWhile(_ != '.')).map(k => k -> Map("a" -> 1)): _*)
+    val read = p(row)
+    read.keySet should not contain "p0"
+    read("p64") shouldBe Map("a" -> 1)
+    read("p64.a") shouldBe 1
+  }
+
+  "The reader" should "classify each miss: absent value, list, scalar, dotted key" in {
+    val p = nestedPathColumns(Seq("obj.a", "obj.x.y"), ListMap.empty, Map.empty)(NativeContext).get
+    def obj(v: Any) = ListMap[String, Any]("obj" -> v)
+    // both read: the parent goes
+    p(obj(Map("a" -> 1, "x" -> Map("y" -> 2)))).toList shouldBe List("obj.a" -> 1, "obj.x.y" -> 2)
+    // `obj.x.y` stops on an absent key in objects without a dotted key: still read
+    p(obj(Map("a" -> 1, "x" -> Map("z" -> 2)))).toList shouldBe List("obj.a" -> 1)
+    // a list before the leaf: not read, kept
+    p(obj(Map("a" -> 1, "x" -> List(Map("y" -> 2))))).keySet should contain("obj")
+    // a dotted key in an object on the way: not read, kept
+    p(obj(Map("a" -> 1, "x.y" -> 3))).keySet should contain("obj")
+    // a scalar parent: nothing is walked, nothing is dropped (D5 b)
+    p(obj(5)).toList shouldBe List("obj" -> 5)
+    // a non-null flat key wins and the row is left as it is
+    val flat = ListMap[String, Any]("obj.a" -> 7, "obj" -> Map("a" -> 1, "x" -> Map("y" -> 2)))
+    p(flat).toList shouldBe flat.toList :+ ("obj.x.y" -> 2)
+    // a flat null falls through to the walk, which replaces it
+    p(ListMap[String, Any]("obj.a" -> null, "obj" -> Map("a" -> 4)))("obj.a") shouldBe 4
+  }
+
+  it should "leave every aggregation row of a GROUP BY over the column as the projector builds it" in {
+    val grouped = """{
+                    |  "took": 1,
+                    |  "hits": { "total": { "value": 3 }, "hits": [] },
+                    |  "aggregations": {
+                    |    "obj.a": {
+                    |      "buckets": [
+                    |        { "key": 1, "doc_count": 2, "c": { "value": 2 } },
+                    |        { "key": 4, "doc_count": 1, "c": { "value": 1 } }
+                    |      ]
+                    |    }
+                    |  }
+                    |}""".stripMargin
+    parseResponseTree(mapper.readTree(grouped), ListMap.empty, ListMap.empty, Seq("obj.a", "c"))(
+      NativeContext
+    ).get.map(_.toList) shouldBe Seq(List("obj.a" -> 1, "c" -> 2), List("obj.a" -> 4, "c" -> 1))
+  }
+
+  it should "append the row's extras after the requested columns, the parent once dropped" in {
+    val whole = page(Seq(hit("d1", """{"id":"d1","n":3,"obj":{"a":1,"b":6}}""")))
+    parseResponseTree(mapper.readTree(whole), ListMap.empty, ListMap.empty, Seq("obj.a", "id"))(
+      NativeContext
+    ).get.map(_.toList) shouldBe Seq(List("obj.a" -> 1, "id" -> "d1", "n" -> 3))
+    // the same row under DocumentContext: main's document shape, the null cell included
+    parseResponseTree(mapper.readTree(whole), ListMap.empty, ListMap.empty, Seq("obj.a", "id"))(
+      DocumentContext
+    ).get.map(_.toList) shouldBe Seq(
+      List("obj.a" -> null, "id" -> "d1", "n" -> 3, "obj" -> Map("a" -> 1, "b" -> 6))
+    )
+  }
+}
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/perf/NestedPathCostProbe.scala b/core/src/test/scala/app/softnetwork/elastic/client/perf/NestedPathCostProbe.scala
new file mode 100644
index 00000000..c3063fc5
--- /dev/null
+++ b/core/src/test/scala/app/softnetwork/elastic/client/perf/NestedPathCostProbe.scala
@@ -0,0 +1,156 @@
+/*
+ * 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.perf
+
+import app.softnetwork.elastic.client.{
+  mapper,
+  ClientAggregation,
+  ConversionContext,
+  ElasticConversion,
+  NativeContext
+}
+import com.fasterxml.jackson.databind.JsonNode
+
+import scala.collection.immutable.ListMap
+
+/** SCHEMA-12 cost probe: what building rows from one 1,000-hit page costs through
+  * `parseResponseTree` -- the public row path, which the nested-path reader joins only for a
+  * projection with an un-aliased dotted column.
+  *
+  * Public API only, so the SAME file runs on main and on the branch: the comparison is main vs
+  * final, run as separate JVMs interleaved (control, branch, control, branch) by the caller.
+  *
+  * {{{
+  * sbt "core/Test/runMain app.softnetwork.elastic.client.perf.NestedPathCostProbe 31 1000"
+  * }}}
+  *
+  * Arguments, optional and positional: `blocks` (default 31) and `hits` per page (default 1,000).
+  * Each block parses the page [[RepsPerBlock]] times; block 1 is reported as COLD (no warm-up
+  * before it), the median of the remaining blocks as WARM.
+  *
+  * Shapes: s1 = five plain columns in `_source` order (`id, n, name, d, flag`); s2 = the aliased
+  * control `obj.a AS k, n`; s3 = a GROUP BY response of `hits` buckets read as `obj.a, c`; and, for
+  * information only, the changed shape `id, obj.a` (S2), which has no main equivalent.
+  *
+  * This probe NEVER asserts: it prints and exits (a timing assertion in a suite is a CI-flake
+  * liability). It is a `main`-bearing object, not a ScalaTest suite.
+  */
+object NestedPathCostProbe extends ElasticConversion {
+
+  val DefaultBlocks: Int = 31
+  val DefaultHits: Int = 1000
+  val RepsPerBlock: Int = 50
+
+  private implicit val ctx: ConversionContext = NativeContext
+
+  private final case class Shape(
+    label: String,
+    page: JsonNode,
+    fieldAliases: ListMap[String, String],
+    fields: Seq[String],
+    aggregations: ListMap[String, ClientAggregation] = ListMap.empty
+  )
+
+  private def hitsPage(n: Int)(source: Int => String): JsonNode =
+    mapper.readTree(
+      (0 until n)
+        .map(i => s"""{"_id":"d$i","_source":${source(i)}}""")
+        .mkString(s"""{"took":1,"hits":{"total":{"value":$n},"hits":[""", ",", "]}}")
+    )
+
+  private def bucketsPage(n: Int): JsonNode =
+    mapper.readTree(
+      (0 until n)
+        .map(i => s"""{"key":$i,"doc_count":${i % 7 + 1},"c":{"value":${i % 7 + 1}}}""")
+        .mkString(
+          s"""{"took":1,"hits":{"total":{"value":$n},"hits":[]},"aggregations":{"obj.a":{"buckets":[""",
+          ",",
+          "]}}}"
+        )
+    )
+
+  private def shapes(n: Int): Seq[Shape] = Seq(
+    Shape(
+      "s1 id, n, name, d, flag",
+      hitsPage(n)(i =>
+        s"""{"id":"d$i","n":$i,"name":"name-$i","d":"2024-01-${f"${i % 28 + 1}%02d"}",""" +
+        s""""flag":${i % 2 == 0}}"""
+      ),
+      ListMap.empty,
+      Seq("id", "n", "name", "d", "flag")
+    ),
+    Shape(
+      "s2 obj.a AS k, n",
+      hitsPage(n)(i => s"""{"obj":{"a":$i},"n":$i}"""),
+      ListMap("obj.a" -> "k"),
+      Seq("k", "n")
+    ),
+    Shape("s3 GROUP BY obj.a (buckets)", bucketsPage(n), ListMap.empty, Seq("obj.a", "c")),
+    Shape(
+      "S2 id, obj.a (info only)",
+      hitsPage(n)(i => s"""{"id":"d$i","obj":{"a":$i}}"""),
+      ListMap.empty,
+      Seq("id", "obj.a")
+    )
+  )
+
+  private def block(s: Shape): Long = {
+    var sink = 0
+    val start = System.nanoTime()
+    var r = 0
+    while (r < RepsPerBlock) {
+      sink += parseResponseTree(s.page, s.fieldAliases, s.aggregations, s.fields).get.size
+      r += 1
+    }
+    val elapsed = System.nanoTime() - start
+    if (sink == Int.MinValue) println("")
+    elapsed / RepsPerBlock
+  }
+
+  private def median(xs: Seq[Long]): Long = {
+    val sorted = xs.sorted
+    sorted(sorted.length / 2)
+  }
+
+  def main(args: Array[String]): Unit = {
+    val blocks = math.max(2, args.lift(0).map(_.toInt).getOrElse(DefaultBlocks))
+    val hits = args.lift(1).map(_.toInt).getOrElse(DefaultHits)
+    println(s"NestedPathCostProbe — $blocks blocks x $RepsPerBlock pages of $hits hits")
+    println(
+      s"JVM ${System.getProperty("java.version")} · Scala ${util.Properties.versionNumberString}"
+    )
+    val all = shapes(hits)
+    // block 1 of every shape is COLD: no warm-up before it
+    val timings: Map[String, Array[Long]] = all.map(s => s.label -> Array.ofDim[Long](blocks)).toMap
+    var b = 0
+    while (b < blocks) {
+      // alternate the order so drift cannot favour one shape
+      val ordered = if (b % 2 == 0) all else all.reverse
+      ordered.foreach(s => timings(s.label)(b) = block(s))
+      b += 1
+    }
+    println()
+    println(f"${"shape"}%-32s ${"cold µs"}%10s ${"warm median µs"}%15s ${"warm best µs"}%13s")
+    all.foreach { s =>
+      val t = timings(s.label)
+      val warm = t.drop(1).toSeq
+      println(
+        f"${s.label}%-32s ${t(0) / 1000.0}%10.1f ${median(warm) / 1000.0}%15.1f ${warm.min / 1000.0}%13.1f"
+      )
+    }
+  }
+}
diff --git a/documentation/sql/dql_statements.md b/documentation/sql/dql_statements.md
index c5fca3fb..3863a460 100644
--- a/documentation/sql/dql_statements.md
+++ b/documentation/sql/dql_statements.md
@@ -85,6 +85,12 @@ ORDER BY id ASC;
 - `profile` is a `STRUCT` column.
 - `profile.city` and `profile.followers` access nested fields.
 - Aliases (`AS full_name`, `AS city`) are returned as column names.
+- An un-aliased nested field is returned under its dotted name:
+  `SELECT id, profile.city FROM dql_users` returns the columns `id` and `profile.city`. The parent
+  object `profile` is normally left out of these rows. A row can still carry it, for example when
+  the `SELECT` list also names `profile`, contains `*`, or reads another field of `profile` under
+  an alias (`profile.followers AS followers`). Rows read into a typed entity (`searchAs`) keep the
+  parent object.
 
 ---
 
diff --git a/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientDottedColumnSpec.scala b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientDottedColumnSpec.scala
new file mode 100644
index 00000000..bfaf552d
--- /dev/null
+++ b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientDottedColumnSpec.scala
@@ -0,0 +1,19 @@
+/*
+ * 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
+
+class JestClientDottedColumnSpec extends DottedColumnSpec
diff --git a/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala
new file mode 100644
index 00000000..31b50b35
--- /dev/null
+++ b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala
@@ -0,0 +1,19 @@
+/*
+ * 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
+
+class RestHighLevelClientDottedColumnSpec extends DottedColumnSpec
diff --git a/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala
new file mode 100644
index 00000000..31b50b35
--- /dev/null
+++ b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientDottedColumnSpec.scala
@@ -0,0 +1,19 @@
+/*
+ * 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
+
+class RestHighLevelClientDottedColumnSpec extends DottedColumnSpec
diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/arrow/JavaClientDottedColumnSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/arrow/JavaClientDottedColumnSpec.scala
new file mode 100644
index 00000000..b0f14187
--- /dev/null
+++ b/es8/java/src/test/scala/app/softnetwork/elastic/arrow/JavaClientDottedColumnSpec.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.arrow
+
+import app.softnetwork.elastic.client.DottedColumnSpec
+import app.softnetwork.elastic.client.result.{
+  ElasticFailure,
+  ElasticSuccess,
+  QueryResult,
+  QueryRows,
+  QueryStream,
+  QueryStructured
+}
+
+import scala.collection.immutable.ListMap
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+/** The ES 8 runner of [[DottedColumnSpec]], in a DOWNSTREAM package on purpose: the consumer smokes
+  * below reach core's rows exactly as the consumers do -- `gateway.run` and the three result
+  * variants -- through public API only (no publish: the cross-repo runs follow the republish).
+  */
+class JavaClientDottedColumnSpec extends DottedColumnSpec {
+
+  /** Arrow's `ElasticSqlExecutor.extractRows`: every variant a gateway result takes. */
+  private def extractRows(result: QueryResult): Seq[ListMap[String, Any]] = result match {
+    case QueryRows(rows, _)           => rows
+    case QueryStructured(response, _) => response.results
+    case QueryStream(stream, _) =>
+      Await.result(stream.map(_._1).runWith(akka.stream.scaladsl.Sink.seq), 60.seconds)
+    case _ => Seq.empty
+  }
+
+  private def gatewayRows(sql: String): Seq[ListMap[String, Any]] =
+    Await.result(client.run(sql), 60.seconds) match {
+      case ElasticSuccess(result) => extractRows(result)
+      case ElasticFailure(error)  => fail(s"[$sql] ${error.message}")
+    }
+
+  "A consumer" should "read the SCHEMA-1 arrow legs by declared name (arrow legs)" in {
+    // un-LIMITed (QueryStream) and LIMITed (one-shot) forms of both legs, read by declared name
+    Seq("", " LIMIT 100").foreach { limit =>
+      withClue(s"[limit='$limit'] ") {
+        val dotted = gatewayRows(s"SELECT obj.a FROM $uniqueTable$limit").map(_.get("obj.a").orNull)
+        val plain = gatewayRows(s"SELECT n FROM $uniqueTable$limit").map(_.get("n").orNull)
+        dotted.size shouldBe uniqueDocs.size
+        dotted should not contain (null: Any)
+        (dotted ++ plain).map(String.valueOf).distinct.size shouldBe 8
+      }
+    }
+  }
+
+  it should "build JDBC metadata from the rows: the SELECT list, typed from a value (jdbc metadata)" in {
+    // `ElasticResultSetMetaData.fromRows`: the union of the rows' keys in order, each typed from the
+    // first non-null value found across the rows
+    Seq("", " LIMIT 100").foreach { limit =>
+      withClue(s"[limit='$limit'] ") {
+        val rows = gatewayRows(s"SELECT id, obj.a FROM $uniqueTable$limit")
+        val seen = scala.collection.mutable.LinkedHashSet.empty[String]
+        rows.foreach(row => row.keys.foreach(k => seen += k))
+        seen.toList shouldBe List("id", "obj.a")
+        rows.iterator.map(_.getOrElse("obj.a", null)).find(_ != null).orNull shouldBe a[
+          java.lang.Integer
+        ]
+      }
+    }
+  }
+
+  it should "preload a changelog with main's documents, which read back with their values (extensions preload)" in {
+    inserted
+    // the statement extensions generates (`INSERT INTO <changelog> AS SELECT <columns> FROM <t>;`),
+    // run through `client.run`: a nested source is written nested, as on main (no flat `obj.a`),
+    // and the scripted column reads it
+    val dn = org.json4s.jackson.JsonMethods.parse(storedSource(changelogTable, "dn"))
+    (dn \ "obj.a") shouldBe org.json4s.JNothing
+    (dn \ "obj" \ "a") shouldBe org.json4s.JInt(1)
+    (dn \ "twice") shouldBe org.json4s.JInt(2)
+    // a view read: QueryStream (no LIMIT) and the one-shot variants (LIMIT)
+    Seq("", " LIMIT 100").foreach { limit =>
+      withClue(s"[limit='$limit'] ") {
+        val back = gatewayRows(s"SELECT id, obj.a FROM $changelogTable$limit")
+          .map(r => String.valueOf(r("id")) -> r.get("obj.a").orNull)
+          .toMap
+        Seq("dn", "df", "db", "dz", "dn4").foreach { id =>
+          withClue(s"$id: ")(back(id) shouldBe valueOfA(id))
+        }
+      }
+    }
+  }
+}
diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientDottedColumnSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientDottedColumnSpec.scala
new file mode 100644
index 00000000..2cb7ff33
--- /dev/null
+++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientDottedColumnSpec.scala
@@ -0,0 +1,19 @@
+/*
+ * 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
+
+class JavaClientDottedColumnSpec extends DottedColumnSpec
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/DottedColumnSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/DottedColumnSpec.scala
new file mode 100644
index 00000000..474790d2
--- /dev/null
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/DottedColumnSpec.scala
@@ -0,0 +1,599 @@
+/*
+ * 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
+
+import akka.actor.ActorSystem
+import akka.stream.scaladsl.Sink
+import app.softnetwork.elastic.client.result.{
+  ElasticFailure,
+  ElasticSuccess,
+  QueryResult,
+  QueryRows,
+  QueryStream,
+  QueryStructured
+}
+import app.softnetwork.elastic.client.spi.ElasticClientFactory
+import app.softnetwork.elastic.scalatest.ElasticDockerTestKit
+import app.softnetwork.elastic.sql.query.SelectStatement
+import app.softnetwork.persistence.generateUUID
+import org.json4s.jackson.JsonMethods.{compact, render}
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+import org.slf4j.{Logger, LoggerFactory}
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Paths}
+import scala.collection.immutable.ListMap
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import scala.util.Try
+
+/** The object an entity extraction reads (`searchAsUnchecked`), for [[DottedColumnSpec]]. */
+case class DottedColumnObj(a: Option[Int])
+
+/** One row of `SELECT id, obj.a FROM dotted_cols` read as an entity, for [[DottedColumnSpec]]. */
+case class DottedColumnRow(id: String, obj: Option[DottedColumnObj])
+
+/** An un-aliased dotted column (`SELECT obj.a FROM t`) carries the document's value under its
+  * output name `obj.a`, on every route core builds rows on, as its aliased form (`obj.a AS k`)
+  * carries it under `k` -- where it used to keep the `_source` shape (`obj -> {a: 1}`) beside a
+  * null-filled `obj.a`.
+  *
+  * The documents are written as Elasticsearch stores them (a flat spelling stays flat), one per
+  * shape the reader must handle: nested (`dn`), flat (`df`), both spellings (`db`), a flat `null`
+  * beside the nested value (`dz`) and its nested twin (`dn4`), an array of objects (`dl`), a
+  * sibling only (`ds`), missing (`dm`), a deeper path (`dx`) and a mixed spelling (`dmx`). The
+  * index has two shards, so the scroll is sliced on Elasticsearch 8.
+  *
+  * PUBLIC API ONLY: the same file runs on a main clone, which is how the "census" (main's answers,
+  * the canary) and the "population" (every statement on every route, as canonical rows) are
+  * measured against the branch -- `DOTTED_CENSUS_OUT` / `DOTTED_POPULATION_OUT` name the files.
+  */
+trait DottedColumnSpec extends AnyFlatSpecLike with ElasticDockerTestKit with Matchers {
+
+  lazy val log: Logger = LoggerFactory.getLogger(getClass.getName)
+
+  implicit val system: ActorSystem = ActorSystem(generateUUID())
+
+  lazy val client: ElasticClientApi = ElasticClientFactory.create(elasticConfig)
+
+  protected val table = "dotted_cols"
+  protected val uniqueTable = "dotted_cols_u"
+  protected val copyTable = "dotted_cols_t2"
+  protected val scriptedTable = "dotted_cols_t2s"
+  protected val changelogTable = "dotted_cl"
+
+  /** The documents of [[table]], as written (`_id` = `id`). */
+  protected val docs: Seq[(String, String)] = Seq(
+    "dn" ->
+    """{"id":"dn","n":1,"dbl":1.5,"a":7,"obj":{"a":1},"children":[{"name":"c1","addr":{"city":"Paris"}},{"name":"c2"}]}""",
+    "df"  -> """{"id":"df","n":2,"dbl":2.5,"obj.a":2}""",
+    "db"  -> """{"id":"db","n":3,"obj.a":3,"obj":{"a":30}}""",
+    "dz"  -> """{"id":"dz","n":4,"obj.a":null,"obj":{"a":4}}""",
+    "dn4" -> """{"id":"dn4","n":5,"obj":{"a":4}}""",
+    "dl"  -> """{"id":"dl","n":6,"obj":[{"a":5},{"a":50}]}""",
+    "ds"  -> """{"id":"ds","n":7,"obj":{"b":6}}""",
+    "dm"  -> """{"id":"dm","n":8,"children":[{"name":"c3"}]}""",
+    "dx"  -> """{"id":"dx","obj":{"x":{"y":9}}}""",
+    "dmx" -> """{"id":"dmx","obj":{"x.y":10}}"""
+  )
+
+  /** What `obj.a` holds per document, as Elasticsearch reads it (D-z holds 4: a flat null is no
+    * value; D-l is multi-valued, which stays NULL until SCHEMA-6; D-mx is `obj.x.y`).
+    */
+  protected val valueOfA: Map[String, Any] = Map(
+    "dn"  -> 1,
+    "df"  -> 2,
+    "db"  -> 3,
+    "dz"  -> 4,
+    "dn4" -> 4,
+    "dl"  -> null,
+    "ds"  -> null,
+    "dm"  -> null,
+    "dx"  -> null,
+    "dmx" -> null
+  )
+
+  /** [[uniqueTable]]: every document carries one `obj.a` (nested or flat, no array, no flat null).
+    */
+  protected val uniqueDocs: Seq[(String, String)] = Seq(
+    "u1" -> """{"id":"u1","n":-2,"obj":{"a":1}}""",
+    "u2" -> """{"id":"u2","n":10,"obj.a":2}""",
+    "u3" -> """{"id":"u3","n":3,"obj":{"a":4}}""",
+    "u4" -> """{"id":"u4","n":5,"obj":{"a":5}}""",
+    "u5" -> """{"id":"u5","obj.a":1}"""
+  )
+
+  private val mapping: String =
+    """{"properties":{
+      |  "id":{"type":"keyword"}, "n":{"type":"integer"}, "dbl":{"type":"double"},
+      |  "a":{"type":"integer"},
+      |  "obj":{"properties":{"a":{"type":"integer"},"b":{"type":"integer"},
+      |    "x":{"properties":{"y":{"type":"integer"}}}}},
+      |  "children":{"type":"nested","properties":{"name":{"type":"keyword"},
+      |    "addr":{"properties":{"city":{"type":"keyword"}}}}}
+      |}}""".stripMargin
+
+  private def drop(index: String): Unit = {
+    Try(Await.result(client.run(s"DROP TABLE IF EXISTS $index"), 60.seconds))
+    Try(client.deleteIndex(index))
+  }
+
+  private def load(index: String, shards: Int, documents: Seq[(String, String)]): Unit = {
+    client
+      .createIndex(index, settings = s"""{"number_of_shards": $shards, "number_of_replicas": 0}""")
+      .get shouldBe true
+    client.setMapping(index, mapping).get shouldBe true
+    documents.foreach { case (id, source) =>
+      client.index(index, id, source) match {
+        case ElasticSuccess(true) =>
+        case other                => fail(s"indexing $id into $index: $other")
+      }
+    }
+    client.refresh(index)
+  }
+
+  protected def runOk(sql: String): QueryResult =
+    Await.result(client.run(sql), 60.seconds) match {
+      case ElasticSuccess(result) => result
+      case ElasticFailure(error)  => fail(s"[$sql] failed: ${error.message}")
+    }
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    Seq(table, uniqueTable, copyTable, scriptedTable, changelogTable).foreach(drop)
+    load(table, 2, docs)
+    load(uniqueTable, 1, uniqueDocs)
+    runOk(s"CREATE TABLE $copyTable (id KEYWORD, obj STRUCT FIELDS(a INT), PRIMARY KEY (id))")
+    Seq(scriptedTable, changelogTable).foreach { t =>
+      runOk(
+        s"CREATE TABLE $t (id KEYWORD, obj STRUCT FIELDS(a INT), " +
+        "twice INT SCRIPT AS (obj.a * 2), PRIMARY KEY (id))"
+      )
+    }
+    ()
+  }
+
+  override def afterAll(): Unit = {
+    Seq(table, uniqueTable, copyTable, scriptedTable, changelogTable).foreach(drop)
+    super.afterAll()
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // Routes (spec section 4 R) and canonical rows
+  // -------------------------------------------------------------------------------------------
+
+  /** Every variant a gateway result can take, as a consumer reads it (arrow's `extractRows`). */
+  protected def rowsOfResult(result: QueryResult): Seq[ListMap[String, Any]] = result match {
+    case QueryRows(rows, _)           => rows
+    case QueryStructured(response, _) => response.results
+    case QueryStream(stream, _) =>
+      Await.result(stream.map(_._1).runWith(Sink.seq), 60.seconds)
+    case other => fail(s"unexpected result $other")
+  }
+
+  /** A row-shaped statement's routes: P1 sync (`search`) and async (`run` -> `searchAsync`, LIMIT
+    * => one-shot), P2 sliced (`run`, no LIMIT, 2 shards), P2 sequential (one slice) and P2 sorted
+    * (`ORDER BY id` => search_after).
+    */
+  protected val rowRoutes: Seq[String] = Seq("P1", "P1-async", "P2", "P2-seq", "P2-sorted")
+
+  /** A `UNION ALL`'s: P3 (every leg LIMITed => one `_msearch`), P4 (an un-LIMITed leg => leg by
+    * leg), through the gateway and through `search`.
+    */
+  protected val unionRoutes: Seq[String] = Seq("P3", "P3-search", "P4", "P4-search")
+
+  private def limited(sql: String): String =
+    if (sql.contains(" UNION ALL "))
+      sql.replace(" UNION ALL ", " LIMIT 100 UNION ALL ") + " LIMIT 100"
+    else s"$sql LIMIT 100"
+
+  protected def rowsOn(sql: String, route: String): Either[String, Seq[ListMap[String, Any]]] =
+    Try {
+      def search(statement: String): Seq[ListMap[String, Any]] =
+        client.search(SelectStatement(statement))(NativeContext) match {
+          case ElasticSuccess(response) => response.results
+          case ElasticFailure(error)    => throw new IllegalStateException(error.message)
+        }
+      def gateway(statement: String): Seq[ListMap[String, Any]] =
+        Await.result(client.run(statement), 60.seconds) match {
+          case ElasticSuccess(result) => rowsOfResult(result)
+          case ElasticFailure(error)  => throw new IllegalStateException(error.message)
+        }
+      route match {
+        case "P1"        => search(limited(sql))
+        case "P1-async"  => gateway(limited(sql))
+        case "P2"        => gateway(sql)
+        case "P2-sorted" => gateway(s"$sql ORDER BY id")
+        case "P2-seq" =>
+          Await.result(
+            client
+              .scroll(SelectStatement(sql), client.defaultScrollConfig.copy(maxSlices = Some(1)))(
+                system,
+                NativeContext
+              )
+              .map(_._1)
+              .runWith(Sink.seq),
+            60.seconds
+          )
+        case "P3"        => gateway(limited(sql))
+        case "P3-search" => search(limited(sql))
+        case "P4"        => gateway(sql)
+        case "P4-search" => search(sql)
+      }
+    }.toEither.left.map(t => s"${t.getClass.getSimpleName}: ${t.getMessage}")
+
+  protected def rowsOk(sql: String, route: String): Seq[ListMap[String, Any]] =
+    rowsOn(sql, route) match {
+      case Right(rows) => rows
+      case Left(error) => fail(s"[$route] $sql: $error")
+    }
+
+  /** A value with its class, maps and lists in their own order. */
+  protected def shown(v: Any): String = v match {
+    case null => "null"
+    case m: scala.collection.Map[_, _] =>
+      m.map { case (k, x) => s"$k=${shown(x)}" }.mkString("{", ",", "}")
+    case s: Seq[_] => s.map(shown).mkString("[", ",", "]")
+    case other     => s"$other:${other.getClass.getSimpleName}"
+  }
+
+  protected def canonical(row: scala.collection.Map[String, Any]): String =
+    row.map { case (k, v) => s"$k=${shown(v)}" }.mkString("{", ", ", "}")
+
+  private def tsv(cells: String*): String =
+    cells.map(_.replace('\t', ' ').replace('\n', ' ')).mkString("\t")
+
+  /** One line per row (sorted where the route fixes no order), or the error. */
+  protected def dumpLines(label: String, sql: String, route: String): Seq[String] =
+    rowsOn(sql, route) match {
+      case Left(error) => Seq(tsv(route, label, sql, "-", s"ERROR $error"))
+      case Right(rows) =>
+        val lines = rows.map(canonical)
+        // only `P2-sorted` fixes the order of the rows; elsewhere they are compared as a multiset
+        val ordered = if (route == "P2-sorted") lines else lines.sorted
+        if (ordered.isEmpty) Seq(tsv(route, label, sql, "-", "NO ROWS"))
+        else ordered.zipWithIndex.map { case (l, i) => tsv(route, label, sql, i.toString, l) }
+    }
+
+  protected def write(env: String, default: String, lines: Seq[String]): Unit = {
+    val out = Paths.get(sys.env.getOrElse(env, default))
+    Option(out.toAbsolutePath.getParent).foreach(p => Files.createDirectories(p))
+    Files.write(out, lines.mkString("", "\n", "\n").getBytes(StandardCharsets.UTF_8))
+    log.info(s"wrote ${lines.size} lines to ${out.toAbsolutePath}")
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // The statements (spec section 4 S and C)
+  // -------------------------------------------------------------------------------------------
+
+  protected def rowStatements: Seq[(String, String)] = Seq(
+    "S1"  -> s"SELECT obj.a FROM $table",
+    "S2"  -> s"SELECT id, obj.a FROM $table",
+    "S3"  -> s"SELECT id, obj.x.y FROM $table",
+    "S4"  -> s"SELECT id, obj.a, obj.b AS k FROM $table",
+    "S5"  -> s"SELECT id, obj, obj.a FROM $table",
+    "S6"  -> s"SELECT id, obj.a, obj.a AS k FROM $table",
+    "S7"  -> s"""SELECT id, "obj.a" FROM $table""",
+    "U1"  -> s"SELECT t.id, t.obj.a FROM $table AS t",
+    "U3"  -> s"SELECT *, obj.a FROM $table",
+    "U4"  -> s"SELECT id, n AS obj, obj.a FROM $table",
+    "U5"  -> s"SELECT id, a AS z, obj.a FROM $table",
+    "U6"  -> s"""SELECT id, n AS "obj.a", obj.b FROM $table""",
+    "C1"  -> s"SELECT id, obj.a AS k FROM $table",
+    "C1x" -> s"SELECT id, obj.x.y AS k FROM $table",
+    "C2"  -> s"""SELECT id, obj.a AS "obj.a" FROM $table""",
+    "C3a" -> s"SELECT id, n FROM $table",
+    "C3b" -> s"SELECT id, n AS m FROM $table",
+    "C4a" -> s"SELECT id, n + 1 FROM $table",
+    "C4b" -> s"SELECT id, ABS(obj.a) FROM $table",
+    "C4c" -> s"SELECT id, obj.a::BIGINT FROM $table",
+    "C4d" -> s"SELECT id, obj.a + 1 FROM $table",
+    "C5"  -> s"SELECT id, obj FROM $table",
+    "C8"  -> s"SELECT * FROM $table"
+  )
+
+  /** UNNEST (C7) and GROUP BY (C6) take no `ORDER BY id`. */
+  protected def unorderedStatements: Seq[(String, String)] = Seq(
+    "C6"  -> s"SELECT obj.a, COUNT(*) AS c FROM $table GROUP BY obj.a",
+    "C7a" -> s"SELECT t.id, c.name FROM $table t JOIN UNNEST(t.children) AS c",
+    "C7b" -> s"SELECT t.id, t.obj.a, c.name FROM $table t JOIN UNNEST(t.children) AS c",
+    "C7c" -> s"SELECT t.id, c.addr.city FROM $table t JOIN UNNEST(t.children) AS c",
+    "AGG" -> s"SELECT id, ARRAY_AGG(obj.a) AS l FROM $table GROUP BY id"
+  )
+
+  protected def windowStatements: Seq[(String, String)] = Seq(
+    "W1" -> s"SELECT id, obj.a, ROW_NUMBER() OVER (ORDER BY id) AS rn FROM $table",
+    "W2" -> s"SELECT id, obj.a, ROW_NUMBER() OVER (PARTITION BY obj.a ORDER BY id) AS rn FROM $table",
+    "W3" -> s"SELECT id, RANK() OVER (ORDER BY obj.a) AS r FROM $table"
+  )
+
+  protected def unionStatements: Seq[(String, String)] = Seq(
+    "S8"  -> s"SELECT obj.a FROM $table UNION ALL SELECT n FROM $table",
+    "S9"  -> s"SELECT n FROM $table UNION ALL SELECT obj.a FROM $table",
+    "S10" -> s"SELECT obj.a FROM $table UNION ALL SELECT dbl FROM $table",
+    "C1u" -> s"SELECT obj.a AS k FROM $table UNION ALL SELECT n AS k FROM $table"
+  )
+
+  protected def entityStatements: Seq[(String, String)] = Seq(
+    // labelled by the subquery's shape: the S1 forms are the ones AC7 changes
+    "S1-in-W" -> s"SELECT id FROM $table WHERE n IN (SELECT obj.a FROM $uniqueTable LIMIT 100)",
+    "S1-notin-W" -> s"SELECT id FROM $table WHERE n NOT IN (SELECT obj.a FROM $uniqueTable LIMIT 100)",
+    "C1-in-W" -> s"SELECT id FROM $table WHERE n IN (SELECT obj.a AS k FROM $uniqueTable LIMIT 100)",
+    "C1-notin-W" -> s"SELECT id FROM $table WHERE n NOT IN (SELECT obj.a AS k FROM $uniqueTable LIMIT 100)",
+    "C-in-P" -> s"SELECT id FROM $table WHERE n IN (SELECT obj.a FROM $uniqueTable)"
+  )
+
+  // -------------------------------------------------------------------------------------------
+  // INSERT ... SELECT (spec I) and what it wrote
+  // -------------------------------------------------------------------------------------------
+
+  /** Runs the INSERT ... SELECT statements once per suite: the plain copy, the scripted target and
+    * the statement extensions generates to preload a materialized-view changelog.
+    */
+  protected lazy val inserted: Unit = {
+    runOk(s"INSERT INTO $copyTable SELECT id, obj.a FROM $table")
+    runOk(s"INSERT INTO $scriptedTable SELECT id, obj.a FROM $table")
+    runOk(s"INSERT INTO $changelogTable AS SELECT id, obj.a FROM $table;")
+    Seq(copyTable, scriptedTable, changelogTable).foreach(client.refresh)
+  }
+
+  /** The `_source` of each target document, as stored. */
+  protected def storedSource(index: String, id: String): String =
+    getDocument(index, id).map(d => compact(render(d \ "_source"))).getOrElse("ABSENT")
+
+  protected def insertLines: Seq[String] = {
+    inserted
+    Seq(copyTable, scriptedTable, changelogTable).flatMap { target =>
+      docs.map(_._1).map(id => tsv("I", target, id, "-", storedSource(target, id))) ++
+      dumpLines(
+        s"I-agg-$target",
+        s"SELECT SUM(obj.a) AS s, COUNT(obj.a) AS c FROM $target",
+        "P1-async"
+      ) ++ dumpLines(s"S2-read-$target", s"SELECT id, obj.a FROM $target", "P2-sorted")
+    }
+  }
+
+  protected def entityRows: Either[String, Seq[DottedColumnRow]] =
+    client.searchAsUnchecked[DottedColumnRow](
+      SelectStatement(s"SELECT id, obj.a FROM $table WHERE n <= 5 ORDER BY id")
+    ) match {
+      case ElasticSuccess(rows) => Right(rows)
+      case ElasticFailure(e)    => Left(e.message)
+    }
+
+  // -------------------------------------------------------------------------------------------
+  // The census (T0-3) and the population (AC3): dumps, main vs branch
+  // -------------------------------------------------------------------------------------------
+
+  "An un-aliased dotted column" should "dump the census of the row paths (T0-3)" in {
+    val lines =
+      rowStatements
+        .filter(s => s._1.startsWith("S") || s._1.startsWith("U") || s._1 == "C1")
+        .flatMap { case (l, sql) => Seq("P1", "P2").flatMap(r => dumpLines(l, sql, r)) } ++
+      unorderedStatements.flatMap { case (l, sql) => dumpLines(l, sql, "P1-async") } ++
+      windowStatements.flatMap { case (l, sql) => dumpLines(l, sql, "P2") } ++
+      unionStatements.flatMap { case (l, sql) =>
+        Seq("P3", "P4").flatMap(r => dumpLines(l, sql, r))
+      } ++
+      entityStatements.flatMap { case (l, sql) => dumpLines(l, sql, "P1-async") } ++
+      insertLines
+    write("DOTTED_CENSUS_OUT", "target/dotted-census.tsv", lines)
+    // the canary, read on main: S rows null-filled beside the leaked parent
+    val canary = lines.count(l => l.contains("obj.a=null") && l.contains("obj={"))
+    log.info(s"census canary: $canary rows carry obj.a=null beside an obj object")
+    lines should not be empty
+  }
+
+  it should "dump the population of every statement on every route (AC3)" in {
+    val lines =
+      rowStatements.flatMap { case (l, sql) => rowRoutes.flatMap(r => dumpLines(l, sql, r)) } ++
+      unorderedStatements.flatMap { case (l, sql) =>
+        Seq("P1", "P1-async", "P2", "P2-seq").flatMap(r => dumpLines(l, sql, r))
+      } ++
+      windowStatements.flatMap { case (l, sql) =>
+        Seq("P1", "P1-async", "P2").flatMap(r => dumpLines(l, sql, r))
+      } ++
+      unionStatements.flatMap { case (l, sql) => unionRoutes.flatMap(r => dumpLines(l, sql, r)) } ++
+      entityStatements.flatMap { case (l, sql) =>
+        Seq("P1", "P2").flatMap(r => dumpLines(l, sql, r))
+      } ++
+      (entityRows match {
+        case Right(rows) =>
+          rows.zipWithIndex.map { case (r, i) => tsv("E", "searchAs", "-", i.toString, r.toString) }
+        case Left(e) => Seq(tsv("E", "searchAs", "-", "-", s"ERROR $e"))
+      }) ++
+      insertLines
+    write("DOTTED_POPULATION_OUT", "target/dotted-population.tsv", lines)
+    lines.exists(_.contains("\tS2\t")) shouldBe true
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // AC1: values on every route, per document, against the aliased control
+  // -------------------------------------------------------------------------------------------
+
+  private def byId(rows: Seq[ListMap[String, Any]]): Map[String, ListMap[String, Any]] =
+    rows.map(r => String.valueOf(r.getOrElse("id", null)) -> r).toMap
+
+  private def cls(v: Any): String = if (v == null) "null" else v.getClass.getName
+
+  it should "carry the value its aliased form carries, per document, on every route (values)" in {
+    rowRoutes.foreach { route =>
+      withClue(s"[$route] ") {
+        val s2 = byId(rowsOk(s"SELECT id, obj.a FROM $table", route))
+        val c1 = byId(rowsOk(s"SELECT id, obj.a AS k FROM $table", route))
+        s2.keySet shouldBe docs.map(_._1).toSet
+        docs.map(_._1).foreach { id =>
+          withClue(s"$id: ") {
+            s2(id)("obj.a") shouldBe valueOfA(id)
+            // A-3: over D-z the walked column reads ES's 4, where the alias loop keeps the null
+            if (id == "dz") c1(id)("k") shouldBe (null: Any)
+            else {
+              s2(id)("obj.a") shouldBe c1(id)("k")
+              cls(s2(id)("obj.a")) shouldBe cls(c1(id)("k"))
+            }
+          }
+        }
+        // S1 / S7 / U1 carry the same multiset of values
+        val values = docs.map(d => valueOfA(d._1)).map(String.valueOf).sorted
+        rowsOk(s"SELECT obj.a FROM $table", route)
+          .map(r => String.valueOf(r("obj.a")))
+          .sorted shouldBe values
+        rowsOk(s"""SELECT id, "obj.a" FROM $table""", route)
+          .map(r => String.valueOf(r("obj.a")))
+          .sorted shouldBe values
+        rowsOk(s"SELECT t.id, t.obj.a FROM $table AS t", route)
+          .map(r => String.valueOf(r("obj.a")))
+          .sorted shouldBe values
+        // S3: a deeper path; a mixed spelling stays NULL (D4)
+        val s3 = byId(rowsOk(s"SELECT id, obj.x.y FROM $table", route))
+        s3("dx")("obj.x.y") shouldBe 9
+        s3("dmx")("obj.x.y") shouldBe (null: Any)
+        // S4 / S6 / U4: the dotted cell beside an alias; S6 reads its alias (A-6)
+        val s4 = byId(rowsOk(s"SELECT id, obj.a, obj.b AS k FROM $table", route))
+        docs.map(_._1).foreach(id => withClue(s"S4 $id: ")(s4(id)("obj.a") shouldBe valueOfA(id)))
+        val s6 = byId(rowsOk(s"SELECT id, obj.a, obj.a AS k FROM $table", route))
+        docs.map(_._1).foreach(id => withClue(s"S6 $id: ")(s6(id)("obj.a") shouldBe s6(id)("k")))
+        s6("dn")("obj.a") shouldBe 1
+        val u6 = byId(rowsOk(s"""SELECT id, n AS "obj.a", obj.b FROM $table""", route))
+        u6("ds")("obj.b") shouldBe 6
+        u6("ds")("obj.a") shouldBe 7
+      }
+    }
+  }
+
+  it should "carry an un-aliased dotted leg's values in a UNION ALL, one-shot and leg by leg (values)" in {
+    val dotted = docs.map(d => valueOfA(d._1))
+    val n = rowsOk(s"SELECT id, n FROM $table", "P2-sorted").map(_("n"))
+    def sortedOf(xs: Seq[Any]): Seq[String] = xs.map(String.valueOf).sorted
+    unionRoutes.foreach { route =>
+      withClue(s"[$route] ") {
+        rowsOk(s"SELECT obj.a FROM $table UNION ALL SELECT n FROM $table", route)
+          .map(_("obj.a")) should contain theSameElementsAs (dotted ++ n)
+        rowsOk(s"SELECT n FROM $table UNION ALL SELECT obj.a FROM $table", route)
+          .map(_("n")) should contain theSameElementsAs (n ++ dotted)
+        // S10 converts the INT leg to the result's DOUBLE
+        val s10 = rowsOk(s"SELECT obj.a FROM $table UNION ALL SELECT dbl FROM $table", route)
+          .map(_("obj.a"))
+        s10.filter(_ != null).foreach(v => v shouldBe a[java.lang.Double])
+        // the legs come in leg order: the first `docs.size` rows are the dotted leg's
+        sortedOf(s10.take(docs.size)) shouldBe sortedOf(dotted.map {
+          case null   => null
+          case i: Int => i.toDouble
+          case other  => other
+        })
+      }
+    }
+  }
+
+  it should "find a window's partition and rank by the value it shows (values)" in {
+    Seq("P1", "P2").foreach { route =>
+      withClue(s"[$route] ") {
+        val w1 = byId(rowsOk(windowStatements.head._2, route))
+        docs.map(_._1).foreach(id => withClue(s"W1 $id: ")(w1(id)("obj.a") shouldBe valueOfA(id)))
+        // W2: the partition key reads the base row's `obj.a`
+        val w2 = byId(rowsOk(windowStatements(1)._2, route))
+        String.valueOf(w2("dn4")("rn")) shouldBe "1"
+        String.valueOf(w2("dz")("rn")) shouldBe "2"
+        String.valueOf(w2("dn")("rn")) shouldBe "1"
+        // W3: D-z and its nested twin tie (A-3)
+        val w3 = byId(rowsOk(windowStatements(2)._2, route))
+        w3("dz")("r") shouldBe w3("dn4")("r")
+      }
+    }
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // AC2: no leaked parent, but every parent that may carry a value
+  // -------------------------------------------------------------------------------------------
+
+  it should "drop the parent it read and keep every other one (parent)" in {
+    rowRoutes.foreach { route =>
+      withClue(s"[$route] ") {
+        val s2 = byId(rowsOk(s"SELECT id, obj.a FROM $table", route))
+        Seq("dn", "dz", "dn4").foreach(id =>
+          withClue(s"$id: ")(s2(id).keySet shouldBe Set("id", "obj.a"))
+        )
+        // a non-null flat spelling won, an array: the parent stays, as main returns it
+        s2("db")("obj") shouldBe Map("a" -> 30)
+        s2("dl")("obj") shouldBe List(Map("a" -> 5), Map("a" -> 50))
+        val s3 = byId(rowsOk(s"SELECT id, obj.x.y FROM $table", route))
+        s3("dx").keySet shouldBe Set("id", "obj.x.y")
+        s3("dmx")("obj") shouldBe Map("x.y" -> 10)
+        // a requested parent, an aliased source under it, an alias collision, a star (A-5, D5)
+        Seq(
+          s"SELECT id, obj, obj.a FROM $table",
+          s"SELECT id, obj.a, obj.b AS k FROM $table",
+          s"SELECT id, obj.a, obj.a AS k FROM $table",
+          s"SELECT id, a AS z, obj.a FROM $table",
+          s"SELECT *, obj.a FROM $table"
+        ).foreach { sql =>
+          withClue(s"$sql: ")(byId(rowsOk(sql, route))("dn") should contain key "obj")
+        }
+      }
+    }
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // AC6: INSERT ... SELECT writes main's documents, and they read back with their values
+  // -------------------------------------------------------------------------------------------
+
+  it should "write main's documents from an INSERT ... SELECT, which read back with their values (insert)" in {
+    inserted
+    Seq(copyTable, scriptedTable, changelogTable).foreach { target =>
+      withClue(s"[$target] ") {
+        // D13: a nested source is written nested, as on main (`{"obj":{"a":1},"id":"dn"}`): the
+        // row's null `obj.a` cell is not written, and no flat spelling carries the value
+        val dn = org.json4s.jackson.JsonMethods.parse(storedSource(target, "dn"))
+        (dn \ "obj.a") shouldBe org.json4s.JNothing
+        (dn \ "obj" \ "a") shouldBe org.json4s.JInt(1)
+        // the un-aliased read walks to the nested value, which main null-filled
+        val back = byId(rowsOk(s"SELECT id, obj.a FROM $target", "P2-sorted"))
+        Seq("dn", "df", "db", "dz", "dn4").foreach { id =>
+          withClue(s"$id: ")(back(id)("obj.a") shouldBe valueOfA(id))
+        }
+      }
+    }
+  }
+
+  // -------------------------------------------------------------------------------------------
+  // AC7: the entity context
+  // -------------------------------------------------------------------------------------------
+
+  it should "resolve a LIMITed WHERE subquery over the column as its aliased form does (entity)" in {
+    def ids(sql: String): Set[String] =
+      rowsOk(sql, "P1-async").map(r => String.valueOf(r("id"))).toSet
+    val in = ids(entityStatements.head._2)
+    in shouldBe ids(entityStatements(2)._2)
+    in shouldBe Set("dn", "df", "dz", "dn4")
+    val notIn = ids(entityStatements(1)._2)
+    notIn shouldBe ids(entityStatements(3)._2)
+    notIn should not be empty
+    // mode P (no LIMIT) reads bucket rows: the unchanged control
+    ids(entityStatements(4)._2) shouldBe in
+    // an entity extraction reads the document's shape, as on main
+    entityRows shouldBe Right(
+      Seq(
+        DottedColumnRow("db", Some(DottedColumnObj(Some(30)))),
+        DottedColumnRow("df", None),
+        DottedColumnRow("dn", Some(DottedColumnObj(Some(1)))),
+        DottedColumnRow("dn4", Some(DottedColumnObj(Some(4)))),
+        DottedColumnRow("dz", Some(DottedColumnObj(Some(4))))
+      )
+    )
+  }
+}
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplGatewayIntegrationSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplGatewayIntegrationSpec.scala
index 3451b251..3b289301 100644
--- a/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplGatewayIntegrationSpec.scala
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplGatewayIntegrationSpec.scala
@@ -657,6 +657,21 @@ trait ReplGatewayIntegrationSpec extends ReplIntegrationTestKit {
     )
   }
 
+  it should "return an un-aliased nested field under its dotted name, without its parent" in {
+    // the aliased test above is the canary that a parent object IS visible on this path
+    val res = executeSync("SELECT id, profile.city FROM dql_users ORDER BY id")
+    assertSelectResult(
+      System.nanoTime(),
+      res,
+      Seq(
+        Map("id" -> 1, "profile.city" -> "Paris"),
+        Map("id" -> 2, "profile.city" -> "Lyon"),
+        Map("id" -> 3, "profile.city" -> "Paris"),
+        Map("id" -> 4, "profile.city" -> "Marseille")
+      )
+    )
+  }
+
   it should "execute a UNION ALL query" in {
     val sql =
       """SELECT id, name FROM dql_users WHERE age > 30