Isolate Real-Time Get from Translog Assertion Logic (#121161) (#121222)

This commit refactors the logic introduced in #121092, ensuring that translog assertion logic related to inference fields resides exclusively within `TranslogOperationAsserter`.

For mappings that contain inference fields, we regenerate the metadata field's content during peer recovery.
This can create discrepancies between the original and regenerated sources, which is expected since indexed values may differ from the original.
This change ensures that we compare the "synthetic" version of index operations in the same way we do for mappings with synthetic source enabled.
In contrast, real-time get (`realtime-get`) simply filters inference fields from the original source, eliminating the need to synthesize them as we do for synthetic sources.

Author: jimczi

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -825,7 +825,8 @@ private GetResult getFromTranslog(
             mappingLookup,
             documentParser,
             config(),
-            translogInMemorySegmentsCount::incrementAndGet
+            translogInMemorySegmentsCount::incrementAndGet,
+            false
         );
         final Searcher searcher = new Searcher(
             "realtime_get",

server/src/main/java/org/elasticsearch/index/engine/TranslogDirectoryReader.java

@@ -87,7 +87,8 @@ static DirectoryReader create(
         MappingLookup mappingLookup,
         DocumentParser documentParser,
         EngineConfig engineConfig,
-        Runnable onSegmentCreated
+        Runnable onSegmentCreated,
+        boolean forceSynthetic
     ) throws IOException {
         final Directory directory = new ByteBuffersDirectory();
         boolean success = false;
@@ -96,7 +97,7 @@ static DirectoryReader create(
             // When using synthetic source, the translog operation must always be reindexed into an in-memory Lucene to ensure consistent
             // output for realtime-get operations. However, this can degrade the performance of realtime-get and update operations.
             // If slight inconsistencies in realtime-get operations are acceptable, the translog operation can be reindexed lazily.
-            if (mappingLookup.isSourceSynthetic() || mappingLookup.inferenceFields().isEmpty() == false) {
+            if (mappingLookup.isSourceSynthetic() || forceSynthetic) {
                 onSegmentCreated.run();
                 leafReader = createInMemoryReader(shardId, engineConfig, directory, documentParser, mappingLookup, false, operation);
             } else {

server/src/main/java/org/elasticsearch/index/engine/TranslogOperationAsserter.java

@@ -52,7 +52,7 @@ static Translog.Index synthesizeSource(EngineConfig engineConfig, Translog.Index
         final ShardId shardId = engineConfig.getShardId();
         final MappingLookup mappingLookup = engineConfig.getMapperService().mappingLookup();
         final DocumentParser documentParser = engineConfig.getMapperService().documentParser();
-        try (var reader = TranslogDirectoryReader.create(shardId, op, mappingLookup, documentParser, engineConfig, () -> {})) {
+        try (var reader = TranslogDirectoryReader.create(shardId, op, mappingLookup, documentParser, engineConfig, () -> {}, true)) {
             final Engine.Searcher searcher = new Engine.Searcher(
                 "assert_translog",
                 reader,

x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilterIT.java

@@ -20,8 +20,9 @@
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.InferenceMetadataFieldsMapper;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapper;
 import org.elasticsearch.inference.SimilarityMeasure;
 import org.elasticsearch.plugins.Plugin;
@@ -49,14 +50,21 @@ public class ShardBulkInferenceActionFilterIT extends ESIntegTestCase {
     public static final String INDEX_NAME = "test-index";
 
     private final boolean useLegacyFormat;
+    private final boolean useSyntheticSource;
 
-    public ShardBulkInferenceActionFilterIT(boolean useLegacyFormat) {
+    public ShardBulkInferenceActionFilterIT(boolean useLegacyFormat, boolean useSyntheticSource) {
         this.useLegacyFormat = useLegacyFormat;
+        this.useSyntheticSource = useSyntheticSource;
     }
 
     @ParametersFactory
     public static Iterable<Object[]> parameters() throws Exception {
-        return List.of(new Object[] { true }, new Object[] { false });
+        return List.of(
+            new Object[] { true, false },
+            new Object[] { true, true },
+            new Object[] { false, false },
+            new Object[] { false, true }
+        );
     }
 
     @Before
@@ -79,37 +87,38 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     @Override
     public Settings indexSettings() {
-        return Settings.builder()
-            .put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current())
+        var builder = Settings.builder()
             .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, randomIntBetween(1, 10))
-            .put(InferenceMetadataFieldsMapper.USE_LEGACY_SEMANTIC_TEXT_FORMAT.getKey(), useLegacyFormat)
-            .build();
+            .put(InferenceMetadataFieldsMapper.USE_LEGACY_SEMANTIC_TEXT_FORMAT.getKey(), useLegacyFormat);
+        if (useSyntheticSource) {
+            builder.put(IndexSettings.RECOVERY_USE_SYNTHETIC_SOURCE_SETTING.getKey(), true);
+            builder.put(IndexSettings.INDEX_MAPPER_SOURCE_MODE_SETTING.getKey(), SourceFieldMapper.Mode.SYNTHETIC.name());
+        }
+        return builder.build();
     }
 
     public void testBulkOperations() throws Exception {
-        indicesAdmin().prepareCreate(INDEX_NAME)
-            .setMapping(
-                String.format(
-                    Locale.ROOT,
-                    """
-                        {
-                            "properties": {
-                                "sparse_field": {
-                                    "type": "semantic_text",
-                                    "inference_id": "%s"
-                                },
-                                "dense_field": {
-                                    "type": "semantic_text",
-                                    "inference_id": "%s"
-                                }
+        prepareCreate(INDEX_NAME).setMapping(
+            String.format(
+                Locale.ROOT,
+                """
+                    {
+                        "properties": {
+                            "sparse_field": {
+                                "type": "semantic_text",
+                                "inference_id": "%s"
+                            },
+                            "dense_field": {
+                                "type": "semantic_text",
+                                "inference_id": "%s"
                             }
                         }
-                        """,
-                    TestSparseInferenceServiceExtension.TestInferenceService.NAME,
-                    TestDenseInferenceServiceExtension.TestInferenceService.NAME
-                )
+                    }
+                    """,
+                TestSparseInferenceServiceExtension.TestInferenceService.NAME,
+                TestDenseInferenceServiceExtension.TestInferenceService.NAME
             )
-            .get();
+        ).get();
 
         int totalBulkReqs = randomIntBetween(2, 100);
         long totalDocs = 0;