simonelbaz
diff --git a/‎data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/rule/RuleEvaluator.java‎
Lines changed: 27 additions & 7 deletions b/‎data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/rule/RuleEvaluator.java‎
Lines changed: 27 additions & 7 deletions
diff --git a/‎data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/transformer/DynamicConfigTransformer.java‎
Lines changed: 163 additions & 0 deletions b/‎data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/transformer/DynamicConfigTransformer.java‎
Lines changed: 163 additions & 0 deletions
diff --git a/‎data-prepper-pipeline-parser/src/test/java/org/opensearch/dataprepper/pipeline/parser/rule/RuleEvaluatorTest.java‎
Lines changed: 41 additions & 0 deletions b/‎data-prepper-pipeline-parser/src/test/java/org/opensearch/dataprepper/pipeline/parser/rule/RuleEvaluatorTest.java‎
Lines changed: 41 additions & 0 deletions
@@ -24,6 +24,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -91,10 +92,23 @@ private RuleFileEvaluation evaluate(String pipelinesJson) {
         try {
             Collection<RuleStream> ruleStreams = transformersFactory.loadRules();
 
-            //walk through all rules and return first valid
+            // Pre-parse all rules and sort by specificity (most conditions first)
+            // so more specific rules like rds-joins match before generic rds
+            List<ParsedRule> parsedRules = new ArrayList<>();
             for (RuleStream ruleStream : ruleStreams) {
                 try {
-                    rulesModel = yamlMapper.readValue(ruleStream.getRuleStream(), RuleTransformerModel.class);
+                    RuleTransformerModel model = yamlMapper.readValue(ruleStream.getRuleStream(), RuleTransformerModel.class);
+                    parsedRules.add(new ParsedRule(model, ruleStream.getName()));
+                } finally {
+                    ruleStream.close();
+                }
+            }
+            parsedRules.sort((a, b) -> Integer.compare(
+                    b.model.getApplyWhen().size(), a.model.getApplyWhen().size()));
+
+            //walk through all rules and return first valid
+            for (ParsedRule parsedRule : parsedRules) {
+                    rulesModel = parsedRule.model;
                     List<String> rules = rulesModel.getApplyWhen();
                     String pluginName = rulesModel.getPluginName();
                     boolean allRulesValid = true;
@@ -107,7 +121,7 @@ private RuleFileEvaluation evaluate(String pipelinesJson) {
                                 break;
                             }
                         } catch (PathNotFoundException e) {
-                            LOG.debug("Json Path not found for {}", ruleStream.getName());
+                            LOG.debug("Json Path not found for {}", parsedRule.fileName);
                             allRulesValid = false;
                             break;
                         }
@@ -116,13 +130,10 @@ private RuleFileEvaluation evaluate(String pipelinesJson) {
                     if (allRulesValid) {
                         return RuleFileEvaluation.builder()
                                 .withPluginName(pluginName)
-                                .withRuleFileName(ruleStream.getName())
+                                .withRuleFileName(parsedRule.fileName)
                                 .withResult(true)
                                 .build();
                     }
-                } finally {
-                    ruleStream.close();
-                }
             }
 
         } catch (FileNotFoundException e) {
@@ -143,4 +154,13 @@ private RuleFileEvaluation evaluate(String pipelinesJson) {
                 .build();
     }
 
+    private static class ParsedRule {
+        final RuleTransformerModel model;
+        final String fileName;
+
+        ParsedRule(RuleTransformerModel model, String fileName) {
+            this.model = model;
+            this.fileName = fileName;
+        }
+    }
 }
@@ -82,6 +82,12 @@ public class DynamicConfigTransformer implements PipelineConfigurationTransforme
 
     private static final String S3_BUFFER_PREFIX = "/buffer";
 
+    /**
+     * Pattern to match overlay directives like "<<overlay sink[*].opensearch>>"
+     * The captured group is the target path (e.g., "sink[*].opensearch")
+     */
+    private static final Pattern OVERLAY_PATTERN = Pattern.compile("^<<overlay\\s+(.+?)>>$");
+
     Configuration parseConfigWithJsonNode = Configuration.builder()
             .jsonProvider(new JacksonJsonNodeJsonProvider())
             .mappingProvider(new JacksonMappingProvider())
@@ -166,6 +172,9 @@ public PipelinesDataFlowModel transformConfiguration(PipelinesDataFlowModel preT
                 }
             });
 
+            // Process <<overlay>> directives — merge overlay fields into resolved config
+            processOverlayDirectives(templateRootNode, pipelineJson);
+
             PipelinesDataFlowModel transformedPipelinesDataFlowModel = getTransformedPipelinesDataFlowModel(pipelineNameThatNeedsTransformation,
                     preTransformedPipelinesDataFlowModel,
                     templateRootNode,
@@ -286,6 +295,10 @@ private void populateMapWithPlaceholderPaths(JsonNode currentNode, String curren
             Iterator<Map.Entry<String, JsonNode>> fields = currentNode.fields();
             while (fields.hasNext()) {
                 Map.Entry<String, JsonNode> entry = fields.next();
+                // Skip overlay directive keys — they are processed separately
+                if (OVERLAY_PATTERN.matcher(entry.getKey()).matches()) {
+                    continue;
+                }
                 String path = currentPath.isEmpty() ? entry.getKey() : currentPath + "." + entry.getKey();
                 populateMapWithPlaceholderPaths(entry.getValue(), path, placeholdersWithPaths);
             }
@@ -489,6 +502,156 @@ public Object invokeMethod(String methodName, Class<?> parameterType, Object arg
     }
 
 
+    /**
+     * Processes <<overlay path>> directives in the template.
+     * For each overlay directive found as a key in any object node, merges the overlay
+     * value into the target path. Supports [*] wildcard to apply to all array elements
+     * matching a specific key.
+     *
+     * Example: <<overlay sink[*].opensearch>> with value {action: "upsert", script: {...}}
+     * will merge those fields into every opensearch sink entry.
+     */
+    private void processOverlayDirectives(JsonNode rootNode, String pipelineJson) {
+        processOverlayDirectivesRecursive(rootNode, pipelineJson);
+    }
+
+    private void processOverlayDirectivesRecursive(JsonNode node, String pipelineJson) {
+        if (node.isObject()) {
+            ObjectNode objectNode = (ObjectNode) node;
+            List<String> overlayKeys = new ArrayList<>();
+
+            // First pass: collect overlay keys
+            Iterator<String> fieldNames = objectNode.fieldNames();
+            while (fieldNames.hasNext()) {
+                String fieldName = fieldNames.next();
+                Matcher matcher = OVERLAY_PATTERN.matcher(fieldName);
+                if (matcher.matches()) {
+                    overlayKeys.add(fieldName);
+                }
+            }
+
+            // Second pass: apply overlays
+            for (String overlayKey : overlayKeys) {
+                Matcher matcher = OVERLAY_PATTERN.matcher(overlayKey);
+                if (matcher.matches()) {
+                    String targetPath = matcher.group(1);
+                    JsonNode overlayValue = objectNode.get(overlayKey);
+                    // Resolve any <<...>> placeholders inside overlay values
+                    resolveOverlayPlaceholders(overlayValue, pipelineJson);
+                    applyOverlay(objectNode, targetPath, overlayValue);
+                    objectNode.remove(overlayKey);
+                }
+            }
+
+            // Recurse into remaining children
+            Iterator<Map.Entry<String, JsonNode>> fields = objectNode.fields();
+            while (fields.hasNext()) {
+                processOverlayDirectivesRecursive(fields.next().getValue(), pipelineJson);
+            }
+        } else if (node.isArray()) {
+            for (JsonNode element : node) {
+                processOverlayDirectivesRecursive(element, pipelineJson);
+            }
+        }
+    }
+
+    /**
+     * Resolves <<...>> placeholders in overlay value nodes using the pipeline config.
+     */
+    private void resolveOverlayPlaceholders(JsonNode node, String pipelineJson) {
+        if (node.isObject()) {
+            ObjectNode objectNode = (ObjectNode) node;
+            List<Map.Entry<String, JsonNode>> entries = new ArrayList<>();
+            objectNode.fields().forEachRemaining(entries::add);
+            for (Map.Entry<String, JsonNode> entry : entries) {
+                if (entry.getValue().isValueNode()) {
+                    String text = entry.getValue().asText();
+                    Matcher matcher = PLACEHOLDER_PATTERN.matcher(text);
+                    if (matcher.find()) {
+                        String placeholder = getValueFromPlaceHolder(text);
+                        String resolved = executeFunctionPlaceholder(placeholder, pipelineJson);
+                        if (isJsonPath(resolved)) {
+                            JsonNode resolvedNode = JsonPath.using(parseConfigWithJsonNode)
+                                    .parse(pipelineJson).read(resolved);
+                            if (resolvedNode != null && resolvedNode.isArray() && resolvedNode.size() == 1
+                                    && resolved.contains(JSON_PATH_ARRAY_DISAMBIGUATOR_PATTERN)) {
+                                resolvedNode = resolvedNode.get(0);
+                            }
+                            objectNode.set(entry.getKey(), resolvedNode);
+                        } else if (resolved != null) {
+                            objectNode.set(entry.getKey(), objectMapper.valueToTree(resolved));
+                        }
+                    }
+                } else {
+                    resolveOverlayPlaceholders(entry.getValue(), pipelineJson);
+                }
+            }
+        } else if (node.isArray()) {
+            for (JsonNode element : node) {
+                resolveOverlayPlaceholders(element, pipelineJson);
+            }
+        }
+    }
+
+    /**
+     * Applies overlay to the target path within the parent node.
+     * Supports paths like "sink[*].opensearch" where [*] means all array elements.
+     */
+    private void applyOverlay(ObjectNode parentNode, String targetPath, JsonNode overlayValue) {
+        // Split path into segments: e.g., "sink[*].opensearch" -> ["sink[*]", "opensearch"]
+        String[] segments = targetPath.split("\\.");
+        applyOverlayAtPath(parentNode, segments, 0, overlayValue);
+    }
+
+    private void applyOverlayAtPath(JsonNode currentNode, String[] segments, int index, JsonNode overlayValue) {
+        if (index >= segments.length) {
+            // Reached the target — deep merge overlay into current node
+            if (currentNode.isObject() && overlayValue.isObject()) {
+                deepMerge((ObjectNode) currentNode, (ObjectNode) overlayValue);
+            }
+            return;
+        }
+
+        String segment = segments[index];
+
+        if (segment.endsWith("[*]")) {
+            // Wildcard array segment: apply to all matching elements
+            String fieldName = segment.substring(0, segment.length() - 3);
+            JsonNode arrayNode = currentNode.get(fieldName);
+            if (arrayNode != null && arrayNode.isArray()) {
+                String nextSegment = (index + 1 < segments.length) ? segments[index + 1] : null;
+                for (JsonNode element : arrayNode) {
+                    if (nextSegment != null && element.isObject() && element.has(nextSegment)) {
+                        // Apply to the nested object (e.g., the "opensearch" object within each sink entry)
+                        applyOverlayAtPath(element.get(nextSegment), segments, index + 2, overlayValue);
+                    } else if (nextSegment == null) {
+                        applyOverlayAtPath(element, segments, index + 1, overlayValue);
+                    }
+                }
+            }
+        } else {
+            // Regular field traversal
+            JsonNode childNode = currentNode.get(segment);
+            if (childNode != null) {
+                applyOverlayAtPath(childNode, segments, index + 1, overlayValue);
+            }
+        }
+    }
+
+    /**
+     * Deep merges source into target. Source values override target values.
+     */
+    private void deepMerge(ObjectNode target, ObjectNode source) {
+        Iterator<Map.Entry<String, JsonNode>> fields = source.fields();
+        while (fields.hasNext()) {
+            Map.Entry<String, JsonNode> entry = fields.next();
+            String fieldName = entry.getKey();
+            JsonNode sourceValue = entry.getValue();
+
+            target.set(fieldName, sourceValue);
+        }
+    }
+
     /**
      * Replaces template node in the jsonPath with the node from
      * original json.
 
@@ -94,6 +94,47 @@ void test_isTransformationNeeded_ForOtherSource_ShouldReturn_False() throws IOEx
         assertFalse(result.isEvaluatedResult());
     }
 
+    @Test
+    void test_moreSpecificRule_matchesFirst_whenMultipleRulesApply() throws IOException {
+        // rds-joins rule has 2 conditions, rds rule has 1 condition.
+        // Both match an RDS pipeline with joins config.
+        // The more specific rule (rds-joins) should win.
+        String rdsRulePath = "src/test/resources/transformation/rules/rds-rule.yaml";
+        String rdsJoinsRulePath = "src/test/resources/transformation/rules/rds-joins-rule.yaml";
+        String templatePath = "src/test/resources/transformation/templates/testSource/rds-joins-template.yaml";
+
+        String pipelineName = "test-pipeline";
+        Map<String, Object> joinsConfig = new HashMap<>();
+        joinsConfig.put("version_field", "__versions");
+        Map<String, Object> sourceOptions = new HashMap<>();
+        sourceOptions.put("joins", joinsConfig);
+        final PluginModel source = new PluginModel("rds", sourceOptions);
+        final List<SinkModel> sinks = Collections.singletonList(
+                new SinkModel("testSink", Collections.emptyList(), null,
+                        Collections.emptyList(), Collections.emptyList(), null));
+        final PipelineModel pipelineModel = new PipelineModel(
+                source, null, Collections.emptyList(), null, sinks, 8, 50);
+        final PipelinesDataFlowModel pipelinesDataFlowModel = new PipelinesDataFlowModel(
+                (PipelineExtensions) null, Collections.singletonMap(pipelineName, pipelineModel));
+
+        TransformersFactory transformersFactory = mock(TransformersFactory.class);
+
+        // Load both rules — rds rule first (less specific), rds-joins second (more specific)
+        RuleStream rdsRule = new RuleStream("rds-rule.yaml", new FileInputStream(rdsRulePath));
+        RuleStream rdsJoinsRule = new RuleStream("rds-joins-rule.yaml", new FileInputStream(rdsJoinsRulePath));
+        when(transformersFactory.loadRules()).thenReturn(List.of(rdsRule, rdsJoinsRule));
+        when(transformersFactory.getPluginTemplateFileStream("rds-joins"))
+                .thenReturn(new FileInputStream(templatePath));
+
+        RuleEvaluator ruleEvaluator = new RuleEvaluator(transformersFactory);
+        RuleEvaluatorResult result = ruleEvaluator.isTransformationNeeded(pipelinesDataFlowModel);
+
+        assertTrue(result.isEvaluatedResult());
+        assertEquals(pipelineName, result.getPipelineName());
+        // The rds-joins template should be selected, not the generic rds template
+        assertTrue(result.getPipelineTemplateModel() != null);
+    }
+
     @Test
     void testThrowsExceptionOnFileError() {
         TransformersFactory transformersFactory = mock(TransformersFactory.class);