[feature/semantic-text] Simplify the integration of the field inference metadata in IndexMetadata

server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java

@@ -209,8 +209,8 @@ private void executeBulkRequestsByShard(Map<ShardId, List<BulkItemRequest>> requ
                     requests.toArray(new BulkItemRequest[0])
                 );
                 var indexMetadata = clusterState.getMetadata().index(shardId.getIndexName());
-                if (indexMetadata != null && indexMetadata.getFieldInferenceMetadata().isEmpty() == false) {
-                    bulkShardRequest.setFieldInferenceMetadata(indexMetadata.getFieldInferenceMetadata());
+                if (indexMetadata != null && indexMetadata.getInferenceFields().isEmpty() == false) {
+                    bulkShardRequest.setInferenceFieldMap(indexMetadata.getInferenceFields());
                 }
                 bulkShardRequest.waitForActiveShards(bulkRequest.waitForActiveShards());
                 bulkShardRequest.timeout(bulkRequest.timeout());

server/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java

@@ -15,14 +15,15 @@
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
 import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.cluster.metadata.FieldInferenceMetadata;
+import org.elasticsearch.cluster.metadata.InferenceFieldMetadata;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.transport.RawIndexingDataTransportRequest;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.Set;
 
 public final class BulkShardRequest extends ReplicatedWriteRequest<BulkShardRequest>
@@ -34,7 +35,7 @@ public final class BulkShardRequest extends ReplicatedWriteRequest<BulkShardRequ
 
     private final BulkItemRequest[] items;
 
-    private transient FieldInferenceMetadata fieldsInferenceMetadataMap = null;
+    private transient Map<String, InferenceFieldMetadata> inferenceFieldMap = null;
 
     public BulkShardRequest(StreamInput in) throws IOException {
         super(in);
@@ -51,24 +52,24 @@ public BulkShardRequest(ShardId shardId, RefreshPolicy refreshPolicy, BulkItemRe
      * Public for test
      * Set the transient metadata indicating that this request requires running inference before proceeding.
      */
-    public void setFieldInferenceMetadata(FieldInferenceMetadata fieldsInferenceMetadata) {
-        this.fieldsInferenceMetadataMap = fieldsInferenceMetadata;
+    public void setInferenceFieldMap(Map<String, InferenceFieldMetadata> fieldInferenceMap) {
+        this.inferenceFieldMap = fieldInferenceMap;
     }
 
     /**
      * Consumes the inference metadata to execute inference on the bulk items just once.
      */
-    public FieldInferenceMetadata consumeFieldInferenceMetadata() {
-        FieldInferenceMetadata ret = fieldsInferenceMetadataMap;
-        fieldsInferenceMetadataMap = null;
+    public Map<String, InferenceFieldMetadata> consumeInferenceFieldMap() {
+        Map<String, InferenceFieldMetadata> ret = inferenceFieldMap;
+        inferenceFieldMap = null;
         return ret;
     }
 
     /**
      * Public for test
      */
-    public FieldInferenceMetadata getFieldsInferenceMetadataMap() {
-        return fieldsInferenceMetadataMap;
+    public Map<String, InferenceFieldMetadata> getInferenceFieldMap() {
+        return inferenceFieldMap;
     }
 
     public long totalSizeInBytes() {

server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java

@@ -40,6 +40,7 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
+import org.elasticsearch.index.mapper.InferenceFieldMapper;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
@@ -184,7 +185,7 @@ protected void shardOperation(final UpdateRequest request, final ActionListener<
         final UpdateHelper.Result result = updateHelper.prepare(request, indexShard, threadPool::absoluteTimeInMillis);
         switch (result.getResponseResult()) {
             case CREATED -> {
-                IndexRequest upsertRequest = result.action();
+                IndexRequest upsertRequest = removeInferenceMetadataField(indexService, result.action());
                 // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request
                 final BytesReference upsertSourceBytes = upsertRequest.source();
                 client.bulk(
@@ -226,7 +227,7 @@ protected void shardOperation(final UpdateRequest request, final ActionListener<
                 );
             }
             case UPDATED -> {
-                IndexRequest indexRequest = result.action();
+                IndexRequest indexRequest = removeInferenceMetadataField(indexService, result.action());
                 // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request
                 final BytesReference indexSourceBytes = indexRequest.source();
                 client.bulk(
@@ -335,4 +336,15 @@ private void handleUpdateFailureWithRetry(
         }
         listener.onFailure(cause instanceof Exception ? (Exception) cause : new NotSerializableExceptionWrapper(cause));
     }
+
+    private IndexRequest removeInferenceMetadataField(IndexService service, IndexRequest request) {
+        var inferenceMetadata = service.getIndexSettings().getIndexMetadata().getInferenceFields();
+        if (inferenceMetadata.isEmpty()) {
+            return request;
+        }
+        Map<String, Object> docMap = request.sourceAsMap();
+        docMap.remove(InferenceFieldMapper.NAME);
+        request.source(docMap);
+        return request;
+    }
 }