diff --git a/CHANGELOG.md b/CHANGELOG.md index c393430f193df..ddfd5a3cea320 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java index ccc4b032b4b06..0df977f1ece12 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java @@ -1003,6 +1003,18 @@ public Iterator> 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 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 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; diff --git a/server/src/main/java/org/opensearch/cluster/metadata/IngestionSource.java b/server/src/main/java/org/opensearch/cluster/metadata/IngestionSource.java index c2fcdb8e0fcd0..eb9c51a88b2ce 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/IngestionSource.java +++ b/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 mapperSettings; private IngestionSource( String type, @@ -55,7 +57,8 @@ private IngestionSource( int blockingQueueSize, boolean allActiveIngestion, TimeValue pointerBasedLagUpdateInterval, - IngestionMessageMapper.MapperType mapperType + IngestionMessageMapper.MapperType mapperType, + Map 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 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 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 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 ); } diff --git a/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java b/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java index 6558dd3b9f055..3be2cc3d150a2 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java +++ b/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 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) diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java index 32403f8bb09c2..5486f1c019695 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/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, diff --git a/server/src/main/java/org/opensearch/indices/pollingingest/mappers/FieldMappingIngestionMessageMapper.java b/server/src/main/java/org/opensearch/indices/pollingingest/mappers/FieldMappingIngestionMessageMapper.java new file mode 100644 index 0000000000000..201b0802cfa55 --- /dev/null +++ b/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}. + * + *

Mapper settings: + *

    + *
  • {@code id_field} — source field to use as document {@code _id}. If absent, ID is auto-generated.
  • + *
  • {@code version_field} — source field to use as document {@code _version} with external versioning.
  • + *
  • {@code op_type_field} — source field (boolean) to determine operation type: {@code true} → delete, + * {@code false} → index.
  • + *
+ */ +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 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"); + } +} diff --git a/server/src/main/java/org/opensearch/indices/pollingingest/mappers/IngestionMessageMapper.java b/server/src/main/java/org/opensearch/indices/pollingingest/mappers/IngestionMessageMapper.java index eab0ec064e8e5..48c84250c6c42 100644 --- a/server/src/main/java/org/opensearch/indices/pollingingest/mappers/IngestionMessageMapper.java +++ b/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 + ) ); } } diff --git a/server/src/test/java/org/opensearch/cluster/metadata/IngestionSourceTests.java b/server/src/test/java/org/opensearch/cluster/metadata/IngestionSourceTests.java index feddeb00eab20..1c3f74ead393e 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/IngestionSourceTests.java +++ b/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 params = new HashMap<>(); + params.put("key", "value"); + Map 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()); + } } diff --git a/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java b/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java index 9632ac5b1ffb5..97709f7cbbe2a 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/MetadataCreateIndexServiceTests.java @@ -3793,4 +3793,89 @@ public void testFieldPropertyOverrideInParseV1Mappings() throws Exception { assertEquals("field1 store should be from template1", true, field1.get("store")); } + public void testValidateIngestionSourceSettingsWithFieldMappingOnCurrentVersion() { + // All nodes on current version with valid mapper_settings — should pass + DiscoveryNodes nodes = DiscoveryNodes.builder().add(newNode("node1")).add(newNode("node2")).build(); + ClusterState state = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).nodes(nodes).build(); + + Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_INGESTION_SOURCE_MAPPER_TYPE, "field_mapping") + .put("index.ingestion_source.mapper_settings.id_field", "user_id") + .put("index.ingestion_source.mapper_settings.version_field", "timestamp") + .put("index.ingestion_source.mapper_settings.op_type_field", "is_deleted") + .build(); + + // Should not throw + MetadataCreateIndexService.validateIngestionSourceSettings(settings, state); + } + + public void testValidateIngestionSourceSettingsWithFieldMappingOnOldVersion() { + // One node on older version — should fail + final Set roles = Collections.unmodifiableSet( + new HashSet<>(Arrays.asList(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE, DiscoveryNodeRole.DATA_ROLE)) + ); + DiscoveryNode oldNode = new DiscoveryNode("old_node", buildNewFakeTransportAddress(), emptyMap(), roles, Version.V_3_5_0); + DiscoveryNodes nodes = DiscoveryNodes.builder().add(newNode("node1")).add(oldNode).build(); + ClusterState state = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).nodes(nodes).build(); + + Settings settings = Settings.builder().put(IndexMetadata.SETTING_INGESTION_SOURCE_MAPPER_TYPE, "field_mapping").build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataCreateIndexService.validateIngestionSourceSettings(settings, state) + ); + assertTrue(e.getMessage().contains("mapper_type [field_mapping] requires all nodes")); + assertTrue(e.getMessage().contains(Version.V_3_5_0.toString())); + } + + public void testValidateIngestionSourceSettingsWithDefaultMapperType() { + // Default mapper type on old nodes — should pass (no version requirement) + final Set roles = Collections.unmodifiableSet( + new HashSet<>(Arrays.asList(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE, DiscoveryNodeRole.DATA_ROLE)) + ); + DiscoveryNode oldNode = new DiscoveryNode("old_node", buildNewFakeTransportAddress(), emptyMap(), roles, Version.V_3_5_0); + DiscoveryNodes nodes = DiscoveryNodes.builder().add(oldNode).build(); + ClusterState state = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).nodes(nodes).build(); + + Settings settings = Settings.builder().put(IndexMetadata.SETTING_INGESTION_SOURCE_MAPPER_TYPE, "default").build(); + + // Should not throw + MetadataCreateIndexService.validateIngestionSourceSettings(settings, state); + } + + public void testValidateIngestionSourceSettingsWithUnknownMapperSettingsKey() { + // field_mapping with an unrecognized mapper_settings key — should fail + DiscoveryNodes nodes = DiscoveryNodes.builder().add(newNode("node1")).build(); + ClusterState state = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).nodes(nodes).build(); + + Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_INGESTION_SOURCE_MAPPER_TYPE, "field_mapping") + .put("index.ingestion_source.mapper_settings.id_feild", "user_id") + .build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataCreateIndexService.validateIngestionSourceSettings(settings, state) + ); + assertTrue(e.getMessage().contains("unknown mapper_settings key [id_feild]")); + assertTrue(e.getMessage().contains("field_mapping")); + } + + public void testValidateIngestionSourceSettingsWithMapperSettingsOnDefaultMapper() { + // default mapper type with mapper_settings — should fail + DiscoveryNodes nodes = DiscoveryNodes.builder().add(newNode("node1")).build(); + ClusterState state = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).nodes(nodes).build(); + + Settings settings = Settings.builder() + .put(IndexMetadata.SETTING_INGESTION_SOURCE_MAPPER_TYPE, "default") + .put("index.ingestion_source.mapper_settings.id_field", "user_id") + .build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataCreateIndexService.validateIngestionSourceSettings(settings, state) + ); + assertTrue(e.getMessage().contains("mapper_settings are not supported for mapper_type [default]")); + } + }