Add mapper_settings support and field_mapping mapper type for pull-based ingestion

CHANGELOG.md

@@ -13,6 +13,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Support TLS cert hot-reload for Arrow Flight transport ([#20700](https://github.com/opensearch-project/OpenSearch/pull/20700))
 - [Workload Management] Enhance Scroll API support for autotagging ([#20151](https://github.com/opensearch-project/OpenSearch/pull/20151))
 - Add indices to search request slowlog ([#20588](https://github.com/opensearch-project/OpenSearch/pull/20588))
+- Add mapper_settings support and field_mapping mapper type for pull-based ingestion([#20722](https://github.com/opensearch-project/OpenSearch/pull/20722))
 
 ### Changed
 - Move Randomness from server to libs/common ([#20570](https://github.com/opensearch-project/OpenSearch/pull/20570))

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -1003,6 +1003,18 @@ public Iterator<Setting<?>> settings() {
         }, Property.IndexScope, Property.Dynamic)
     );
 
+    /**
+     * Prefix setting for mapper-specific options. These settings are passed to the configured mapper type.
+     * For example, the {@code field_mapping} mapper type uses {@code id_field}, {@code version_field},
+     * and {@code op_type_field} settings.
+     */
+    public static final Setting.AffixSetting<Object> INGESTION_SOURCE_MAPPER_SETTINGS = Setting.prefixKeySetting(
+        "index.ingestion_source.mapper_settings.",
+        key -> new Setting<>(key, "", (value) -> {
+            return value;
+        }, Property.IndexScope, Property.Final)
+    );
+
     /**
      * an internal index format description, allowing us to find out if this index is upgraded or needs upgrading
      */
@@ -1261,6 +1273,7 @@ public IngestionSource getIngestionSource() {
             final boolean allActiveIngestionEnabled = INGESTION_SOURCE_ALL_ACTIVE_INGESTION_SETTING.get(settings);
             final TimeValue pointerBasedLagUpdateInterval = INGESTION_SOURCE_POINTER_BASED_LAG_UPDATE_INTERVAL_SETTING.get(settings);
             final IngestionMessageMapper.MapperType mapperType = INGESTION_SOURCE_MAPPER_TYPE_SETTING.get(settings);
+            final Map<String, Object> mapperSettings = INGESTION_SOURCE_MAPPER_SETTINGS.getAsMap(settings);
 
             return new IngestionSource.Builder(ingestionSourceType).setParams(ingestionSourceParams)
                 .setPointerInitReset(pointerInitReset)
@@ -1272,6 +1285,7 @@ public IngestionSource getIngestionSource() {
                 .setAllActiveIngestion(allActiveIngestionEnabled)
                 .setPointerBasedLagUpdateInterval(pointerBasedLagUpdateInterval)
                 .setMapperType(mapperType)
+                .setMapperSettings(mapperSettings)
                 .build();
         }
         return null;

server/src/main/java/org/opensearch/cluster/metadata/IngestionSource.java

@@ -15,6 +15,7 @@
 import org.opensearch.indices.pollingingest.StreamPoller;
 import org.opensearch.indices.pollingingest.mappers.IngestionMessageMapper;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -43,6 +44,7 @@ public class IngestionSource {
     private final boolean allActiveIngestion;
     private final TimeValue pointerBasedLagUpdateInterval;
     private final IngestionMessageMapper.MapperType mapperType;
+    private final Map<String, Object> mapperSettings;
 
     private IngestionSource(
         String type,
@@ -55,7 +57,8 @@ private IngestionSource(
         int blockingQueueSize,
         boolean allActiveIngestion,
         TimeValue pointerBasedLagUpdateInterval,
-        IngestionMessageMapper.MapperType mapperType
+        IngestionMessageMapper.MapperType mapperType,
+        Map<String, Object> mapperSettings
     ) {
         this.type = type;
         this.pointerInitReset = pointerInitReset;
@@ -68,6 +71,7 @@ private IngestionSource(
         this.allActiveIngestion = allActiveIngestion;
         this.pointerBasedLagUpdateInterval = pointerBasedLagUpdateInterval;
         this.mapperType = mapperType;
+        this.mapperSettings = mapperSettings != null ? Collections.unmodifiableMap(mapperSettings) : Collections.emptyMap();
     }
 
     public String getType() {
@@ -114,6 +118,10 @@ public IngestionMessageMapper.MapperType getMapperType() {
         return mapperType;
     }
 
+    public Map<String, Object> getMapperSettings() {
+        return mapperSettings;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -129,7 +137,8 @@ public boolean equals(Object o) {
             && Objects.equals(blockingQueueSize, ingestionSource.blockingQueueSize)
             && Objects.equals(allActiveIngestion, ingestionSource.allActiveIngestion)
             && Objects.equals(pointerBasedLagUpdateInterval, ingestionSource.pointerBasedLagUpdateInterval)
-            && Objects.equals(mapperType, ingestionSource.mapperType);
+            && Objects.equals(mapperType, ingestionSource.mapperType)
+            && Objects.equals(mapperSettings, ingestionSource.mapperSettings);
     }
 
     @Override
@@ -145,7 +154,8 @@ public int hashCode() {
             blockingQueueSize,
             allActiveIngestion,
             pointerBasedLagUpdateInterval,
-            mapperType
+            mapperType,
+            mapperSettings
         );
     }
 
@@ -178,6 +188,8 @@ public String toString() {
             + ", mapperType='"
             + mapperType
             + '\''
+            + ", mapperSettings="
+            + mapperSettings
             + '}';
     }
 
@@ -240,6 +252,7 @@ public static class Builder {
             Settings.EMPTY
         );
         private IngestionMessageMapper.MapperType mapperType = INGESTION_SOURCE_MAPPER_TYPE_SETTING.getDefault(Settings.EMPTY);
+        private Map<String, Object> mapperSettings = new HashMap<>();
 
         public Builder(String type) {
             this.type = type;
@@ -255,6 +268,7 @@ public Builder(IngestionSource ingestionSource) {
             this.allActiveIngestion = ingestionSource.allActiveIngestion;
             this.pointerBasedLagUpdateInterval = ingestionSource.pointerBasedLagUpdateInterval;
             this.mapperType = ingestionSource.mapperType;
+            this.mapperSettings = new HashMap<>(ingestionSource.mapperSettings);
         }
 
         public Builder setPointerInitReset(PointerInitReset pointerInitReset) {
@@ -312,6 +326,11 @@ public Builder setMapperType(IngestionMessageMapper.MapperType mapperType) {
             return this;
         }
 
+        public Builder setMapperSettings(Map<String, Object> mapperSettings) {
+            this.mapperSettings = mapperSettings;
+            return this;
+        }
+
         public IngestionSource build() {
             return new IngestionSource(
                 type,
@@ -324,7 +343,8 @@ public IngestionSource build() {
                 blockingQueueSize,
                 allActiveIngestion,
                 pointerBasedLagUpdateInterval,
-                mapperType
+                mapperType,
+                mapperSettings
             );
         }
 

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java

@@ -107,6 +107,8 @@
 import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.indices.ShardLimitValidator;
 import org.opensearch.indices.SystemIndices;
+import org.opensearch.indices.pollingingest.mappers.FieldMappingIngestionMessageMapper;
+import org.opensearch.indices.pollingingest.mappers.IngestionMessageMapper;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.node.remotestore.RemoteStoreNodeAttribute;
 import org.opensearch.node.remotestore.RemoteStoreNodeService;
@@ -1269,6 +1271,54 @@ private static void validateSearchOnlyReplicasSettings(Settings indexSettings) {
         }
     }
 
+    /**
+     * Validates ingestion source settings for version compatibility and mapper settings correctness.
+     * In a mixed cluster, older nodes may not recognize newer mapper types (e.g., field_mapping),
+     * which would cause failures when those nodes try to initialize the ingestion engine.
+     * Also validates that mapper_settings keys are recognized for the configured mapper_type.
+     */
+    static void validateIngestionSourceSettings(Settings settings, ClusterState state) {
+        if (IndexMetadata.INGESTION_SOURCE_MAPPER_TYPE_SETTING.exists(settings) == false) {
+            return;
+        }
+
+        IngestionMessageMapper.MapperType mapperType = IndexMetadata.INGESTION_SOURCE_MAPPER_TYPE_SETTING.get(settings);
+        Map<String, Object> mapperSettings = IndexMetadata.INGESTION_SOURCE_MAPPER_SETTINGS.getAsMap(settings);
+
+        switch (mapperType) {
+            case FIELD_MAPPING:
+                // Version check for mixed cluster compatibility
+                Version minNodeVersion = state.nodes().getMinNodeVersion();
+                if (minNodeVersion.before(Version.V_3_6_0)) {
+                    throw new IllegalArgumentException(
+                        "mapper_type [field_mapping] requires all nodes in the cluster to be on version ["
+                            + Version.V_3_6_0
+                            + "] or later, but the minimum node version is ["
+                            + minNodeVersion
+                            + "]"
+                    );
+                }
+                // Validate mapper_settings keys
+                for (String key : mapperSettings.keySet()) {
+                    if (FieldMappingIngestionMessageMapper.VALID_SETTINGS.contains(key) == false) {
+                        throw new IllegalArgumentException(
+                            "unknown mapper_settings key ["
+                                + key
+                                + "] for mapper_type [field_mapping]. Valid keys are: "
+                                + FieldMappingIngestionMessageMapper.VALID_SETTINGS
+                        );
+                    }
+                }
+                break;
+            default:
+                // default and raw_payload mappers don't use mapper_settings
+                if (mapperSettings.isEmpty() == false) {
+                    throw new IllegalArgumentException("mapper_settings are not supported for mapper_type [" + mapperType.getName() + "]");
+                }
+                break;
+        }
+    }
+
     /**
      * Updates index settings to set replication strategy by default based on cluster level settings or remote store
      * node attributes
@@ -1631,6 +1681,7 @@ private void validate(CreateIndexClusterStateUpdateRequest request, ClusterState
         validateIndexName(request.index(), state);
         validateIndexSettings(request.index(), request.settings(), forbidPrivateIndexSettings);
         validateContext(request);
+        validateIngestionSourceSettings(request.settings(), state);
     }
 
     public void validateIndexSettings(String indexName, final Settings settings, final boolean forbidPrivateIndexSettings)

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -290,6 +290,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexMetadata.INGESTION_SOURCE_ALL_ACTIVE_INGESTION_SETTING,
                 IndexMetadata.INGESTION_SOURCE_POINTER_BASED_LAG_UPDATE_INTERVAL_SETTING,
                 IndexMetadata.INGESTION_SOURCE_MAPPER_TYPE_SETTING,
+                IndexMetadata.INGESTION_SOURCE_MAPPER_SETTINGS,
 
                 // Settings for search replica
                 IndexMetadata.INDEX_NUMBER_OF_SEARCH_REPLICAS_SETTING,

server/src/main/java/org/opensearch/indices/pollingingest/mappers/FieldMappingIngestionMessageMapper.java

@@ -0,0 +1,47 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.indices.pollingingest.mappers;
+
+import org.opensearch.index.IngestionShardPointer;
+import org.opensearch.index.Message;
+import org.opensearch.indices.pollingingest.ShardUpdateMessage;
+
+import java.util.Set;
+
+/**
+ * Mapper implementation that extracts document metadata ({@code _id}, {@code _version}, {@code _op_type})
+ * from configurable top-level fields in the raw message payload. The remaining fields become the document
+ * {@code _source}.
+ *
+ * <p>Mapper settings:
+ * <ul>
+ *   <li>{@code id_field} — source field to use as document {@code _id}. If absent, ID is auto-generated.</li>
+ *   <li>{@code version_field} — source field to use as document {@code _version} with external versioning.</li>
+ *   <li>{@code op_type_field} — source field (boolean) to determine operation type: {@code true} → delete,
+ *       {@code false} → index.</li>
+ * </ul>
+ */
+public class FieldMappingIngestionMessageMapper implements IngestionMessageMapper {
+
+    /** Mapper setting key: source field to use as document _id */
+    public static final String ID_FIELD = "id_field";
+    /** Mapper setting key: source field to use as document _version */
+    public static final String VERSION_FIELD = "version_field";
+    /** Mapper setting key: source field to determine operation type (index vs delete) */
+    public static final String OP_TYPE_FIELD = "op_type_field";
+
+    /** Valid mapper_settings keys for this mapper type */
+    public static final Set<String> VALID_SETTINGS = Set.of(ID_FIELD, VERSION_FIELD, OP_TYPE_FIELD);
+
+    @Override
+    public ShardUpdateMessage mapAndProcess(IngestionShardPointer pointer, Message message) throws IllegalArgumentException {
+        // TODO: pending implementation
+        throw new UnsupportedOperationException("FieldMappingIngestionMessageMapper is not yet implemented");
+    }
+}

server/src/main/java/org/opensearch/indices/pollingingest/mappers/IngestionMessageMapper.java

@@ -40,7 +40,8 @@ public interface IngestionMessageMapper {
     @ExperimentalApi
     enum MapperType {
         DEFAULT("default"),
-        RAW_PAYLOAD("raw_payload");
+        RAW_PAYLOAD("raw_payload"),
+        FIELD_MAPPING("field_mapping");
 
         private final String name;
 
@@ -59,7 +60,11 @@ public static MapperType fromString(String value) {
                 }
             }
             throw new IllegalArgumentException(
-                String.format(Locale.ROOT, "Unknown ingestion mapper type: %s. Valid values are: default, raw_payload", value)
+                String.format(
+                    Locale.ROOT,
+                    "Unknown ingestion mapper type: %s. Valid values are: default, raw_payload, field_mapping",
+                    value
+                )
             );
         }
     }

server/src/test/java/org/opensearch/cluster/metadata/IngestionSourceTests.java

@@ -108,7 +108,7 @@ public void testToString() {
             .setErrorStrategy(DROP)
             .build();
         String expected =
-            "IngestionSource{type='type',pointer_init_reset='PointerInitReset{type='RESET_BY_OFFSET', value=1000}',error_strategy='DROP', params={key=value}, maxPollSize=1000, pollTimeout=1000, numProcessorThreads=1, blockingQueueSize=100, allActiveIngestion=false, pointerBasedLagUpdateInterval=10s, mapperType='DEFAULT'}";
+            "IngestionSource{type='type',pointer_init_reset='PointerInitReset{type='RESET_BY_OFFSET', value=1000}',error_strategy='DROP', params={key=value}, maxPollSize=1000, pollTimeout=1000, numProcessorThreads=1, blockingQueueSize=100, allActiveIngestion=false, pointerBasedLagUpdateInterval=10s, mapperType='DEFAULT', mapperSettings={}}";
         assertEquals(expected, source.toString());
     }
 
@@ -137,4 +137,44 @@ public void testAllActiveIngestionConstructorAndGetter() {
         IngestionSource ingestionSourceClone = new IngestionSource.Builder(sourceEnabled).build();
         assertTrue(ingestionSourceClone.isAllActiveIngestionEnabled());
     }
+
+    public void testMapperSettings() {
+        Map<String, Object> params = new HashMap<>();
+        params.put("key", "value");
+        Map<String, Object> mapperSettings = new HashMap<>();
+        mapperSettings.put("id_field", "user_id");
+        mapperSettings.put("version_field", "timestamp");
+        mapperSettings.put("op_type_field", "is_deleted");
+
+        IngestionSource source = new IngestionSource.Builder("type").setParams(params)
+            .setPointerInitReset(pointerInitReset)
+            .setErrorStrategy(DROP)
+            .setMapperSettings(mapperSettings)
+            .build();
+
+        assertEquals(mapperSettings, source.getMapperSettings());
+        assertEquals("user_id", source.getMapperSettings().get("id_field"));
+        assertEquals("timestamp", source.getMapperSettings().get("version_field"));
+        assertEquals("is_deleted", source.getMapperSettings().get("op_type_field"));
+
+        // Test copy constructor preserves mapper settings
+        IngestionSource copy = new IngestionSource.Builder(source).build();
+        assertEquals(mapperSettings, copy.getMapperSettings());
+
+        // Test equals with mapper settings
+        IngestionSource source2 = new IngestionSource.Builder("type").setParams(params)
+            .setPointerInitReset(pointerInitReset)
+            .setErrorStrategy(DROP)
+            .setMapperSettings(new HashMap<>(mapperSettings))
+            .build();
+        assertEquals(source, source2);
+        assertEquals(source.hashCode(), source2.hashCode());
+
+        // Test empty mapper settings by default
+        IngestionSource sourceNoMapperSettings = new IngestionSource.Builder("type").setParams(params)
+            .setPointerInitReset(pointerInitReset)
+            .setErrorStrategy(DROP)
+            .build();
+        assertTrue(sourceNoMapperSettings.getMapperSettings().isEmpty());
+    }
 }