Store ignored source in unique stored fields per entry

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/ScaledFloatFieldMapper.java

@@ -380,7 +380,11 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
             }
             // Multi fields don't have fallback synthetic source.
             if (isSyntheticSource && blContext.parentField(name()) == null) {
-                return new FallbackSyntheticSourceBlockLoader(fallbackSyntheticSourceBlockLoaderReader(), name()) {
+                return new FallbackSyntheticSourceBlockLoader(
+                    fallbackSyntheticSourceBlockLoaderReader(),
+                    name(),
+                    blContext.indexSettings().getIndexVersionCreated()
+                ) {
                     @Override
                     public Builder builder(BlockFactory factory, int expectedCount) {
                         return factory.doubles(expectedCount);

server/src/main/java/org/elasticsearch/index/IndexVersions.java

@@ -180,6 +180,7 @@ private static Version parseUnchecked(String version) {
     public static final IndexVersion SPARSE_VECTOR_PRUNING_INDEX_OPTIONS_SUPPORT = def(9_031_0_00, Version.LUCENE_10_2_2);
     public static final IndexVersion DEFAULT_DENSE_VECTOR_TO_BBQ_HNSW = def(9_032_0_00, Version.LUCENE_10_2_2);
     public static final IndexVersion MATCH_ONLY_TEXT_STORED_AS_BYTES = def(9_033_0_00, Version.LUCENE_10_2_2);
+    public static final IndexVersion IGNORED_SOURCE_FIELDS_PER_ENTRY = def(9_034_0_00, Version.LUCENE_10_2_2);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/index/fieldvisitor/StoredFieldLoader.java

@@ -9,12 +9,14 @@
 
 package org.elasticsearch.index.fieldvisitor;
 
+import org.apache.lucene.index.FieldInfo;
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.StoredFields;
 import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.lucene.index.SequentialStoredFieldsLeafReader;
+import org.elasticsearch.index.mapper.IgnoredSourceFieldMapper;
 import org.elasticsearch.search.fetch.StoredFieldsSpec;
 
 import java.io.IOException;
@@ -203,9 +205,25 @@ private static class ReaderStoredFieldLoader implements LeafStoredFieldLoader {
         private final CustomFieldsVisitor visitor;
         private int doc = -1;
 
+        private static CustomFieldsVisitor getFieldsVisitor(Set<String> fields, boolean loadSource) {
+            if (fields.contains(IgnoredSourceFieldMapper.NAME)) {
+                return new CustomFieldsVisitor(fields, loadSource) {
+                    @Override
+                    public Status needsField(FieldInfo fieldInfo) {
+                        if (fieldInfo.name.startsWith(IgnoredSourceFieldMapper.NAME)) {
+                            return Status.YES;
+                        }
+                        return super.needsField(fieldInfo);
+                    }
+                };
+            }
+
+            return new CustomFieldsVisitor(fields, loadSource);
+        }
+
         ReaderStoredFieldLoader(CheckedBiConsumer<Integer, FieldsVisitor, IOException> reader, boolean loadSource, Set<String> fields) {
             this.reader = reader;
-            this.visitor = new CustomFieldsVisitor(fields, loadSource);
+            this.visitor = getFieldsVisitor(fields, loadSource);
         }
 
         @Override

server/src/main/java/org/elasticsearch/index/get/GetResult.java

@@ -244,7 +244,7 @@ public XContentBuilder toXContentEmbedded(XContentBuilder builder, Params params
 
         for (DocumentField field : metaFields.values()) {
             // TODO: can we avoid having an exception here?
-            if (field.getName().equals(IgnoredFieldMapper.NAME) || field.getName().equals(IgnoredSourceFieldMapper.NAME)) {
+            if (field.getName().equals(IgnoredFieldMapper.NAME) || field.getName().startsWith(IgnoredSourceFieldMapper.NAME)) {
                 builder.field(field.getName(), field.getValues());
             } else {
                 builder.field(field.getName(), field.<Object>getValue());

server/src/main/java/org/elasticsearch/index/get/ShardGetService.java

@@ -322,7 +322,8 @@ private GetResult innerGetFetch(
             ? new SourceLoader.Synthetic(
                 sourceFilter,
                 () -> mappingLookup.getMapping().syntheticFieldLoader(sourceFilter),
-                mapperMetrics.sourceFieldMetrics()
+                mapperMetrics.sourceFieldMetrics(),
+                mappingLookup.getMapping().ignoredFieldsLoader()
             )
             : mappingLookup.newSourceLoader(sourceFilter, mapperMetrics.sourceFieldMetrics());
         StoredFieldLoader storedFieldLoader = buildStoredFieldLoader(storedFieldSet, fetchSourceContext, loader);

server/src/main/java/org/elasticsearch/index/mapper/AbstractGeometryFieldMapper.java

@@ -194,7 +194,11 @@ protected BlockLoader blockLoaderFromSource(BlockLoaderContext blContext) {
         protected abstract Object nullValueAsSource(T nullValue);
 
         protected BlockLoader blockLoaderFromFallbackSyntheticSource(BlockLoaderContext blContext) {
-            return new FallbackSyntheticSourceBlockLoader(new GeometriesFallbackSyntheticSourceReader(), name()) {
+            return new FallbackSyntheticSourceBlockLoader(
+                new GeometriesFallbackSyntheticSourceReader(),
+                name(),
+                blContext.indexSettings().getIndexVersionCreated()
+            ) {
                 @Override
                 public Builder builder(BlockFactory factory, int expectedCount) {
                     return factory.bytesRefs(expectedCount);

server/src/main/java/org/elasticsearch/index/mapper/BooleanFieldMapper.java

@@ -352,7 +352,11 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
 
             // Multi fields don't have fallback synthetic source.
             if (isSyntheticSource && blContext.parentField(name()) == null) {
-                return new FallbackSyntheticSourceBlockLoader(fallbackSyntheticSourceBlockLoaderReader(), name()) {
+                return new FallbackSyntheticSourceBlockLoader(
+                    fallbackSyntheticSourceBlockLoaderReader(),
+                    name(),
+                    blContext.indexSettings().getIndexVersionCreated()
+                ) {
                     @Override
                     public Builder builder(BlockFactory factory, int expectedCount) {
                         return factory.booleans(expectedCount);

server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java

@@ -1002,7 +1002,11 @@ public BlockLoader blockLoader(BlockLoaderContext blContext) {
 
             // Multi fields don't have fallback synthetic source.
             if (isSyntheticSource && blContext.parentField(name()) == null) {
-                return new FallbackSyntheticSourceBlockLoader(fallbackSyntheticSourceBlockLoaderReader(), name()) {
+                return new FallbackSyntheticSourceBlockLoader(
+                    fallbackSyntheticSourceBlockLoaderReader(),
+                    name(),
+                    blContext.indexSettings().getIndexVersionCreated()
+                ) {
                     @Override
                     public Builder builder(BlockFactory factory, int expectedCount) {
                         return factory.longs(expectedCount);

server/src/main/java/org/elasticsearch/index/mapper/FallbackSyntheticSourceBlockLoader.java

@@ -11,6 +11,9 @@
 
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.search.fetch.StoredFieldsSpec;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
@@ -24,6 +27,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.Stack;
+import java.util.stream.Collectors;
 
 /**
  * Block loader for fields that use fallback synthetic source implementation.
@@ -39,10 +43,14 @@
 public abstract class FallbackSyntheticSourceBlockLoader implements BlockLoader {
     private final Reader<?> reader;
     private final String fieldName;
+    private final Set<String> fieldPaths;
+    private final IndexVersion indexCreatedVersion;
 
-    protected FallbackSyntheticSourceBlockLoader(Reader<?> reader, String fieldName) {
+    protected FallbackSyntheticSourceBlockLoader(Reader<?> reader, String fieldName, IndexVersion indexCreatedVersion) {
         this.reader = reader;
         this.fieldName = fieldName;
+        this.indexCreatedVersion = indexCreatedVersion;
+        this.fieldPaths = splitIntoFieldPaths(fieldName);
     }
 
     @Override
@@ -52,12 +60,19 @@ public ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) throws
 
     @Override
     public RowStrideReader rowStrideReader(LeafReaderContext context) throws IOException {
-        return new IgnoredSourceRowStrideReader<>(fieldName, reader);
+        return new IgnoredSourceRowStrideReader<>(fieldName, fieldPaths, reader, indexCreatedVersion);
     }
 
     @Override
     public StoredFieldsSpec rowStrideStoredFieldSpec() {
-        return new StoredFieldsSpec(false, false, Set.of(IgnoredSourceFieldMapper.NAME));
+        Set<String> ignoredFieldNames;
+        if (indexCreatedVersion.onOrAfter(IndexVersions.IGNORED_SOURCE_FIELDS_PER_ENTRY)) {
+            ignoredFieldNames = fieldPaths.stream().map(IgnoredSourceFieldMapper::ignoredFieldName).collect(Collectors.toSet());
+        } else {
+            ignoredFieldNames = Set.of(IgnoredSourceFieldMapper.NAME);
+        }
+
+        return new StoredFieldsSpec(false, false, ignoredFieldNames);
     }
 
     @Override
@@ -70,46 +85,73 @@ public SortedSetDocValues ordinals(LeafReaderContext context) throws IOException
         throw new UnsupportedOperationException();
     }
 
+    public static Set<String> splitIntoFieldPaths(String fieldName) {
+        var paths = new HashSet<String>();
+        paths.add("_doc");
+        var current = new StringBuilder();
+        for (var part : fieldName.split("\\.")) {
+            if (current.isEmpty() == false) {
+                current.append('.');
+            }
+            current.append(part);
+            paths.add(current.toString());
+        }
+        return paths;
+    }
+
     private static class IgnoredSourceRowStrideReader<T> implements RowStrideReader {
         // Contains name of the field and all its parents
-        private final Set<String> fieldNames;
         private final String fieldName;
+        private final Set<String> fieldPaths;
         private final Reader<T> reader;
+        private final IndexVersion indexCreatedVersion;
 
-        IgnoredSourceRowStrideReader(String fieldName, Reader<T> reader) {
+        IgnoredSourceRowStrideReader(String fieldName, Set<String> fieldPaths, Reader<T> reader, IndexVersion indexCreatedVersion) {
             this.fieldName = fieldName;
+            this.fieldPaths = fieldPaths;
             this.reader = reader;
-            this.fieldNames = new HashSet<>() {
-                {
-                    add("_doc");
-                }
-            };
-
-            var current = new StringBuilder();
-            for (String part : fieldName.split("\\.")) {
-                if (current.isEmpty() == false) {
-                    current.append('.');
-                }
-                current.append(part);
-                fieldNames.add(current.toString());
-            }
-
+            this.indexCreatedVersion = indexCreatedVersion;
         }
 
         @Override
         public void read(int docId, StoredFields storedFields, Builder builder) throws IOException {
-            var ignoredSource = storedFields.storedFields().get(IgnoredSourceFieldMapper.NAME);
-            if (ignoredSource == null) {
-                builder.appendNull();
-                return;
-            }
-
             Map<String, List<IgnoredSourceFieldMapper.NameValue>> valuesForFieldAndParents = new HashMap<>();
 
-            for (Object value : ignoredSource) {
-                IgnoredSourceFieldMapper.NameValue nameValue = IgnoredSourceFieldMapper.decode(value);
-                if (fieldNames.contains(nameValue.name())) {
-                    valuesForFieldAndParents.computeIfAbsent(nameValue.name(), k -> new ArrayList<>()).add(nameValue);
+            if (indexCreatedVersion.onOrAfter(IndexVersions.IGNORED_SOURCE_FIELDS_PER_ENTRY)) {
+                List<Object> ignoredSource = null;
+                for (var parentPath : fieldPaths) {
+                    ignoredSource = storedFields.storedFields().get(IgnoredSourceFieldMapper.ignoredFieldName(parentPath));
+                    if (ignoredSource == null) {
+                        continue;
+                    }
+                    assert ignoredSource.size() == 1;
+
+                    List<IgnoredSourceFieldMapper.NameValue> nameValues = IgnoredSourceFieldMapper.decodeMulti(
+                        (BytesRef) ignoredSource.getFirst()
+                    );
+
+                    for (var nameValue : nameValues) {
+                        assert fieldPaths.contains(nameValue.name());
+                        valuesForFieldAndParents.computeIfAbsent(nameValue.name(), k -> new ArrayList<>()).add(nameValue);
+                    }
+                }
+                if (valuesForFieldAndParents.isEmpty()) {
+                    builder.appendNull();
+                    return;
+                }
+
+            } else {
+                var ignoredSource = storedFields.storedFields().get(IgnoredSourceFieldMapper.NAME);
+                if (ignoredSource == null) {
+                    builder.appendNull();
+                    return;
+                }
+
+                for (Object value : ignoredSource) {
+                    IgnoredSourceFieldMapper.NameValue nameValue = IgnoredSourceFieldMapper.decode(value);
+                    if (fieldPaths.contains(nameValue.name())) {
+                        valuesForFieldAndParents.computeIfAbsent(nameValue.name(), k -> new ArrayList<>()).add(nameValue);
+                    }
                 }
             }