diff --git a/core/src/main/resources/softnetwork-elastic.conf b/core/src/main/resources/softnetwork-elastic.conf
index bd236766..b0e549c2 100644
--- a/core/src/main/resources/softnetwork-elastic.conf
+++ b/core/src/main/resources/softnetwork-elastic.conf
@@ -34,6 +34,11 @@ elastic {
   connection-timeout = 5s
   socket-timeout     = 30s
 
+  # When enabled, result rows surface the Elasticsearch document id as an `_id` column.
+  # Disabled by default: SQL results carry only the selected columns.
+  include-document-id = false
+  include-document-id = ${?ELASTIC_INCLUDE_DOCUMENT_ID}
+
   # Cluster discovery
   discovery {
     enabled = false
diff --git a/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala b/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala
index 9d410f89..a1d4f805 100644
--- a/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala
+++ b/core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala
@@ -39,6 +39,9 @@ import java.time.Duration
   *   Metrics and monitoring configuration
  * @param watcher
  *   Credentials for the watcher component (if applicable)
+ * @param includeDocumentId
+ *   When enabled, result rows surface the Elasticsearch document id as an `_id` column (disabled
+ *   by default)
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -47,7 +50,8 @@ case class ElasticConfig(
   connectionTimeout: Duration,
   socketTimeout: Duration,
   metrics: MetricsConfig,
-  watcher: ElasticCredentials)
+  watcher: ElasticCredentials,
+  includeDocumentId: Boolean = false)
 
 object ElasticConfig extends StrictLogging {
   def apply(config: Config): ElasticConfig = {
diff --git a/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala b/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala
index 21ba2422..78eac170 100644
--- a/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala
+++ b/core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala
@@ -39,6 +39,9 @@ import java.time.Duration
   *   Metrics and monitoring configuration
   * @param watcher
   *   Credentials for the watcher component (if applicable)
+  * @param includeDocumentId
+  *   When enabled, result rows surface the Elasticsearch document id as an `_id` column (disabled
+  *   by default)
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -47,7 +50,8 @@ case class ElasticConfig(
   connectionTimeout: Duration,
   socketTimeout: Duration,
   metrics: MetricsConfig,
-  watcher: ElasticCredentials
+  watcher: ElasticCredentials,
+  includeDocumentId: Boolean = false
 )
 
 object ElasticConfig extends StrictLogging {
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/AggregateApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/AggregateApi.scala
index a2b07e7e..a5db96a2 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/AggregateApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/AggregateApi.scala
@@ -100,7 +100,9 @@ trait SingleValueAggregateApi
 
           case Some(_: Map[_, _]) =>
             val typedMaps = s.asInstanceOf[Seq[Map[String, Any]]]
-            val metadataKeys = Set("_id", "_index", "_score", "_sort")
+            // `_id` is the only hit metadata a per-hit map can still carry (opt-in via
+            // `elastic.include-document-id`); it never counts as an aggregated value.
+            val metadataKeys = Set("_id")
 
             // Check if all maps have the same single non-metadata key
             val nonMetadataKeys = typedMaps.flatMap(_.keys.filterNot(metadataKeys.contains))
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala
index b37c58e0..7972933b 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala
@@ -60,4 +60,9 @@ trait ElasticClientApi
   def config: Config = ConfigFactory.load()
 
   final lazy val elasticConfig: ElasticConfig = ElasticConfig(config)
+
+  /** Result rows surface the document id as an `_id` column only when enabled through
+    * `elastic.include-document-id` (disabled by default).
+    */
+  override protected def includeDocumentId: Boolean = elasticConfig.includeDocumentId
 }
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 a2e576c8..900f4991 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
@@ -28,11 +28,30 @@ import scala.jdk.CollectionConverters._
 
 trait ElasticConversion {
 
+  import ElasticConversion.DocumentIdField
+
   // Distinctly named to avoid colliding with the `logger` member that
   // `ClientCompanion`'s self-type contributes where this trait is mixed in.
   private val conversionLogger: org.slf4j.Logger =
     org.slf4j.LoggerFactory.getLogger(getClass)
 
+  /** Whether result rows surface the Elasticsearch document id as an `_id` column. Disabled by
+    * default; wired to the `elastic.include-document-id` HOCON setting through
+    * [[ElasticClientApi]].
+    */
+  protected def includeDocumentId: Boolean = false
+
+  /** Whether result rows keep an `_id` column: either the document-id column is enabled, or the
+    * query selects `_id` explicitly. Evaluated once per parse / per stream — never per row. Only
+    * window-enrichment base rows ever carry an `_id` the user did not ask for (the ordinal lookup
+    * matches rows by document id — see `retainDocumentId` on the parse methods); the enrichment
+    * paths hoist this decision and drop the id while rebuilding each row. Plain search and scroll
+    * rows never get one injected in the first place, keeping the hot paths free of any per-row
+    * strip.
+    */
+  private[client] def keepsDocumentId(requestedFields: Seq[String]): Boolean =
+    includeDocumentId || requestedFields.contains(DocumentIdField)
+
   def convertTo[T](map: Map[String, Any])(implicit m: Manifest[T], formats: Formats): T = {
     val jValue = Extraction.decompose(map)
     jValue.extract[T]
@@ -68,7 +87,8 @@ trait ElasticConversion {
     aggregations: ListMap[String, ClientAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    explodeNested: Boolean = true
+    explodeNested: Boolean = true,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] = {
     var json = mapper.readTree(results)
     if (json.has("responses")) {
@@ -76,10 +96,26 @@ trait ElasticConversion {
     }
     // Check if it's a multi-search response (array of responses)
     if (json.isArray) {
-      parseMultiSearchResponse(json, fieldAliases, aggregations, fields, nestedHits, explodeNested)
+      parseMultiSearchResponse(
+        json,
+        fieldAliases,
+        aggregations,
+        fields,
+        nestedHits,
+        explodeNested,
+        retainDocumentId
+      )
     } else {
       // Single search response
-      parseSingleSearchResponse(json, fieldAliases, aggregations, fields, nestedHits, explodeNested)
+      parseSingleSearchResponse(
+        json,
+        fieldAliases,
+        aggregations,
+        fields,
+        nestedHits,
+        explodeNested,
+        retainDocumentId
+      )
     }
   }
 
@@ -91,7 +127,8 @@ trait ElasticConversion {
     aggregations: ListMap[String, ClientAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    explodeNested: Boolean = true
+    explodeNested: Boolean = true,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] =
     Try {
       val responses = jsonArray.elements().asScala.toList
@@ -111,7 +148,15 @@ trait ElasticConversion {
         // Parse each response and combine all rows
         val allRows = responses.flatMap { response =>
           if (!response.has("error")) {
-            jsonToRows(response, fieldAliases, aggregations, fields, nestedHits, explodeNested)
+            jsonToRows(
+              response,
+              fieldAliases,
+              aggregations,
+              fields,
+              nestedHits,
+              explodeNested,
+              retainDocumentId
+            )
           } else {
             Seq.empty
           }
@@ -128,7 +173,8 @@ trait ElasticConversion {
     aggregations: ListMap[String, ClientAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    explodeNested: Boolean = true
+    explodeNested: Boolean = true,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] =
     Try {
       // check if it is an error response
@@ -138,7 +184,15 @@ trait ElasticConversion {
           .getOrElse("Unknown Elasticsearch error")
         throw new Exception(s"Elasticsearch error: $errorMsg")
       } else {
-        jsonToRows(json, fieldAliases, aggregations, fields, nestedHits, explodeNested)
+        jsonToRows(
+          json,
+          fieldAliases,
+          aggregations,
+          fields,
+          nestedHits,
+          explodeNested,
+          retainDocumentId
+        )
       }
     }
 
@@ -150,7 +204,8 @@ trait ElasticConversion {
     aggregations: ListMap[String, ClientAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    explodeNested: Boolean = true
+    explodeNested: Boolean = true,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val hitsNode = Option(json.path("hits").path("hits"))
       .filter(_.isArray)
@@ -162,7 +217,7 @@ trait ElasticConversion {
     val rows = (hitsNode, aggsNode) match {
       case (Some(hits), None) if hits.nonEmpty =>
         // Case 1 : only hits
-        val parsed = parseSimpleHits(hits, fieldAliases, fields)
+        val parsed = parseSimpleHits(hits, fieldAliases, fields, retainDocumentId)
         if (explodeNested) flattenInnerHits(parsed, nestedHits)
         else parsed
 
@@ -201,7 +256,7 @@ trait ElasticConversion {
           extractAllTopHits(aggs, fieldAliases, aggregations),
           aggregations
         )
-        val parsed = parseSimpleHits(hits, fieldAliases, fields)
+        val parsed = parseSimpleHits(hits, fieldAliases, fields, retainDocumentId)
         val flattened =
           if (explodeNested) flattenInnerHits(parsed, nestedHits)
           else parsed
@@ -233,12 +288,18 @@ trait ElasticConversion {
     *   all requested fields in their original SQL SELECT order (output column names). When
     *   non-empty, each row is normalized to include all requested fields (with null for missing
     *   ones) and ordered to match the SQL SELECT order.
+    * @param retainDocumentId
+    *   force the `_id` row key even when [[keepsDocumentId]] does not hold — used by
+    *   window-enrichment base queries, whose ordinal lookup matches rows by document id.
     */
   def parseSimpleHits(
     hits: List[JsonNode],
     fieldAliases: ListMap[String, String],
-    requestedFields: Seq[String] = Seq.empty
+    requestedFields: Seq[String] = Seq.empty,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
+    // One decision per parse — the per-row hot path does no `_id` work when it is not kept
+    val withDocumentId = retainDocumentId || keepsDocumentId(requestedFields)
     hits.map { hit =>
       var source = extractSource(hit, fieldAliases)
       fieldAliases.foreach(entry => {
@@ -249,7 +310,8 @@ trait ElasticConversion {
           }
         }
       })
-      val metadata = extractHitMetadata(hit)
+      val metadata =
+        if (withDocumentId) extractHitId(hit) else ListMap.empty[String, Any]
       val innerHits = extractInnerHits(hit, fieldAliases)
       val fieldsNode = Option(hit.path("fields"))
         .filter(!_.isMissingNode)
@@ -261,8 +323,8 @@ 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 (metadata like _id,
-    * _index, etc.) are appended after the requested fields.
+    * Fields missing from the row are added with null value. Extra fields (such as the internally
+    * carried `_id`) are appended after the requested fields.
     */
   protected def normalizeRow(
     row: ListMap[String, Any],
@@ -327,20 +389,19 @@ trait ElasticConversion {
     }
   }
 
-  /** Extract hit metadata (_id, _index, _score)
+  /** Extract the hit `_id`.
+    *
+    * This is the only hit metadata ever extracted: `_index`, `_score` and `_sort` are never
+    * surfaced. A parsed row carries `_id` only when [[includeDocumentId]] is enabled, the query
+    * selects `_id` explicitly, or the parse serves a window-enrichment base query (the ranking
+    * ordinal lookup in `SearchApi.enrichDocumentWithWindowValues` matches rows by document id —
+    * `retainDocumentId` on the parse methods).
     */
-  def extractHitMetadata(hit: JsonNode): ListMap[String, Any] = {
-    ListMap(
-      "_id"    -> Option(hit.get("_id")).map(_.asText()),
-      "_index" -> Option(hit.get("_index")).map(_.asText()),
-      "_score" -> Option(hit.get("_score")).map(n =>
-        if (n.isDouble || n.isFloat) n.asDouble() else n.asLong().toDouble
-      ),
-      "_sort" -> Option(hit.get("sort"))
-        .filter(_.isArray)
-        .map(sortNode => sortNode.elements().asScala.map(jsonNodeToAny(_, ListMap.empty)).toList)
-    ).collect { case (k, Some(v)) => k -> v }
-  }
+  def extractHitId(hit: JsonNode): ListMap[String, Any] =
+    Option(hit.get("_id")) match {
+      case Some(id) => ListMap(DocumentIdField -> id.asText())
+      case None     => ListMap.empty
+    }
 
   /** Extract hit _source
     */
@@ -368,9 +429,12 @@ trait ElasticConversion {
                 .elements()
                 .asScala
                 .map { innerHit =>
-                  // Extract source and metadata for each inner hit
+                  // Inner-hit `_id` is never needed internally: only expose it when the
+                  // document-id column is enabled (nested rows are out of reach of the
+                  // top-level egress strip).
                   val source = extractSource(innerHit, fieldAliases)
-                  val metadata = extractHitMetadata(innerHit)
+                  val metadata =
+                    if (includeDocumentId) extractHitId(innerHit) else ListMap.empty[String, Any]
 
                   // Recursively handle nested inner_hits if present
                   val nestedInnerHits = extractInnerHits(innerHit, fieldAliases)
@@ -813,13 +877,18 @@ trait ElasticConversion {
           // Ranking windows need every hit's `_id` for the per-row ordinal
           // lookup in `SearchApi.enrichDocumentWithWindowValues`. Skip the
           // "single-source-field → return the scalar" shortcut so the
-          // per-hit map preserves both the ORDER BY columns AND the metadata.
+          // per-hit map preserves both the ORDER BY columns AND the id.
+          // Everywhere else the per-hit `_id` is only exposed when the
+          // document-id column is enabled (these maps are nested inside
+          // aggregation values, out of reach of the top-level egress strip).
           val isRanking = agg.exists(_.ranking)
+          def optionalHitId(hit: JsonNode): ListMap[String, Any] =
+            if (includeDocumentId) extractHitId(hit) else ListMap.empty
           val processedHits = hits.map { hit =>
             val source = extractSource(hit, fieldAliases)
             if (hasMultipleValues) {
               if (isRanking) {
-                val metadata = extractHitMetadata(hit)
+                val metadata = extractHitId(hit)
                 val innerHits = extractInnerHits(hit, fieldAliases)
                 source ++ metadata ++ innerHits
               } else {
@@ -835,13 +904,13 @@ trait ElasticConversion {
                     }
                   case _ =>
                     // Multiple fields: return as object
-                    val metadata = extractHitMetadata(hit)
+                    val metadata = optionalHitId(hit)
                     val innerHits = extractInnerHits(hit, fieldAliases)
                     source ++ metadata ++ innerHits
                 }
               }
             } else {
-              val metadata = extractHitMetadata(hit)
+              val metadata = optionalHitId(hit)
               val innerHits = extractInnerHits(hit, fieldAliases)
               source ++ metadata ++ innerHits ++ ListMap("bucket_root" -> bucketRoot)
             }
@@ -990,4 +1059,10 @@ trait ElasticConversion {
   }
 }
 
-object ElasticConversion extends ElasticConversion
+object ElasticConversion extends ElasticConversion {
+
+  /** Name of the opt-in document-id column surfaced on result rows when
+    * `elastic.include-document-id` is enabled.
+    */
+  val DocumentIdField: String = "_id"
+}
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 80c94b47..c1552941 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala
@@ -1334,7 +1334,9 @@ trait IndicesApi extends ElasticClientHelpers {
           implicit val context: ConversionContext = NativeContext
           Right(
             scroll(single).map { case (row, _) =>
-              val jsonNode: JsonNode = row - "_id" - "_index" - "_score" - "_sort"
+              // Even with `elastic.include-document-id` enabled, the id column must never be
+              // written into the target documents' _source.
+              val jsonNode: JsonNode = row - ElasticConversion.DocumentIdField
               jsonNode.toString
             }
           )
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 7d054a4c..5f377a53 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ScrollApi.scala
@@ -163,6 +163,7 @@ trait ScrollApi extends ElasticClientHelpers {
         )
           return scrollWithWindowEnrichment(single, config)
 
+        val requestedFields = extractOutputFieldNames(single)
         val elasticQuery =
           ElasticQuery(
             single,
@@ -174,9 +175,11 @@ trait ScrollApi extends ElasticClientHelpers {
           elasticQuery,
           single.fieldAliases,
           single.sqlAggregations,
-          config,
+          // `_id` is injected at parse time only when it will be kept — the streamed rows
+          // need no per-row strip on this hot path.
+          config.copy(retainDocumentId = keepsDocumentId(requestedFields)),
           single.sorts.nonEmpty,
-          extractOutputFieldNames(single),
+          requestedFields,
           single.nestedHitsMappings
         )
 
@@ -486,6 +489,11 @@ trait ScrollApi extends ElasticClientHelpers {
     val baseQuery = createBaseQuery(request)
 
     // Stream and enrich
+    val outputFields = extractOutputFieldNames(request)
+
+    // Determine if we should keep the document ID in the output
+    val shouldKeepDocumentId = keepsDocumentId(outputFields)
+
     Source
       .futureSource(
         windowCacheFuture.map {
@@ -498,14 +506,19 @@ trait ScrollApi extends ElasticClientHelpers {
               ),
               baseQuery.fieldAliases,
               baseQuery.sqlAggregations,
-              config,
+              // The base rows must carry `_id`: the ordinal lookup below matches each row to
+              // its per-partition rankings by document id.
+              config.copy(retainDocumentId = true),
               baseQuery.sorts.nonEmpty,
               extractOutputFieldNames(baseQuery),
               baseQuery.nestedHitsMappings
             )
               .map { case (doc, metrics) =>
                 val enrichedDoc = enrichDocumentWithWindowValues(doc, cache, request)
-                val normalizedDoc = normalizeRow(enrichedDoc, extractOutputFieldNames(request))
+                var normalizedDoc = normalizeRow(enrichedDoc, outputFields)
+                if (!shouldKeepDocumentId) {
+                  normalizedDoc = normalizedDoc - ElasticConversion.DocumentIdField
+                }
                 (normalizedDoc, metrics)
               }
 
@@ -526,7 +539,7 @@ trait ScrollApi extends ElasticClientHelpers {
                 ),
                 baseQuery.fieldAliases,
                 baseQuery.sqlAggregations,
-                config,
+                config.copy(retainDocumentId = shouldKeepDocumentId),
                 baseQuery.sorts.nonEmpty
               )
             }
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 82a07741..f58dbd4f 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
@@ -208,6 +208,20 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty
+  )(implicit context: ConversionContext): ElasticResult[ElasticResponse] =
+    singleSearchInternal(elasticQuery, fieldAliases, aggregations, fields, nestedHits)
+
+  /** [[singleSearch]] with an explicit document-id retention decision. `retainDocumentId = true` is
+    * reserved for the window-enrichment base query, which matches rows to their ranking ordinals by
+    * document id AFTER parsing (the id is stripped again after enrichment).
+    */
+  private[client] def singleSearchInternal(
+    elasticQuery: ElasticQuery,
+    fieldAliases: ListMap[String, String],
+    aggregations: ListMap[String, SQLAggregation],
+    fields: Seq[String] = Seq.empty,
+    nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
+    retainDocumentId: Boolean = false
   )(implicit context: ConversionContext): ElasticResult[ElasticResponse] = {
     validateJson("search", elasticQuery.query) match {
       case Some(error) =>
@@ -243,7 +257,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
             aggs,
             fields,
             nestedHits,
-            elasticQuery.explodeNested
+            elasticQuery.explodeNested,
+            retainDocumentId
           )
         ) match {
           case success @ ElasticSuccess(_) =>
@@ -1517,7 +1532,9 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
 
     logger.info(s"🔍 Executing base query without window functions ${baseQuery.sql}")
 
-    singleSearch(
+    // Retain `_id` on the parsed rows: the enrichment step matches each base row to its
+    // ranking ordinals by document id. The id is stripped after enrichment.
+    singleSearchInternal(
       ElasticQuery(
         baseQuery,
         collection.immutable.Seq(baseQuery.sources: _*),
@@ -1526,7 +1543,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
       baseQuery.fieldAliases,
       baseQuery.sqlAggregations,
       extractOutputFieldNames(baseQuery),
-      baseQuery.nestedHitsMappings
+      baseQuery.nestedHitsMappings,
+      retainDocumentId = true
     )
   }
 
@@ -1600,10 +1618,18 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers {
     val baseRows = response.results
     val outputFields = extractOutputFieldNames(request)
 
-    // Enrich each row with window values, then normalize field order
+    // Determine ONCE whether the document ID stays in the output — never per row
+    val shouldKeepDocumentId = keepsDocumentId(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)
-      normalizeRow(enriched, outputFields)
+      val normalized = normalizeRow(enriched, outputFields)
+      if (shouldKeepDocumentId) normalized
+      else normalized - ElasticConversion.DocumentIdField
     }
 
     ElasticResult.success(response.copy(results = enrichedRows))
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/scroll/package.scala b/core/src/main/scala/app/softnetwork/elastic/client/scroll/package.scala
index 8df320a8..ac837123 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/scroll/package.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/scroll/package.scala
@@ -29,7 +29,12 @@ package object scroll {
       true, // false = force classic scroll even where PIT/search_after is available (slower; for clusters restricting the PIT API)
     metrics: ScrollMetrics = ScrollMetrics(), // Initial scroll metrics
     retryConfig: RetryConfig = RetryConfig(), // Retry configuration
-    failOnWindowError: Option[Boolean] = None
+    failOnWindowError: Option[Boolean] = None,
+    // Internal (set by ScrollApi, not by callers): parsed pages keep the document id as an
+    // `_id` row key. True for window-enrichment base queries (the ordinal lookup matches rows
+    // by document id), when `elastic.include-document-id` is enabled, or when the query
+    // selects `_id` explicitly. False keeps the hot scroll path free of any per-row overhead.
+    retainDocumentId: Boolean = false
   )
 
   /** Scroll strategy based on query type
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala
index af512661..c1234bfe 100644
--- a/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala
+++ b/core/src/test/scala/app/softnetwork/elastic/client/ElasticConversionSpec.scala
@@ -48,8 +48,8 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver
     parseResponse(results, ListMap.empty, ListMap.empty) match {
       case Success(rows) =>
         rows.foreach(println)
-      // Map(name -> Laptop, price -> 999.99, category -> Electronics, tags -> List(computer, portable), _id -> 1, _index -> products, _score -> 1.0)
-      // Map(name -> Mouse, price -> 29.99, category -> Electronics, _id -> 2, _index -> products, _score -> 0.8)
+      // Map(name -> Laptop, price -> 999.99, category -> Electronics, tags -> List(computer, portable))
+      // Map(name -> Mouse, price -> 29.99, category -> Electronics)
       case Failure(error) =>
         throw error
     }
@@ -86,7 +86,7 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver
     ) match {
       case Success(rows) =>
         rows.foreach(println)
-        // Map(name -> Alice, address -> Map(street -> 123 Main St, city -> Wonderland, country -> Fictionland), _id -> u1, _index -> users, _score -> 1.0)
+        // Map(name -> Alice, address -> Map(street -> 123 Main St, city -> Wonderland, country -> Fictionland))
         val users = rows.map(row => convertTo[User](row))
         users.foreach(println)
         // User(u1,Alice,Address(123 Main St,Wonderland,Fictionland))
@@ -1617,6 +1617,141 @@ class ElasticConversionSpec extends AnyFlatSpec with Matchers with ElasticConver
     tryParseAsDateTime("2026/08/06") shouldBe None
     tryParseAsDateTime("v1:2026") shouldBe None
   }
+
+  // -------------------------------------------------------------------------
+  // Hit metadata: `_index` / `_score` / `_sort` are never surfaced; `_id` is
+  // carried through parsing (ranking-window lookup) and stripped at the public
+  // egress unless `elastic.include-document-id` is enabled or `_id` is selected.
+  // -------------------------------------------------------------------------
+
+  private object EnabledDocumentIdConversion extends ElasticConversion {
+    override protected def includeDocumentId: Boolean = true
+  }
+
+  private val metadataRichResponse =
+    """{
+      |  "took": 5,
+      |  "hits": {
+      |    "total": { "value": 1, "relation": "eq" },
+      |    "max_score": 1.0,
+      |    "hits": [
+      |      {
+      |        "_index": "products",
+      |        "_id": "1",
+      |        "_score": 1.0,
+      |        "sort": [999, "laptop"],
+      |        "_source": { "name": "Laptop", "price": 999.99 }
+      |      }
+      |    ]
+      |  }
+      |}""".stripMargin
+
+  "hit metadata" should "surface no metadata at all on parsed rows by default" in {
+    parseResponse(metadataRichResponse, ListMap.empty, ListMap.empty) match {
+      case Success(rows) =>
+        rows should have size 1
+        val row = rows.head
+        row("name") shouldBe "Laptop"
+        // No injection at all on the default path — the hot paths never pay a per-row strip
+        row.keySet should contain noneOf ("_id", "_index", "_score", "_sort")
+      case Failure(error) =>
+        throw error
+    }
+  }
+
+  it should "carry _id when the parse retains it for window enrichment" in {
+    parseResponse(
+      metadataRichResponse,
+      ListMap.empty,
+      ListMap.empty,
+      retainDocumentId = true
+    ) match {
+      case Success(rows) =>
+        rows.head("_id") shouldBe "1"
+        rows.head.keySet should contain noneOf ("_index", "_score", "_sort")
+      case Failure(error) =>
+        throw error
+    }
+  }
+
+  it should "carry _id when the query selects it explicitly or the column is enabled" in {
+    // Explicit selection
+    parseResponse(
+      metadataRichResponse,
+      ListMap.empty,
+      ListMap.empty,
+      fields = Seq("name", "_id")
+    ) match {
+      case Success(rows) =>
+        rows.head("_id") shouldBe "1"
+      case Failure(error) => throw error
+    }
+    // Enabled document-id column
+    EnabledDocumentIdConversion.parseResponse(
+      metadataRichResponse,
+      ListMap.empty,
+      ListMap.empty
+    ) match {
+      case Success(rows) =>
+        rows.head("_id") shouldBe "1"
+      case Failure(error) => throw error
+    }
+  }
+
+  it should "keep the document id only when enabled or selected" in {
+    // Disabled by default
+    keepsDocumentId(Seq("name")) shouldBe false
+    // Kept when the query selects `_id` explicitly
+    keepsDocumentId(Seq("name", "_id")) shouldBe true
+    // Kept when the document-id column is enabled
+    EnabledDocumentIdConversion.keepsDocumentId(Seq("name")) shouldBe true
+  }
+
+  it should "expose the inner-hit document id only when enabled" in {
+    val results =
+      """{
+        |  "took": 5,
+        |  "hits": {
+        |    "total": { "value": 1, "relation": "eq" },
+        |    "hits": [
+        |      {
+        |        "_index": "users",
+        |        "_id": "u1",
+        |        "_score": 1.0,
+        |        "_source": { "name": "Alice" },
+        |        "inner_hits": {
+        |          "orders": {
+        |            "hits": {
+        |              "hits": [
+        |                {
+        |                  "_index": "orders",
+        |                  "_id": "o1",
+        |                  "_score": 1.0,
+        |                  "_source": { "total": 42 }
+        |                }
+        |              ]
+        |            }
+        |          }
+        |        }
+        |      }
+        |    ]
+        |  }
+        |}""".stripMargin
+
+    def innerHitKeys(conversion: ElasticConversion): Set[String] =
+      conversion.parseResponse(results, ListMap.empty, ListMap.empty) match {
+        case Success(rows) =>
+          rows.head("orders") match {
+            case hits: List[_] =>
+              hits.head.asInstanceOf[ListMap[String, Any]].keySet.toSet
+            case other => fail(s"Unexpected inner hits: $other")
+          }
+        case Failure(error) => throw error
+      }
+
+    innerHitKeys(this) shouldBe Set("total")
+    innerHitKeys(EnabledDocumentIdConversion) shouldBe Set("total", "_id")
+  }
 }
 
 case class Products(category: String, top_products: List[Product], avg_price: Double)
diff --git a/documentation/client/common_principles.md b/documentation/client/common_principles.md
index b137f656..330140c9 100644
--- a/documentation/client/common_principles.md
+++ b/documentation/client/common_principles.md
@@ -483,6 +483,11 @@ elastic {
   connection-timeout = 5s
   socket-timeout     = 30s
   
+  # Result rows
+  # When enabled, every result row surfaces the Elasticsearch document id as an
+  # `_id` column. Disabled by default: SQL results carry only the selected columns.
+  include-document-id = false
+  
   # Cluster discovery
   discovery {
     enabled   = false
@@ -516,6 +521,9 @@ export ELASTIC_CREDENTIALS_PASSWORD="secret"
 
 # Override port
 export ELASTIC_PORT=9243
+
+# Surface the document id as an `_id` column on result rows
+export ELASTIC_INCLUDE_DOCUMENT_ID=true
 ```
 
 ### Loading Configuration
diff --git a/documentation/client/indices.md b/documentation/client/indices.md
index db99ff36..406e6476 100644
--- a/documentation/client/indices.md
+++ b/documentation/client/indices.md
@@ -523,7 +523,8 @@ Inserts documents produced by a SELECT query.
 Behavior:
 
 - Executes a scroll query.
-- Removes Elasticsearch metadata (`_id`, `_index`, `_score`, `_sort`).
+- Never writes the document id into the target documents (the optional `_id` column — see
+  `elastic.include-document-id` — is removed before indexing).
 - Maps SELECT output to INSERT columns **by name**.
 - Supports aliasing.
 - Supports INSERT without column list.
diff --git a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala
index f629d3f6..311b386e 100644
--- a/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala
+++ b/es6/jest/src/main/scala/app/softnetwork/elastic/client/jest/JestScrollApi.scala
@@ -80,7 +80,12 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers {
 
                 // Extract ALL results (hits + aggregations)
                 val results =
-                  extractAllResults(result.getJsonObject.toString, fieldAliases, aggregations)
+                  extractAllResults(
+                    result.getJsonObject.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.info(
                   s"Initial scroll returned ${results.size} results, scrollId: $scrollId"
@@ -104,7 +109,12 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers {
                 }
                 val newScrollId = result.getJsonObject.get("_scroll_id").getAsString
                 val results =
-                  extractAllResults(result.getJsonObject.toString, fieldAliases, aggregations)
+                  extractAllResults(
+                    result.getJsonObject.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.debug(s"Scroll returned ${results.size} results")
 
@@ -204,7 +214,8 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers {
               throw new IOException(s"Search after failed: ${result.getErrorMessage}")
             }
             // Extract ONLY hits (no aggregations)
-            val hits = extractHitsOnly(result.getJsonObject.toString, fieldAliases)
+            val hits =
+              extractHitsOnly(result.getJsonObject.toString, fieldAliases, config.retainDocumentId)
 
             if (hits.isEmpty) {
               None
@@ -268,12 +279,14 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers {
   private def extractAllResults(
     jsonString: String,
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, SQLAggregation]
+    aggregations: ListMap[String, SQLAggregation],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     parseResponse(
       jsonString,
       fieldAliases,
-      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2))
+      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)),
+      retainDocumentId = retainDocumentId
     ) match {
       case Success(rows) => rows
       case Failure(ex) =>
@@ -286,10 +299,16 @@ trait JestScrollApi extends ScrollApi with JestClientHelpers {
     */
   private def extractHitsOnly(
     jsonString: String,
-    fieldAliases: ListMap[String, String]
+    fieldAliases: ListMap[String, String],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
 
-    parseResponse(jsonString, fieldAliases, ListMap.empty) match {
+    parseResponse(
+      jsonString,
+      fieldAliases,
+      ListMap.empty,
+      retainDocumentId = retainDocumentId
+    ) match {
       case Success(rows) => rows
       case Failure(ex) =>
         logger.error(s"Failed to parse Jest search after response: ${ex.getMessage}", ex)
diff --git a/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientHitMetadataSpec.scala b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientHitMetadataSpec.scala
new file mode 100644
index 00000000..cf1e999e
--- /dev/null
+++ b/es6/jest/src/test/scala/app/softnetwork/elastic/client/JestClientHitMetadataSpec.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 JestClientHitMetadataSpec extends HitMetadataSpec
diff --git a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
index 1bf97c25..3324df10 100644
--- a/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
+++ b/es6/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
@@ -1489,7 +1489,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
                 }
 
                 // Extract both hits AND aggregations
-                val results = extractAllResults(response.toString, fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    response.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.info(s"Initial scroll returned ${results.size} results, scrollId: $scrollId")
 
@@ -1518,7 +1524,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
                 }
 
                 val newScrollId = result.getScrollId
-                val results = extractAllResults(result.toString, fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    result.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.debug(s"Scroll returned ${results.size} results")
 
@@ -1624,7 +1636,7 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
             }
 
             // Extract ONLY hits (no aggregations for search_after)
-            val hits = extractHitsOnly(response.toString, fieldAliases)
+            val hits = extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId)
 
             if (hits.isEmpty) {
               None
@@ -1669,12 +1681,14 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
   private def extractAllResults(
     jsonString: String,
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, SQLAggregation]
+    aggregations: ListMap[String, SQLAggregation],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     parseResponse(
       jsonString,
       fieldAliases,
-      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2))
+      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)),
+      retainDocumentId = retainDocumentId
     ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} rows from response")
@@ -1689,9 +1703,15 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
     */
   private def extractHitsOnly(
     jsonString: String,
-    fieldAliases: ListMap[String, String]
+    fieldAliases: ListMap[String, String],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
-    parseResponse(jsonString, fieldAliases, ListMap.empty) match {
+    parseResponse(
+      jsonString,
+      fieldAliases,
+      ListMap.empty,
+      retainDocumentId = retainDocumentId
+    ) match {
       case Success(rows) => rows
       case Failure(ex) =>
         logger.error(s"Failed to parse search after response: ${ex.getMessage}", ex)
diff --git a/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.scala b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.scala
new file mode 100644
index 00000000..a88d5972
--- /dev/null
+++ b/es6/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.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 RestHighLevelClientHitMetadataSpec extends HitMetadataSpec
diff --git a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
index 2e972347..d58920ab 100644
--- a/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
+++ b/es7/rest/src/main/scala/app/softnetwork/elastic/client/rest/RestHighLevelClientApi.scala
@@ -1519,7 +1519,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
                 }
 
                 // Extract both hits AND aggregations
-                val results = extractAllResults(response.toString, fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    response.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.info(s"Initial scroll returned ${results.size} results, scrollId: $scrollId")
 
@@ -1548,7 +1554,13 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
                 }
 
                 val newScrollId = result.getScrollId
-                val results = extractAllResults(result.toString, fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    result.toString,
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 logger.debug(s"Scroll returned ${results.size} results")
 
@@ -1657,7 +1669,7 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
             }
 
             // Extract ONLY hits (no aggregations for search_after)
-            val hits = extractHitsOnly(response.toString, fieldAliases)
+            val hits = extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId)
 
             if (hits.isEmpty) {
               None
@@ -1794,7 +1806,8 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
                     )
                   }
 
-                  val hits = extractHitsOnly(response.toString, fieldAliases)
+                  val hits =
+                    extractHitsOnly(response.toString, fieldAliases, config.retainDocumentId)
 
                   if (hits.isEmpty) {
                     None // end of stream — watchTermination owns the single PIT close (#202)
@@ -1886,12 +1899,14 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
   private def extractAllResults(
     jsonString: String,
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, SQLAggregation]
+    aggregations: ListMap[String, SQLAggregation],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     parseResponse(
       jsonString,
       fieldAliases,
-      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2))
+      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)),
+      retainDocumentId = retainDocumentId
     ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} rows from response")
@@ -1906,9 +1921,15 @@ trait RestHighLevelClientScrollApi extends ScrollApi with RestHighLevelClientHel
     */
   private def extractHitsOnly(
     jsonString: String,
-    fieldAliases: ListMap[String, String]
+    fieldAliases: ListMap[String, String],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
-    parseResponse(jsonString, fieldAliases, ListMap.empty) match {
+    parseResponse(
+      jsonString,
+      fieldAliases,
+      ListMap.empty,
+      retainDocumentId = retainDocumentId
+    ) match {
       case Success(rows) => rows
       case Failure(ex) =>
         logger.error(s"Failed to parse search after response: ${ex.getMessage}", ex)
diff --git a/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.scala b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.scala
new file mode 100644
index 00000000..a88d5972
--- /dev/null
+++ b/es7/rest/src/test/scala/app/softnetwork/elastic/client/RestHighLevelClientHitMetadataSpec.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 RestHighLevelClientHitMetadataSpec extends HitMetadataSpec
diff --git a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
index 02a70a35..95adb33e 100644
--- a/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
+++ b/es8/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
@@ -1415,7 +1415,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                   throw new IllegalStateException("Scroll ID is null in response")
                 }
 
-                val results = extractAllResults(Left(response), fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    Left(response),
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 if (results.isEmpty || scrollId == null) None
                 else Some((Some(scrollId), results))
@@ -1447,7 +1453,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                 }
 
                 val newScrollId = response.scrollId()
-                val results = extractAllResults(Right(response), fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    Right(response),
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 if (results.isEmpty) {
                   clearScroll(scrollId)
@@ -1620,7 +1632,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                     throw new IOException(s"PIT search_after failed: $errorMsg")
                   }
 
-                  val hits = extractHitsOnly(response, fieldAliases)
+                  val hits = extractHitsOnly(response, fieldAliases, config.retainDocumentId)
 
                   if (hits.isEmpty) {
                     None // end of stream — watchTermination owns the single PIT close (#202)
@@ -1733,7 +1745,8 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
   private def extractAllResults(
     response: Either[SearchResponse[JMap[String, Object]], ScrollResponse[JMap[String, Object]]],
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, SQLAggregation]
+    aggregations: ListMap[String, SQLAggregation],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val jsonString =
       response match {
@@ -1744,7 +1757,8 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
     parseResponse(
       jsonString,
       fieldAliases,
-      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2))
+      aggregations.map(kv => kv._1 -> implicitly[ClientAggregation](kv._2)),
+      retainDocumentId = retainDocumentId
     ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} rows from response (hits + aggregations)")
@@ -1759,11 +1773,17 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
     */
   private def extractHitsOnly(
     response: SearchResponse[JMap[String, Object]],
-    fieldAliases: ListMap[String, String]
+    fieldAliases: ListMap[String, String],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val jsonString = convertToJson(response)
 
-    parseResponse(jsonString, fieldAliases, ListMap.empty) match {
+    parseResponse(
+      jsonString,
+      fieldAliases,
+      ListMap.empty,
+      retainDocumentId = retainDocumentId
+    ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} hits from response")
         rows
diff --git a/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.scala b/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.scala
new file mode 100644
index 00000000..5d3fa58e
--- /dev/null
+++ b/es8/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.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 JavaClientHitMetadataSpec extends HitMetadataSpec
diff --git a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
index fa116d2b..ae6bab25 100644
--- a/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
+++ b/es9/java/src/main/scala/app/softnetwork/elastic/client/java/JavaClientApi.scala
@@ -1415,7 +1415,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                   throw new IllegalStateException("Scroll ID is null in response")
                 }
 
-                val results = extractAllResults(Left(response), fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    Left(response),
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 if (results.isEmpty || scrollId == null) None
                 else Some((Some(scrollId), results))
@@ -1447,7 +1453,13 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                 }
 
                 val newScrollId = response.scrollId()
-                val results = extractAllResults(Right(response), fieldAliases, aggregations)
+                val results =
+                  extractAllResults(
+                    Right(response),
+                    fieldAliases,
+                    aggregations,
+                    config.retainDocumentId
+                  )
 
                 if (results.isEmpty) {
                   clearScroll(scrollId)
@@ -1620,7 +1632,7 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
                     throw new IOException(s"PIT search_after failed: $errorMsg")
                   }
 
-                  val hits = extractHitsOnly(response, fieldAliases)
+                  val hits = extractHitsOnly(response, fieldAliases, config.retainDocumentId)
 
                   if (hits.isEmpty) {
                     None // end of stream — watchTermination owns the single PIT close (#202)
@@ -1733,7 +1745,8 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
   private def extractAllResults(
     response: Either[SearchResponse[JMap[String, Object]], ScrollResponse[JMap[String, Object]]],
     fieldAliases: ListMap[String, String],
-    aggregations: ListMap[String, SQLAggregation]
+    aggregations: ListMap[String, SQLAggregation],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val jsonString =
       response match {
@@ -1744,7 +1757,8 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
     parseResponse(
       jsonString,
       fieldAliases,
-      aggregations.map(kv => kv._1 -> kv._2)
+      aggregations.map(kv => kv._1 -> kv._2),
+      retainDocumentId = retainDocumentId
     ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} rows from response (hits + aggregations)")
@@ -1759,11 +1773,17 @@ trait JavaClientScrollApi extends ScrollApi with JavaClientHelpers {
     */
   private def extractHitsOnly(
     response: SearchResponse[JMap[String, Object]],
-    fieldAliases: ListMap[String, String]
+    fieldAliases: ListMap[String, String],
+    retainDocumentId: Boolean
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val jsonString = convertToJson(response)
 
-    parseResponse(jsonString, fieldAliases, ListMap.empty) match {
+    parseResponse(
+      jsonString,
+      fieldAliases,
+      ListMap.empty,
+      retainDocumentId = retainDocumentId
+    ) match {
       case Success(rows) =>
         logger.debug(s"Parsed ${rows.size} hits from response")
         rows
diff --git a/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.scala b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.scala
new file mode 100644
index 00000000..5d3fa58e
--- /dev/null
+++ b/es9/java/src/test/scala/app/softnetwork/elastic/client/JavaClientHitMetadataSpec.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 JavaClientHitMetadataSpec extends HitMetadataSpec
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayIntegrationTestKit.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayIntegrationTestKit.scala
index d6f6a6e1..1823f2a7 100644
--- a/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayIntegrationTestKit.scala
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/GatewayIntegrationTestKit.scala
@@ -92,7 +92,9 @@ trait GatewayIntegrationTestKit extends AnyFlatSpecLike with Matchers with Scala
   // -------------------------------------------------------------------------
 
   private def normalizeRow(row: ListMap[String, Any]): ListMap[String, Any] = {
-    val updated = row - "_id" - "_index" - "_score" - "_version" - "_sort"
+    // `_id` only ever appears when `elastic.include-document-id` is enabled — strip it (and the
+    // get-api `_version`) so expected-row assertions stay independent of the client configuration.
+    val updated = row - "_id" - "_version"
     updated.map(entry =>
       entry._2 match {
         case m: ListMap[_, _] =>
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/HitMetadataSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/HitMetadataSpec.scala
new file mode 100644
index 00000000..c9819a11
--- /dev/null
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/HitMetadataSpec.scala
@@ -0,0 +1,260 @@
+/*
+ * 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.Source
+import app.softnetwork.elastic.client.bulk._
+import app.softnetwork.elastic.client.result.{ElasticFailure, ElasticSuccess}
+import app.softnetwork.elastic.client.spi.ElasticClientSpi
+import app.softnetwork.elastic.scalatest.ElasticDockerTestKit
+import app.softnetwork.elastic.sql.query.SelectStatement
+import app.softnetwork.persistence.generateUUID
+import com.typesafe.config.ConfigFactory
+import org.scalatest.flatspec.AnyFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.collection.immutable.ListMap
+import scala.concurrent.{Await, ExecutionContext}
+import scala.concurrent.duration._
+import scala.language.implicitConversions
+
+/** Result rows must never surface Elasticsearch hit metadata: `_index`, `_score` and `_sort` are
+  * gone for good, and `_id` only appears when `elastic.include-document-id` is enabled (disabled by
+  * default).
+  *
+  * `_id` is still carried internally through parsing — the ranking-window enrichment matches base
+  * rows to their per-partition ordinals by document id — so this spec also pins exact ROW_NUMBER
+  * ordinals on both the one-shot (LIMIT) and scroll-routed (no LIMIT, #209) paths to prove the
+  * egress strip does not starve that lookup.
+  */
+trait HitMetadataSpec extends AnyFlatSpecLike with ElasticDockerTestKit with Matchers {
+
+  lazy val log: Logger = LoggerFactory.getLogger(getClass.getName)
+
+  implicit val system: ActorSystem = ActorSystem(generateUUID())
+
+  implicit val context: ConversionContext = NativeContext
+
+  /** Both clients are instantiated straight from the SPI rather than through
+    * [[ElasticClientFactory]]: the factory caches clients per cluster URL, so a second `create`
+    * against the same cluster returns the first client regardless of configuration. The flag is
+    * pinned explicitly on BOTH clients so an ambient `ELASTIC_INCLUDE_DOCUMENT_ID` in the
+    * environment cannot flip the assertions.
+    */
+  private def spiClient(includeDocumentId: Boolean): ElasticClientApi =
+    java.util.ServiceLoader
+      .load(classOf[ElasticClientSpi])
+      .iterator()
+      .next()
+      .client(
+        ConfigFactory
+          .parseString(s"elastic.include-document-id = $includeDocumentId")
+          .withFallback(elasticConfig)
+      )
+
+  lazy val client: ElasticClientApi = spiClient(includeDocumentId = false)
+
+  lazy val clientWithDocumentId: ElasticClientApi = spiClient(includeDocumentId = true)
+
+  private val index = "hit_metadata"
+
+  private val forbiddenKeys = Set("_index", "_score", "_sort")
+
+  /** Category `cat_i` holds 4 docs with amounts i*100 + 1..4 — exact per-partition oracles. */
+  private val categories = 5
+
+  private val docsPerCategory = 4
+
+  private val totalDocs = categories * docsPerCategory
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+
+    val settings = """{"number_of_shards": 3, "number_of_replicas": 0}"""
+    val mapping =
+      """{
+        |  "properties": {
+        |    "id":       { "type": "keyword" },
+        |    "category": { "type": "keyword" },
+        |    "amount":   { "type": "integer" }
+        |  }
+        |}""".stripMargin
+
+    client.createIndex(index, settings = settings).get shouldBe true
+    client.setMapping(index, mapping).get shouldBe true
+
+    val docs = (for {
+      c <- 1 to categories
+      d <- 1 to docsPerCategory
+    } yield {
+      val category = f"cat_$c%02d"
+      s"""{"id":"${category}_$d","category":"$category","amount":${c * 100 + d}}"""
+    }).toList
+
+    implicit val bulkOptions: BulkOptions = BulkOptions(
+      defaultIndex = index,
+      logEvery = 1000
+    )
+
+    implicit def listToSource[T](list: List[T]): Source[T, NotUsed] =
+      Source.fromIterator(() => list.iterator)
+
+    client.bulk[String](docs, identity, idKey = Some(Set("id"))) match {
+      case ElasticSuccess(_) => // ok
+      case ElasticFailure(error) =>
+        error.cause.foreach(_.printStackTrace())
+        fail(s"Bulk indexing failed: ${error.message}")
+    }
+
+    client.refresh(index).get shouldBe true
+  }
+
+  override def afterAll(): Unit = {
+    client.deleteIndex(index)
+    system.terminate()
+    super.afterAll()
+  }
+
+  private def rowsOf(
+    api: ElasticClientApi,
+    sql: String
+  ): Seq[ListMap[String, Any]] =
+    api.search(SelectStatement(sql)) match {
+      case ElasticSuccess(response) => response.results
+      case ElasticFailure(error)    => fail(s"Query failed: ${error.message}")
+    }
+
+  "SELECT with LIMIT (one-shot path)" should "surface no hit metadata" in {
+    val rows = rowsOf(client, s"SELECT id, category, amount FROM $index ORDER BY amount LIMIT 5")
+    rows should have size 5
+    rows.foreach { row =>
+      row.keySet shouldBe Set("id", "category", "amount")
+    }
+  }
+
+  "SELECT without LIMIT (scroll-routed path)" should "surface no hit metadata" in {
+    val rows = rowsOf(client, s"SELECT id, category, amount FROM $index ORDER BY amount")
+    rows should have size totalDocs.toLong
+    rows.foreach { row =>
+      row.keySet shouldBe Set("id", "category", "amount")
+    }
+  }
+
+  "window-enriched SELECT" should "keep exact ordinals while surfacing no hit metadata" in {
+    // One-shot (LIMIT ≤ max_result_window) — enrichResponseWithWindowValues egress
+    val oneShot = rowsOf(
+      client,
+      s"""SELECT
+            category,
+            amount,
+            ROW_NUMBER() OVER (PARTITION BY category ORDER BY amount DESC) AS rnum
+          FROM $index LIMIT $totalDocs"""
+    )
+    // Scroll-routed (no LIMIT, #209) — scrollWithWindowEnrichment egress
+    val scrolled = rowsOf(
+      client,
+      s"""SELECT
+            category,
+            amount,
+            ROW_NUMBER() OVER (PARTITION BY category ORDER BY amount DESC) AS rnum
+          FROM $index"""
+    )
+
+    Seq("one-shot" -> oneShot, "scroll-routed" -> scrolled).foreach { case (path, rows) =>
+      withClue(s"$path path: ") {
+        rows should have size totalDocs.toLong
+        rows.foreach { row =>
+          row.keySet shouldBe Set("category", "amount", "rnum")
+        }
+        // The ordinal lookup keys on the internally carried `_id`: highest amount per
+        // category must rank 1, next 2, … — an off ordinal means the strip starved it.
+        rows.foreach { row =>
+          val amount = row("amount").toString.toInt
+          val expectedRank = docsPerCategory - (amount % 100) + 1
+          row("rnum").toString.toLong shouldBe expectedRank.toLong
+        }
+      }
+    }
+  }
+
+  "searchAsync" should "surface no hit metadata on the asynchronous path" in {
+    implicit val ec: ExecutionContext = system.dispatcher
+    val rows = Await.result(
+      client.searchAsync(SelectStatement(s"SELECT id, category FROM $index LIMIT 5")),
+      30.seconds
+    ) match {
+      case ElasticSuccess(response) => response.results
+      case ElasticFailure(error)    => fail(s"Query failed: ${error.message}")
+    }
+    rows should have size 5
+    rows.foreach { row =>
+      row.keySet shouldBe Set("id", "category")
+    }
+  }
+
+  "explicit SELECT _id" should "surface the document id even when disabled" in {
+    val rows = rowsOf(client, s"SELECT _id, id FROM $index LIMIT 5")
+    rows should have size 5
+    rows.foreach { row =>
+      row.keySet shouldBe Set("_id", "id")
+      row("_id") shouldBe row("id")
+    }
+  }
+
+  "UNION ALL" should "surface no hit metadata" in {
+    val rows = rowsOf(
+      client,
+      s"""SELECT id, category FROM $index WHERE category = 'cat_01'
+          UNION ALL
+          SELECT id, category FROM $index WHERE category = 'cat_02'"""
+    )
+    rows should have size (2L * docsPerCategory)
+    rows.foreach { row =>
+      row.keySet shouldBe Set("id", "category")
+    }
+  }
+
+  "GROUP BY aggregation" should "surface no hit metadata" in {
+    val rows = rowsOf(
+      client,
+      s"SELECT category, COUNT(*) AS cnt FROM $index GROUP BY category"
+    )
+    rows should have size categories.toLong
+    rows.foreach { row =>
+      (row.keySet & (forbiddenKeys + "_id")) shouldBe empty
+    }
+  }
+
+  "include-document-id = true" should "surface _id — and only _id — on every path" in {
+    // One-shot
+    val oneShot =
+      rowsOf(clientWithDocumentId, s"SELECT id, category FROM $index ORDER BY amount LIMIT 5")
+    oneShot should have size 5
+    // Scroll-routed
+    val scrolled = rowsOf(clientWithDocumentId, s"SELECT id, category FROM $index")
+    scrolled should have size totalDocs.toLong
+
+    (oneShot ++ scrolled).foreach { row =>
+      row.keySet shouldBe Set("id", "category", "_id")
+      // Documents are bulk-indexed with idKey = "id", so the surfaced `_id` must match it
+      row("_id") shouldBe row("id")
+    }
+  }
+}
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplIntegrationTestKit.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplIntegrationTestKit.scala
index 9e382174..966b0eb3 100644
--- a/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplIntegrationTestKit.scala
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/repl/ReplIntegrationTestKit.scala
@@ -129,7 +129,9 @@ trait ReplIntegrationTestKit
   // -------------------------------------------------------------------------
 
   private def normalizeRow(row: Map[String, Any]): Map[String, Any] = {
-    val updated = row - "_id" - "_index" - "_score" - "_version" - "_sort"
+    // `_id` only ever appears when `elastic.include-document-id` is enabled — strip it (and the
+    // get-api `_version`) so expected-row assertions stay independent of the client configuration.
+    val updated = row - "_id" - "_version"
     updated.map { entry =>
       entry._2 match {
         case m: Map[_, _] =>