Script: UpdateByQuery can read doc version if requested (#88740)

Allow UpdateByQuery to read the doc version if set in the request via
`version=true`.

If `version=true` is unset or false, the `ctx._version` is `-1`
indicating internal versioning via seq.

Fixes: #55745

Author: stu-elastic

docs/changelog/88740.yaml

@@ -0,0 +1,5 @@
+pr: 88740
+summary: "Script: `UpdateByQuery` can read doc version if requested"
+area: "Infra/Scripting"
+type: bug
+issues: []

modules/reindex/src/main/java/org/elasticsearch/reindex/TransportUpdateByQueryAction.java

@@ -100,8 +100,8 @@ static class AsyncIndexBySearchAction extends AbstractAsyncBulkByScrollAction<Up
         ) {
             super(
                 task,
-                // use sequence number powered optimistic concurrency control
-                false,
+                // use sequence number powered optimistic concurrency control unless requested
+                request.getSearchRequest().source() != null && Boolean.TRUE.equals(request.getSearchRequest().source().version()),
                 true,
                 logger,
                 client,

modules/reindex/src/test/java/org/elasticsearch/reindex/UpdateByQueryVersionTests.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.reindex;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.index.reindex.BulkByScrollResponse;
+import org.elasticsearch.index.reindex.ScrollableHitSource.Hit;
+import org.elasticsearch.index.reindex.UpdateByQueryRequest;
+
+public class UpdateByQueryVersionTests extends AbstractAsyncBulkByScrollActionMetadataTestCase<UpdateByQueryRequest, BulkByScrollResponse> {
+
+    UpdateByQueryRequest request;
+
+    public void testVersion() {
+        request = null;
+        assertFalse(action().mainRequest.getSearchRequest().source().version());
+
+        request = new UpdateByQueryRequest();
+        request.getSearchRequest().source().version(false);
+        assertFalse(action().mainRequest.getSearchRequest().source().version());
+
+        request = new UpdateByQueryRequest();
+        request.getSearchRequest().source().version(null);
+        assertFalse(action().mainRequest.getSearchRequest().source().version());
+
+        request = new UpdateByQueryRequest();
+        request.getSearchRequest().source().version(true);
+        assertTrue(action().mainRequest.getSearchRequest().source().version());
+    }
+
+    @Override
+    protected TestAction action() {
+        return new TestAction();
+    }
+
+    @Override
+    protected UpdateByQueryRequest request() {
+        return request != null ? request : new UpdateByQueryRequest();
+    }
+
+    private class TestAction extends TransportUpdateByQueryAction.AsyncIndexBySearchAction {
+        TestAction() {
+            super(
+                UpdateByQueryVersionTests.this.task,
+                UpdateByQueryVersionTests.this.logger,
+                null,
+                UpdateByQueryVersionTests.this.threadPool,
+                null,
+                request(),
+                ClusterState.EMPTY_STATE,
+                listener()
+            );
+        }
+
+        @Override
+        public RequestWrapper<?> copyMetadata(RequestWrapper<?> request, Hit doc) {
+            return super.copyMetadata(request, doc);
+        }
+    }
+}

modules/reindex/src/yamlRestTest/resources/rest-api-spec/test/update_by_query/80_scripting.yml

@@ -432,3 +432,70 @@
             lang: painless
             source: syntax errors are fun!
   - match: {error.reason: 'compile error'}
+
+---
+"Can read version":
+  - skip:
+      version: " - 8.3.99"
+      reason: fixed in 8.4.0
+
+  - do:
+      index:
+        index:  twitter
+        id:     "1"
+        body:   { "user": "kimchy" }
+        version: 500
+        version_type: "external"
+  - do:
+      indices.refresh: {}
+
+  - do:
+      update_by_query:
+        refresh: true
+        index:   twitter
+        version: true
+        body:
+          script:
+            lang: painless
+            source: |
+              ctx._source.old_version = ctx._version;
+
+  - match: {updated: 1}
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: twitter
+        version: true
+        body:
+          query:
+            match_all: {}
+
+  - match: { hits.total: 1 }
+  - match: { hits.hits.0._source.old_version: 500 }
+
+  - do:
+      update_by_query:
+        refresh: true
+        index:   twitter
+        # no version
+        body:
+          script:
+            lang: painless
+            source: |
+              ctx._source.old_version = ctx._version;
+
+  - match: {updated: 1}
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: twitter
+        version: true
+        body:
+          query:
+            match_all: {}
+
+  - match: { hits.total: 1 }
+  - match: { hits.hits.0._source.old_version: -1 }
+