Merge branch 'main' into esql/settings_stack_telemetry

Author: luigidellaquila

docs/changelog/141570.yaml

@@ -0,0 +1,5 @@
+area: "Authorization"
+issues: []
+pr: 141570
+summary: Update View CRUD Actions to be Index Actions
+type: enhancement

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

@@ -35,10 +35,10 @@
 import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
-import org.elasticsearch.index.reindex.ClientScrollableHitSource;
+import org.elasticsearch.index.reindex.ClientScrollablePaginatedHitSource;
+import org.elasticsearch.index.reindex.PaginatedHitSource;
+import org.elasticsearch.index.reindex.PaginatedHitSource.SearchFailure;
 import org.elasticsearch.index.reindex.ResumeInfo.WorkerResumeInfo;
-import org.elasticsearch.index.reindex.ScrollableHitSource;
-import org.elasticsearch.index.reindex.ScrollableHitSource.SearchFailure;
 import org.elasticsearch.index.reindex.WorkerBulkByScrollTaskState;
 import org.elasticsearch.script.CtxMap;
 import org.elasticsearch.script.Metadata;
@@ -99,14 +99,14 @@ public abstract class AbstractAsyncBulkByScrollAction<
     private final ParentTaskAssigningClient bulkClient;
     private final ActionListener<BulkByScrollResponse> listener;
     private final Retry bulkRetry;
-    private final ScrollableHitSource scrollSource;
+    private final PaginatedHitSource paginatedHitSource;
 
     /**
      * This BiFunction is used to apply various changes depending of the Reindex action and  the search hit,
      * from copying search hit metadata (parent, routing, etc) to potentially transforming the
      * {@link RequestWrapper} completely.
      */
-    private final BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> scriptApplier;
+    private final BiFunction<RequestWrapper<?>, PaginatedHitSource.Hit, RequestWrapper<?>> scriptApplier;
     private int lastBatchSize;
     /**
      * Keeps track of the total number of bulk operations performed
@@ -176,15 +176,15 @@ public abstract class AbstractAsyncBulkByScrollAction<
         this.listener = listener;
         BackoffPolicy backoffPolicy = buildBackoffPolicy();
         bulkRetry = new Retry(BackoffPolicy.wrap(backoffPolicy, worker::countBulkRetry), threadPool);
-        scrollSource = buildScrollableResultSource(
+        paginatedHitSource = buildScrollableResultSource(
             backoffPolicy,
             prepareSearchRequest(mainRequest, needsSourceDocumentVersions, needsSourceDocumentSeqNoAndPrimaryTerm, needsVectors)
         );
         scriptApplier = Objects.requireNonNull(buildScriptApplier(), "script applier must not be null");
     }
 
     /**
-     * Prepares a search request to be used in a ScrollableHitSource.
+     * Prepares a search request to be used in a {@link PaginatedHitSource}.
      * Preparation might set a sort order (if not set already) and disable scroll if max docs is small enough.
      */
     // Visible for testing
@@ -238,7 +238,7 @@ static <Request extends AbstractBulkByScrollRequest<Request>> SearchRequest prep
      *
      * Public for testings....
      */
-    public BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> buildScriptApplier() {
+    public BiFunction<RequestWrapper<?>, PaginatedHitSource.Hit, RequestWrapper<?>> buildScriptApplier() {
         // The default script applier executes a no-op
         return (request, searchHit) -> request;
     }
@@ -248,12 +248,12 @@ public BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>>
      * metadata or scripting. That will be handled by copyMetadata and
      * apply functions that can be overridden.
      */
-    protected abstract RequestWrapper<?> buildRequest(ScrollableHitSource.Hit doc);
+    protected abstract RequestWrapper<?> buildRequest(PaginatedHitSource.Hit doc);
 
     /**
      * Copies the metadata from a hit to the request.
      */
-    protected RequestWrapper<?> copyMetadata(RequestWrapper<?> request, ScrollableHitSource.Hit doc) {
+    protected RequestWrapper<?> copyMetadata(RequestWrapper<?> request, PaginatedHitSource.Hit doc) {
         copyRouting(request, doc.getRouting());
         return request;
     }
@@ -270,7 +270,7 @@ protected void copyRouting(RequestWrapper<?> request, String routing) {
      * from the bulk request. It is also where we fail on invalid search hits, like
      * when the document has no source but it's required.
      */
-    protected boolean accept(ScrollableHitSource.Hit doc) {
+    protected boolean accept(PaginatedHitSource.Hit doc) {
         if (doc.getSource() == null) {
             /*
              * Either the document didn't store _source or we didn't fetch it for some reason. Since we don't allow the user to
@@ -282,9 +282,9 @@ protected boolean accept(ScrollableHitSource.Hit doc) {
         return true;
     }
 
-    protected BulkRequest buildBulk(Iterable<? extends ScrollableHitSource.Hit> docs) {
+    protected BulkRequest buildBulk(Iterable<? extends PaginatedHitSource.Hit> docs) {
         BulkRequest bulkRequest = new BulkRequest();
-        for (ScrollableHitSource.Hit doc : docs) {
+        for (PaginatedHitSource.Hit doc : docs) {
             if (accept(doc)) {
                 RequestWrapper<?> request = scriptApplier.apply(copyMetadata(buildRequest(doc), doc), doc);
                 if (request != null) {
@@ -295,8 +295,8 @@ protected BulkRequest buildBulk(Iterable<? extends ScrollableHitSource.Hit> docs
         return bulkRequest;
     }
 
-    protected ScrollableHitSource buildScrollableResultSource(BackoffPolicy backoffPolicy, SearchRequest searchRequest) {
-        return new ClientScrollableHitSource(
+    protected PaginatedHitSource buildScrollableResultSource(BackoffPolicy backoffPolicy, SearchRequest searchRequest) {
+        return new ClientScrollablePaginatedHitSource(
             logger,
             backoffPolicy,
             threadPool,
@@ -338,17 +338,17 @@ public void start() {
                 WorkerResumeInfo workerResumeInfo = resumeInfo.getWorker().get();
                 startTime.set(workerResumeInfo.startTime());
                 worker.restoreState(workerResumeInfo.status());
-                scrollSource.resume(workerResumeInfo);
+                paginatedHitSource.resume(workerResumeInfo);
             } else {
                 startTime.set(System.nanoTime());
-                scrollSource.start();
+                paginatedHitSource.start();
             }
         } catch (Exception e) {
             finishHim(e);
         }
     }
 
-    void onScrollResponse(ScrollableHitSource.AsyncResponse asyncResponse) {
+    void onScrollResponse(PaginatedHitSource.AsyncResponse asyncResponse) {
         onScrollResponse(new ScrollConsumableHitsResponse(asyncResponse));
     }
 
@@ -362,10 +362,10 @@ void onScrollResponse(ScrollConsumableHitsResponse asyncResponse) {
      * Process a scroll response.
      * @param lastBatchStartTimeNS the time when the last batch started. Used to calculate the throttling delay.
      * @param lastBatchSizeToUse the size of the last batch. Used to calculate the throttling delay.
-     * @param asyncResponse the response to process from ScrollableHitSource
+     * @param asyncResponse the response to process from {@link PaginatedHitSource}
      */
     void onScrollResponse(long lastBatchStartTimeNS, int lastBatchSizeToUse, ScrollConsumableHitsResponse asyncResponse) {
-        ScrollableHitSource.Response response = asyncResponse.response();
+        PaginatedHitSource.Response response = asyncResponse.response();
         logger.debug("[{}]: got scroll response with [{}] hits", task.getId(), asyncResponse.remainingHits());
         if (task.isCancelled()) {
             logger.debug("[{}]: finishing early because the task was cancelled", task.getId());
@@ -420,7 +420,7 @@ void prepareBulkRequest(long thisBatchStartTimeNS, ScrollConsumableHitsResponse
             return;
         }
         worker.countBatch();
-        final List<? extends ScrollableHitSource.Hit> hits;
+        final List<? extends PaginatedHitSource.Hit> hits;
 
         if (mainRequest.getMaxDocs() != MAX_DOCS_ALL_MATCHES) {
             // Truncate the hits if we have more than the request max docs
@@ -528,7 +528,7 @@ void onBulkResponse(BulkResponse response, Runnable onSuccess) {
                 return;
             }
 
-            if (scrollSource.hasScroll() == false) {
+            if (paginatedHitSource.hasScroll() == false) {
                 // Index contains fewer matching docs than max_docs (found < max_docs <= scroll size)
                 refreshAndFinish(emptyList(), emptyList(), false);
                 return;
@@ -611,7 +611,7 @@ protected void finishHim(Exception failure) {
      */
     protected void finishHim(Exception failure, List<Failure> indexingFailures, List<SearchFailure> searchFailures, boolean timedOut) {
         logger.debug("[{}]: finishing without any catastrophic failures", task.getId());
-        scrollSource.close(threadPool.getThreadContext().preserveContext(() -> {
+        paginatedHitSource.close(threadPool.getThreadContext().preserveContext(() -> {
             if (failure == null) {
                 BulkByScrollResponse response = buildResponse(
                     timeValueNanos(System.nanoTime() - startTime.get()),
@@ -645,7 +645,7 @@ void addDestinationIndices(Collection<String> indices) {
      * Set the last returned scrollId. Exists entirely for testing.
      */
     void setScroll(String scroll) {
-        scrollSource.setScroll(scroll);
+        paginatedHitSource.setScroll(scroll);
     }
 
     /**
@@ -841,7 +841,7 @@ public static RequestWrapper<DeleteRequest> wrap(DeleteRequest request) {
      */
     public abstract static class ScriptApplier<T extends Metadata>
         implements
-            BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> {
+            BiFunction<RequestWrapper<?>, PaginatedHitSource.Hit, RequestWrapper<?>> {
 
         // "index" is the default operation
         protected static final String INDEX = "index";
@@ -867,7 +867,7 @@ public ScriptApplier(
         }
 
         @Override
-        public RequestWrapper<?> apply(RequestWrapper<?> request, ScrollableHitSource.Hit doc) {
+        public RequestWrapper<?> apply(RequestWrapper<?> request, PaginatedHitSource.Hit doc) {
             if (script == null) {
                 return request;
             }
@@ -883,7 +883,7 @@ public RequestWrapper<?> apply(RequestWrapper<?> request, ScrollableHitSource.Hi
             return requestFromOp(request, metadata.getOp());
         }
 
-        protected abstract CtxMap<T> execute(ScrollableHitSource.Hit doc, Map<String, Object> source);
+        protected abstract CtxMap<T> execute(PaginatedHitSource.Hit doc, Map<String, Object> source);
 
         protected abstract void updateRequest(RequestWrapper<?> request, T metadata);
 
@@ -909,24 +909,24 @@ protected RequestWrapper<?> requestFromOp(RequestWrapper<?> request, String op)
     }
 
     static class ScrollConsumableHitsResponse {
-        private final ScrollableHitSource.AsyncResponse asyncResponse;
-        private final List<? extends ScrollableHitSource.Hit> hits;
+        private final PaginatedHitSource.AsyncResponse asyncResponse;
+        private final List<? extends PaginatedHitSource.Hit> hits;
         private int consumedOffset = 0;
 
-        ScrollConsumableHitsResponse(ScrollableHitSource.AsyncResponse asyncResponse) {
+        ScrollConsumableHitsResponse(PaginatedHitSource.AsyncResponse asyncResponse) {
             this.asyncResponse = asyncResponse;
             this.hits = asyncResponse.response().getHits();
         }
 
-        ScrollableHitSource.Response response() {
+        PaginatedHitSource.Response response() {
             return asyncResponse.response();
         }
 
-        List<? extends ScrollableHitSource.Hit> consumeRemainingHits() {
+        List<? extends PaginatedHitSource.Hit> consumeRemainingHits() {
             return consumeHits(remainingHits());
         }
 
-        List<? extends ScrollableHitSource.Hit> consumeHits(int numberOfHits) {
+        List<? extends PaginatedHitSource.Hit> consumeHits(int numberOfHits) {
             if (numberOfHits < 0) {
                 throw new IllegalArgumentException("Invalid number of hits to consume [" + numberOfHits + "]");
             }

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

@@ -16,7 +16,7 @@
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
-import org.elasticsearch.index.reindex.ScrollableHitSource;
+import org.elasticsearch.index.reindex.PaginatedHitSource;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -38,14 +38,14 @@ public AsyncDeleteByQueryAction(
     }
 
     @Override
-    protected boolean accept(ScrollableHitSource.Hit doc) {
+    protected boolean accept(PaginatedHitSource.Hit doc) {
         // Delete-by-query does not require the source to delete a document
         // and the default implementation checks for it
         return true;
     }
 
     @Override
-    protected RequestWrapper<DeleteRequest> buildRequest(ScrollableHitSource.Hit doc) {
+    protected RequestWrapper<DeleteRequest> buildRequest(PaginatedHitSource.Hit doc) {
         DeleteRequest delete = new DeleteRequest();
         delete.index(doc.getIndex());
         delete.id(doc.getId());
@@ -59,7 +59,7 @@ protected RequestWrapper<DeleteRequest> buildRequest(ScrollableHitSource.Hit doc
      * don't care for a deletion.
      */
     @Override
-    protected RequestWrapper<?> copyMetadata(RequestWrapper<?> request, ScrollableHitSource.Hit doc) {
+    protected RequestWrapper<?> copyMetadata(RequestWrapper<?> request, PaginatedHitSource.Hit doc) {
         request.setRouting(doc.getRouting());
         return request;
     }

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

@@ -11,7 +11,7 @@
 
 import org.elasticsearch.action.bulk.BulkItemResponse.Failure;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
-import org.elasticsearch.index.reindex.ScrollableHitSource.SearchFailure;
+import org.elasticsearch.index.reindex.PaginatedHitSource.SearchFailure;
 import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;

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

@@ -47,12 +47,12 @@
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
+import org.elasticsearch.index.reindex.PaginatedHitSource;
 import org.elasticsearch.index.reindex.ReindexAction;
 import org.elasticsearch.index.reindex.ReindexRequest;
 import org.elasticsearch.index.reindex.RemoteInfo;
-import org.elasticsearch.index.reindex.ScrollableHitSource;
 import org.elasticsearch.index.reindex.WorkerBulkByScrollTaskState;
-import org.elasticsearch.reindex.remote.RemoteScrollableHitSource;
+import org.elasticsearch.reindex.remote.RemoteScrollablePaginatedHitSource;
 import org.elasticsearch.script.CtxMap;
 import org.elasticsearch.script.ReindexMetadata;
 import org.elasticsearch.script.ReindexScript;
@@ -161,7 +161,7 @@ public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
                 var searchExceptionSample = Optional.ofNullable(bulkByScrollResponse.getSearchFailures())
                     .stream()
                     .flatMap(List::stream)
-                    .map(ScrollableHitSource.SearchFailure::getReason)
+                    .map(PaginatedHitSource.SearchFailure::getReason)
                     .findFirst();
                 var bulkExceptionSample = Optional.ofNullable(bulkByScrollResponse.getBulkFailures())
                     .stream()
@@ -312,13 +312,13 @@ private IndexMode destinationIndexMode(ProjectState state) {
         }
 
         @Override
-        protected ScrollableHitSource buildScrollableResultSource(BackoffPolicy backoffPolicy, SearchRequest searchRequest) {
+        protected PaginatedHitSource buildScrollableResultSource(BackoffPolicy backoffPolicy, SearchRequest searchRequest) {
             if (mainRequest.getRemoteInfo() != null) {
                 RemoteInfo remoteInfo = mainRequest.getRemoteInfo();
                 createdThreads = synchronizedList(new ArrayList<>());
                 assert sslConfig != null : "Reindex ssl config must be set";
                 RestClient restClient = buildRestClient(remoteInfo, sslConfig, task.getId(), createdThreads);
-                return new RemoteScrollableHitSource(
+                return new RemoteScrollablePaginatedHitSource(
                     logger,
                     backoffPolicy,
                     threadPool,
@@ -337,7 +337,7 @@ protected ScrollableHitSource buildScrollableResultSource(BackoffPolicy backoffP
         protected void finishHim(
             Exception failure,
             List<BulkItemResponse.Failure> indexingFailures,
-            List<ScrollableHitSource.SearchFailure> searchFailures,
+            List<PaginatedHitSource.SearchFailure> searchFailures,
             boolean timedOut
         ) {
             super.finishHim(failure, indexingFailures, searchFailures, timedOut);
@@ -351,7 +351,7 @@ protected void finishHim(
         }
 
         @Override
-        public BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>> buildScriptApplier() {
+        public BiFunction<RequestWrapper<?>, PaginatedHitSource.Hit, RequestWrapper<?>> buildScriptApplier() {
             Script script = mainRequest.getScript();
             if (script != null) {
                 assert scriptService != null : "Script service must be set";
@@ -361,7 +361,7 @@ public BiFunction<RequestWrapper<?>, ScrollableHitSource.Hit, RequestWrapper<?>>
         }
 
         @Override
-        protected RequestWrapper<IndexRequest> buildRequest(ScrollableHitSource.Hit doc) {
+        protected RequestWrapper<IndexRequest> buildRequest(PaginatedHitSource.Hit doc) {
             IndexRequest index = new IndexRequest();
 
             // Copy the index from the request so we always write where it asked to write
@@ -456,7 +456,7 @@ static class ReindexScriptApplier extends ScriptApplier<ReindexMetadata> {
             }
 
             @Override
-            protected CtxMap<ReindexMetadata> execute(ScrollableHitSource.Hit doc, Map<String, Object> source) {
+            protected CtxMap<ReindexMetadata> execute(PaginatedHitSource.Hit doc, Map<String, Object> source) {
                 if (reindex == null) {
                     reindex = scriptService.compile(script, ReindexScript.CONTEXT);
                 }