Store ignored source in unique stored fields per entry

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

@@ -39,6 +39,7 @@
 import org.elasticsearch.index.mapper.FallbackSyntheticSourceBlockLoader;
 import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.IgnoreMalformedStoredValues;
+import org.elasticsearch.index.mapper.IgnoredSourceFieldMapper;
 import org.elasticsearch.index.mapper.MapperBuilderContext;
 import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.mapper.SimpleMappedFieldType;
@@ -380,7 +381,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(),
+                    IgnoredSourceFieldMapper.ignoredSourceFormat(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().ignoredSourceFormat()
             )
             : 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(),
+                IgnoredSourceFieldMapper.ignoredSourceFormat(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(),
+                    IgnoredSourceFieldMapper.ignoredSourceFormat(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

@@ -1018,7 +1018,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(),
+                    IgnoredSourceFieldMapper.ignoredSourceFormat(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

@@ -17,13 +17,13 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 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 +39,19 @@
 public abstract class FallbackSyntheticSourceBlockLoader implements BlockLoader {
     private final Reader<?> reader;
     private final String fieldName;
-
-    protected FallbackSyntheticSourceBlockLoader(Reader<?> reader, String fieldName) {
+    private final Set<String> fieldPaths;
+    private final IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat;
+
+    protected FallbackSyntheticSourceBlockLoader(
+        Reader<?> reader,
+        String fieldName,
+        IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat
+    ) {
+        assert ignoredSourceFormat != IgnoredSourceFieldMapper.IgnoredSourceFormat.NO_IGNORED_SOURCE;
         this.reader = reader;
         this.fieldName = fieldName;
+        this.ignoredSourceFormat = ignoredSourceFormat;
+        this.fieldPaths = splitIntoFieldPaths(fieldName);
     }
 
     @Override
@@ -52,12 +61,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, ignoredSourceFormat);
     }
 
     @Override
     public StoredFieldsSpec rowStrideStoredFieldSpec() {
-        return new StoredFieldsSpec(false, false, Set.of(IgnoredSourceFieldMapper.NAME));
+        Set<String> ignoredFieldNames;
+        if (ignoredSourceFormat == IgnoredSourceFieldMapper.IgnoredSourceFormat.PER_FIELD_IGNORED_SOURCE) {
+            ignoredFieldNames = fieldPaths.stream().map(IgnoredSourceFieldMapper::ignoredFieldName).collect(Collectors.toSet());
+        } else {
+            ignoredFieldNames = Set.of(IgnoredSourceFieldMapper.NAME);
+        }
+
+        return new StoredFieldsSpec(false, false, ignoredFieldNames);
     }
 
     @Override
@@ -70,49 +86,51 @@ 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;
-
-        IgnoredSourceRowStrideReader(String fieldName, Reader<T> reader) {
+        private final IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat;
+
+        IgnoredSourceRowStrideReader(
+            String fieldName,
+            Set<String> fieldPaths,
+            Reader<T> reader,
+            IgnoredSourceFieldMapper.IgnoredSourceFormat ignoredSourceFormat
+        ) {
             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.ignoredSourceFormat = ignoredSourceFormat;
         }
 
         @Override
         public void read(int docId, StoredFields storedFields, Builder builder) throws IOException {
-            var ignoredSource = storedFields.storedFields().get(IgnoredSourceFieldMapper.NAME);
-            if (ignoredSource == null) {
+            Map<String, List<IgnoredSourceFieldMapper.NameValue>> valuesForFieldAndParents = ignoredSourceFormat.loadSingleIgnoredField(
+                fieldPaths,
+                storedFields.storedFields()
+            );
+
+            if (valuesForFieldAndParents.isEmpty()) {
                 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);
-                }
-            }
-
             // TODO figure out how to handle XContentDataHelper#voidValue()
 
             var blockValues = new ArrayList<T>();