Merge branch 'main' into ia-enable-ccm-by-default

Author: jonathan-buttner

docs/changelog/139244.yaml

@@ -0,0 +1,6 @@
+pr: 139244
+summary: "Inference command: support for CCS"
+area: ES|QL
+type: enhancement
+issues:
+ - 136860

docs/reference/query-languages/esql/_snippets/functions/examples/text_embedding.md

@@ -2,11 +2,11 @@
 * [`AVG_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-avg_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`COUNT_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-count_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`COUNT_DISTINCT_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-count_distinct_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
-* [`DELTA`](../../functions-operators/time-series-aggregation-functions.md#esql-rate) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
+* [`DELTA`](../../functions-operators/time-series-aggregation-functions.md#esql-delta) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`FIRST_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-first_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
-* [`IDELTA`](../../functions-operators/time-series-aggregation-functions.md#esql-rate) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
-* [`INCREASE`](../../functions-operators/time-series-aggregation-functions.md#esql-rate) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
-* [`IRATE`](../../functions-operators/time-series-aggregation-functions.md#esql-rate) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
+* [`IDELTA`](../../functions-operators/time-series-aggregation-functions.md#esql-idelta) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
+* [`INCREASE`](../../functions-operators/time-series-aggregation-functions.md#esql-increase) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
+* [`IRATE`](../../functions-operators/time-series-aggregation-functions.md#esql-irate) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`LAST_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-last_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`MAX_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-max_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`
 * [`MIN_OVER_TIME`](../../functions-operators/time-series-aggregation-functions.md#esql-min_over_time) {applies_to}`stack: preview 9.2` {applies_to}`serverless: preview`

docs/reference/query-languages/esql/_snippets/lists/time-series-aggregation-functions.md

@@ -10,7 +10,6 @@
 package org.elasticsearch.ingest.geoip;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -44,10 +43,6 @@
 
 public class GeoIpTaskState implements PersistentTaskState, VersionedNamedWriteable {
 
-    private static boolean includeSha256(TransportVersion version) {
-        return version.onOrAfter(TransportVersions.V_8_15_0);
-    }
-
     private static final ParseField DATABASES = new ParseField("databases");
 
     static final GeoIpTaskState EMPTY = new GeoIpTaskState(Map.of());
@@ -78,14 +73,7 @@ public static GeoIpTaskState fromXContent(XContentParser parser) throws IOExcept
 
     GeoIpTaskState(StreamInput input) throws IOException {
         databases = input.readImmutableMap(
-            in -> new Metadata(
-                in.readLong(),
-                in.readVInt(),
-                in.readVInt(),
-                in.readString(),
-                in.readLong(),
-                includeSha256(in.getTransportVersion()) ? input.readOptionalString() : null
-            )
+            in -> new Metadata(in.readLong(), in.readVInt(), in.readVInt(), in.readString(), in.readLong(), input.readOptionalString())
         );
     }
 
@@ -144,9 +132,7 @@ public void writeTo(StreamOutput out) throws IOException {
             o.writeVInt(v.lastChunk);
             o.writeString(v.md5);
             o.writeLong(v.lastCheck);
-            if (includeSha256(o.getTransportVersion())) {
-                o.writeOptionalString(v.sha256);
-            }
+            o.writeOptionalString(v.sha256);
         });
     }
 

docs/reference/query-languages/esql/kibana/definition/functions/text_embedding.json

@@ -32,6 +32,7 @@
 import org.elasticsearch.index.mapper.TestDocumentParserContext;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.query.TermQueryBuilder;
+import org.elasticsearch.script.field.BinaryDocValuesField;
 import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
 import org.elasticsearch.test.ESTestCase;
@@ -88,7 +89,7 @@ public void testStoringQueryBuilders() throws IOException {
             when(searchExecutionContext.getWriteableRegistry()).thenReturn(writableRegistry());
             when(searchExecutionContext.getParserConfig()).thenReturn(parserConfig());
             when(searchExecutionContext.getForField(fieldMapper.fieldType(), fielddataOperation)).thenReturn(
-                new BytesBinaryIndexFieldData(fieldMapper.fullPath(), CoreValuesSourceType.KEYWORD)
+                new BytesBinaryIndexFieldData(fieldMapper.fullPath(), CoreValuesSourceType.KEYWORD, BinaryDocValuesField::new)
             );
             when(searchExecutionContext.getFieldType(Mockito.anyString())).thenAnswer(invocation -> {
                 final String fieldName = (String) invocation.getArguments()[0];

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpTaskState.java

@@ -84,6 +84,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.blobstore.support.BlobContainerUtils.getRegisterUsingConsistentRead;
@@ -144,10 +145,11 @@ public void writeBlob(OperationPurpose purpose, String blobName, InputStream inp
         throws IOException {
         assert BlobContainer.assertPurposeConsistency(purpose, blobName);
         assert inputStream.markSupported() : "No mark support on inputStream breaks the S3 SDK's ability to retry requests";
+        final var condition = failIfAlreadyExists ? ConditionalOperation.IF_NONE_MATCH : ConditionalOperation.NONE;
         if (blobSize <= getLargeBlobThresholdInBytes()) {
-            executeSingleUpload(purpose, blobStore, buildKey(blobName), inputStream, blobSize, failIfAlreadyExists);
+            executeSingleUpload(purpose, blobStore, buildKey(blobName), inputStream, blobSize, condition);
         } else {
-            executeMultipartUpload(purpose, blobStore, buildKey(blobName), inputStream, blobSize, failIfAlreadyExists);
+            executeMultipartUpload(purpose, blobStore, buildKey(blobName), inputStream, blobSize, condition);
         }
     }
 
@@ -536,6 +538,59 @@ String buildKey(String blobName) {
         return keyPath + blobName;
     }
 
+    /**
+     * Enumeration of mutually exlusive conditional operations supported by S3.
+     *
+     * @see <a href=https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-requests.html>S3-conditional-requests</a>
+     */
+    sealed interface ConditionalOperation permits ConditionalOperation.IfMatch, ConditionalOperation.IfNoneMatch,
+        ConditionalOperation.None {
+        ConditionalOperation NONE = new None();
+        ConditionalOperation IF_NONE_MATCH = new IfNoneMatch();
+
+        static ConditionalOperation ifMatch(String etag) {
+            return new IfMatch(etag);
+        }
+
+        record None() implements ConditionalOperation {}
+
+        record IfNoneMatch() implements ConditionalOperation {}
+
+        record IfMatch(String etag) implements ConditionalOperation {}
+    }
+
+    static void putObject(
+        OperationPurpose purpose,
+        S3BlobStore s3BlobStore,
+        String blobName,
+        long contentLength,
+        Supplier<RequestBody> body,
+        ConditionalOperation condition
+    ) {
+        final var putRequestBuilder = PutObjectRequest.builder()
+            .bucket(s3BlobStore.bucket())
+            .key(blobName)
+            .contentLength(contentLength)
+            .storageClass(s3BlobStore.getStorageClass())
+            .acl(s3BlobStore.getCannedACL());
+        if (s3BlobStore.serverSideEncryption()) {
+            putRequestBuilder.serverSideEncryption(ServerSideEncryption.AES256);
+        }
+        if (s3BlobStore.supportsConditionalWrites(purpose)) {
+            switch (condition) {
+                case ConditionalOperation.IfMatch ifMatch -> putRequestBuilder.ifMatch(ifMatch.etag);
+                case ConditionalOperation.IfNoneMatch ignored -> putRequestBuilder.ifNoneMatch("*");
+                case ConditionalOperation.None ignored -> {
+                }
+            }
+        }
+        S3BlobStore.configureRequestForMetrics(putRequestBuilder, s3BlobStore, Operation.PUT_OBJECT, purpose);
+        final var putRequest = putRequestBuilder.build();
+        try (var client = s3BlobStore.clientReference()) {
+            client.client().putObject(putRequest, body.get());
+        }
+    }
+
     /**
      * Uploads a blob using a single upload request
      */
@@ -545,33 +600,17 @@ void executeSingleUpload(
         final String blobName,
         final InputStream input,
         final long blobSize,
-        final boolean failIfAlreadyExists
+        final ConditionalOperation condition
     ) throws IOException {
-        try (var clientReference = s3BlobStore.clientReference()) {
+        try {
             // Extra safety checks
             if (blobSize > MAX_FILE_SIZE.getBytes()) {
                 throw new IllegalArgumentException("Upload request size [" + blobSize + "] can't be larger than " + MAX_FILE_SIZE);
             }
             if (blobSize > s3BlobStore.bufferSizeInBytes()) {
                 throw new IllegalArgumentException("Upload request size [" + blobSize + "] can't be larger than buffer size");
             }
-
-            final var putRequestBuilder = PutObjectRequest.builder()
-                .bucket(s3BlobStore.bucket())
-                .key(blobName)
-                .contentLength(blobSize)
-                .storageClass(s3BlobStore.getStorageClass())
-                .acl(s3BlobStore.getCannedACL());
-            if (s3BlobStore.serverSideEncryption()) {
-                putRequestBuilder.serverSideEncryption(ServerSideEncryption.AES256);
-            }
-            if (failIfAlreadyExists && s3BlobStore.supportsConditionalWrites(purpose)) {
-                putRequestBuilder.ifNoneMatch("*");
-            }
-            S3BlobStore.configureRequestForMetrics(putRequestBuilder, blobStore, Operation.PUT_OBJECT, purpose);
-
-            final var putRequest = putRequestBuilder.build();
-            clientReference.client().putObject(putRequest, RequestBody.fromInputStream(input, blobSize));
+            putObject(purpose, s3BlobStore, blobName, blobSize, () -> RequestBody.fromInputStream(input, blobSize), condition);
         } catch (final SdkException e) {
             throw new IOException("Unable to upload object [" + blobName + "] using a single upload", e);
         }
@@ -590,7 +629,7 @@ private void executeMultipart(
         final long partSize,
         final long blobSize,
         final PartOperation partOperation,
-        final boolean failIfAlreadyExists
+        final ConditionalOperation condition
     ) throws IOException {
 
         ensureMultiPartUploadSize(blobSize);
@@ -661,8 +700,13 @@ private void executeMultipart(
                 .uploadId(uploadId)
                 .multipartUpload(b -> b.parts(parts));
 
-            if (failIfAlreadyExists && s3BlobStore.supportsConditionalWrites(purpose)) {
-                completeMultipartUploadRequestBuilder.ifNoneMatch("*");
+            if (s3BlobStore.supportsConditionalWrites(purpose)) {
+                switch (condition) {
+                    case ConditionalOperation.IfMatch ifMatch -> completeMultipartUploadRequestBuilder.ifMatch(ifMatch.etag);
+                    case ConditionalOperation.IfNoneMatch ignored -> completeMultipartUploadRequestBuilder.ifNoneMatch("*");
+                    case ConditionalOperation.None ignored -> {
+                    }
+                }
             }
 
             S3BlobStore.configureRequestForMetrics(completeMultipartUploadRequestBuilder, blobStore, operation, purpose);
@@ -690,7 +734,7 @@ void executeMultipartUpload(
         final String blobName,
         final InputStream input,
         final long blobSize,
-        final boolean failIfAlreadyExists
+        final ConditionalOperation condition
     ) throws IOException {
         executeMultipart(
             purpose,
@@ -708,7 +752,7 @@ void executeMultipartUpload(
                     return CompletedPart.builder().partNumber(partNum).eTag(uploadResponse.eTag()).build();
                 }
             },
-            failIfAlreadyExists
+            condition
         );
     }
 
@@ -756,7 +800,7 @@ void executeMultipartCopy(
                     return CompletedPart.builder().partNumber(partNum).eTag(uploadPartCopyResponse.copyPartResult().eTag()).build();
                 }
             }),
-            false
+            ConditionalOperation.NONE
         );
     }