Initial implementation of random sampling (elastic#135660)

Author: masseyke

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -52,7 +52,6 @@
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.TriConsumer;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
@@ -1391,45 +1390,11 @@ private void attemptToSampleData(
              * We need both the original document and the fully updated document for sampling, so we make a copy of the original
              * before overwriting it here. We can discard it after sampling.
              */
-            samplingService.maybeSample(projectMetadata, indexRequest.index(), () -> {
-                IndexRequest original = copyIndexRequestForSampling(indexRequest);
-                updateIndexRequestMetadata(original, originalDocumentMetadata);
-                return original;
-            }, ingestDocument);
+            samplingService.maybeSample(projectMetadata, originalDocumentMetadata.getIndex(), indexRequest, ingestDocument);
 
         }
     }
 
-    /**
-     * Creates a copy of an IndexRequest to be used by random sampling.
-     * @param original The IndexRequest to be copied
-     * @return A copy of the IndexRequest
-     */
-    private IndexRequest copyIndexRequestForSampling(IndexRequest original) {
-        IndexRequest clonedRequest = new IndexRequest(original.index());
-        clonedRequest.id(original.id());
-        clonedRequest.routing(original.routing());
-        clonedRequest.version(original.version());
-        clonedRequest.versionType(original.versionType());
-        clonedRequest.setPipeline(original.getPipeline());
-        clonedRequest.setFinalPipeline(original.getFinalPipeline());
-        clonedRequest.setIfSeqNo(original.ifSeqNo());
-        clonedRequest.setIfPrimaryTerm(original.ifPrimaryTerm());
-        clonedRequest.setRefreshPolicy(original.getRefreshPolicy());
-        clonedRequest.waitForActiveShards(original.waitForActiveShards());
-        clonedRequest.timeout(original.timeout());
-        clonedRequest.opType(original.opType());
-        clonedRequest.setParentTask(original.getParentTask());
-        clonedRequest.setRequireDataStream(original.isRequireDataStream());
-        clonedRequest.setRequireAlias(original.isRequireAlias());
-        clonedRequest.setIncludeSourceOnError(original.getIncludeSourceOnError());
-        BytesReference source = original.source();
-        if (source != null) {
-            clonedRequest.source(source, original.getContentType());
-        }
-        return clonedRequest;
-    }
-
     private static void executePipeline(
         final IngestDocument ingestDocument,
         final Pipeline pipeline,