diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java new file mode 100644 index 000000000000..e72841a96d96 --- /dev/null +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.elasticsearch; + +import org.apache.nifi.components.DescribedValue; + +/** + * Controls how the Identifier Field, Index Field, and Timestamp Field property values are + * interpreted when locating a field within a document. + */ +public enum FieldReferenceMode implements DescribedValue { + LITERAL("Literal Field Name", "Literal Field Name", + "The configured value is the exact name of a top-level field in the document; \"/\" and \"\\\" " + + "are treated as literal characters. This is the default and matches the behavior of earlier releases."), + NESTED_PATH("Nested Field Path", "Nested Field Path", + "The configured value is a \"/\"-delimited path into nested objects (e.g. \"@metadata/id\"). " + + "To reference a field name that contains a literal \"/\", escape it as \"\\/\"; a literal backslash is \"\\\\\"."); + + private final String value; + private final String displayName; + private final String description; + + FieldReferenceMode(final String value, final String displayName, final String description) { + this.value = value; + this.displayName = displayName; + this.description = description; + } + + @Override + public String getValue() { + return value; + } + + @Override + public String getDisplayName() { + return displayName; + } + + @Override + public String getDescription() { + return description; + } +} diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java index 29b05d854c14..1ee5c936b09d 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java @@ -237,10 +237,25 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { .dependsOn(INPUT_FORMAT, InputFormat.SINGLE_JSON) .build(); + static final PropertyDescriptor FIELD_PATH_MODE = new PropertyDescriptor.Builder() + .name("Field Path Mode") + .description(""" + How the Identifier Field, Index Field, and Timestamp Field property values are interpreted \ + when locating a field within each document. In "Literal Field Name" mode (the default) each \ + value is the exact name of a top-level field. In "Nested Field Path" mode each value is a \ + "/"-delimited path into nested objects; see the processor's Additional Details for the full \ + path syntax, including how to reference a field name that contains a literal "/".\ + """) + .required(true) + .allowableValues(FieldReferenceMode.class) + .defaultValue(FieldReferenceMode.LITERAL.getValue()) + .build(); + static final PropertyDescriptor IDENTIFIER_FIELD = new PropertyDescriptor.Builder() .name("Identifier Field") .description(""" - The name of the field within each document to use as the Elasticsearch document ID. \ + The name of the field within each document to use as the Elasticsearch document ID, \ + interpreted as a literal field name or a nested "/"-delimited path per the Field Path Mode property. \ If the field is not present in a document or this property is left blank, no document ID is set \ and Elasticsearch will auto-generate one.\ """) @@ -255,7 +270,8 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { .description(""" Whether to keep the Identifier Field in the document body after extracting it \ for use as the Elasticsearch document ID. \ - When true (default), the field is left in the document; set to false to remove it before indexing.\ + When true (default), the field is left in the document; set to false to remove it before indexing. \ + For a nested ("/"-delimited) path, any parent object left empty by the removal is also pruned.\ """) .required(true) .allowableValues("true", "false") @@ -266,7 +282,8 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { static final PropertyDescriptor INDEX_FIELD = new PropertyDescriptor.Builder() .name("Index Field") .description(""" - The name of the field within each document to use as the Elasticsearch index name. \ + The name of the field within each document to use as the Elasticsearch index name, \ + interpreted as a literal field name or a nested "/"-delimited path per the Field Path Mode property. \ If the field is not present in a document or this property is left blank, \ the configured Index property value is used as the fallback.\ """) @@ -280,7 +297,8 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { .description(""" Whether to keep the Index Field in the document body after extracting it \ for use as the Elasticsearch index name. \ - When true (default), the field is left in the document; set to false to remove it before indexing.\ + When true (default), the field is left in the document; set to false to remove it before indexing. \ + For a nested ("/"-delimited) path, any parent object left empty by the removal is also pruned.\ """) .required(true) .allowableValues("true", "false") @@ -292,7 +310,8 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { .name("Timestamp Field") .description(""" The name of a field within each document whose value will be written to \ - Elasticsearch as the @timestamp field. \ + Elasticsearch as the @timestamp field, interpreted as a literal field name or a nested \ + "/"-delimited path per the Field Path Mode property. \ If the field is absent or this property is left blank, no @timestamp is set.\ """) .required(false) @@ -305,7 +324,8 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { .description(""" Whether to keep the Timestamp Field in the document body after copying its \ value to @timestamp. \ - When true (default), the field is left in the document; set to false to remove it before indexing.\ + When true (default), the field is left in the document; set to false to remove it before indexing. \ + For a nested ("/"-delimited) path, any parent object left empty by the removal is also pruned.\ """) .required(true) .allowableValues("true", "false") @@ -341,6 +361,7 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch { MAX_BATCH_SIZE, SUPPRESS_NULLS, ID_ATTRIBUTE, + FIELD_PATH_MODE, IDENTIFIER_FIELD, RETAIN_IDENTIFIER_FIELD, INDEX_FIELD, @@ -466,11 +487,20 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final String idAttribute = inputFormat == InputFormat.SINGLE_JSON ? context.getProperty(ID_ATTRIBUTE).getValue() : null; + // In Literal Field Name mode the configured values are escaped into single-segment paths so the + // shared path helpers resolve them to the exact (verbatim) field name; in Nested Field Path mode + // they are already "/"-delimited paths and used as-is. + final boolean nestedFieldPaths = FieldReferenceMode.NESTED_PATH.getValue().equals(context.getProperty(FIELD_PATH_MODE).getValue()); final String documentIdField = inputFormat != InputFormat.SINGLE_JSON - ? context.getProperty(IDENTIFIER_FIELD).evaluateAttributeExpressions().getValue() + ? fieldPath(context.getProperty(IDENTIFIER_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths) : null; - final String documentIndexField = context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue(); - final String documentTimestampField = context.getProperty(TIMESTAMP_FIELD).evaluateAttributeExpressions().getValue(); + final String documentIndexField = fieldPath(context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths); + final String documentTimestampField = fieldPath(context.getProperty(TIMESTAMP_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths); + // The id/index field paths are loop-invariant, so classify them as nested-vs-flat and decode + // the flat names once here rather than per record in the NDJSON raw-bytes fast path below. + final boolean nestedExtractionField = isNestedPath(documentIdField) || isNestedPath(documentIndexField); + final String documentIdFieldName = documentIdField == null ? null : decodeSegment(documentIdField); + final String documentIndexFieldName = documentIndexField == null ? null : decodeSegment(documentIndexField); // The Retain toggles depend on their source field being set, so only read them when it is — // reading a property whose dependency is unsatisfied is invalid. When the source field is // blank no extraction or stripping occurs, so the retain value is irrelevant anyway. @@ -541,16 +571,18 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final boolean stripId = !retainIdentifierField && StringUtils.isNotBlank(documentIdField); final boolean stripIdx = !retainIndexField && StringUtils.isNotBlank(documentIndexField); final boolean needsTimestamp = StringUtils.isNotBlank(documentTimestampField); - if (suppressingWriter != null || stripId || stripIdx || needsTimestamp) { + // The raw streaming scan only matches flat field names; a nested + // (/-delimited) id or index path requires the parsed Map. + if (suppressingWriter != null || stripId || stripIdx || needsTimestamp || nestedExtractionField) { // Map is needed anyway — extract both fields from the Map directly. final Map contentMap = mapReader.readValue(trimmedLine); id = resolveId(contentMap, documentIdField, flowFileIdAttribute); docIndex = resolveIndex(contentMap, documentIndexField, index); if (stripId) { - contentMap.remove(documentIdField); + removeAtPath(contentMap, documentIdField); } if (stripIdx) { - contentMap.remove(documentIndexField); + removeAtPath(contentMap, documentIndexField); } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); rawJsonBytes = suppressingWriter != null @@ -558,7 +590,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session : mapper.writeValueAsBytes(contentMap); } else { // Raw-bytes path: single streaming scan finds both fields at once. - final String[] extracted = extractIdAndIndex(trimmedLine, documentIdField, flowFileIdAttribute, documentIndexField, index); + // The field names are pre-decoded and known to be flat (non-nested) here. + final String[] extracted = extractIdAndIndex(trimmedLine, documentIdFieldName, flowFileIdAttribute, documentIndexFieldName, index); id = extracted[0]; docIndex = extracted[1]; rawJsonBytes = trimmedLine.getBytes(StandardCharsets.UTF_8); @@ -580,10 +613,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final String id = resolveId(contentMap, documentIdField, flowFileIdAttribute); final String docIndex = resolveIndex(contentMap, documentIndexField, index); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { - contentMap.remove(documentIdField); + removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { - contentMap.remove(documentIndexField); + removeAtPath(contentMap, documentIndexField); } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); opRequest = IndexOperationRequest.builder() @@ -649,10 +682,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session id = resolveId(contentMap, documentIdField, flowFileIdAttribute); docIndex = resolveIndex(contentMap, documentIndexField, index); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { - contentMap.remove(documentIdField); + removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { - contentMap.remove(documentIndexField); + removeAtPath(contentMap, documentIndexField); } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); rawJsonBytes = suppressingWriter.writeValueAsBytes(contentMap); @@ -667,10 +700,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session if (node.isObject()) { final ObjectNode objectNode = (ObjectNode) node; if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { - objectNode.remove(documentIdField); + removeAtPath(objectNode, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { - objectNode.remove(documentIndexField); + removeAtPath(objectNode, documentIndexField); } applyTimestamp(objectNode, documentTimestampField, retainTimestampField); } @@ -695,10 +728,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final String id = resolveId(contentMap, documentIdField, flowFileIdAttribute); final String docIndex = resolveIndex(contentMap, documentIndexField, index); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { - contentMap.remove(documentIdField); + removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { - contentMap.remove(documentIndexField); + removeAtPath(contentMap, documentIndexField); } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); opRequest = IndexOperationRequest.builder() @@ -736,7 +769,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final String id = StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null; final String docIndex = resolveIndex(contentMap, documentIndexField, index); if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { - contentMap.remove(documentIndexField); + removeAtPath(contentMap, documentIndexField); } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); final IndexOperationRequest opRequest = IndexOperationRequest.builder() @@ -1029,36 +1062,38 @@ private void outputBulkRequestFlowFile(final List operati } /** - * Copies the value of {@code timestampField} to {@code @timestamp} in the Map. - * If {@code retain} is false, the source field is removed after copying. - * Does nothing when {@code timestampField} is blank or not present in the document. + * Copies the value at {@code timestampField} to a top-level {@code @timestamp} in the Map. + * The field may be a {@code /}-delimited path into nested objects (e.g. {@code @metadata/event_time}). + * If {@code retain} is false, the source field is removed after copying, pruning any ancestor + * objects that become empty. Does nothing when {@code timestampField} is blank or absent. */ private void applyTimestamp(final Map contentMap, final String timestampField, final boolean retain) { if (StringUtils.isBlank(timestampField)) { return; } - final Object value = contentMap.get(timestampField); + final Object value = valueAtPath(contentMap, timestampField); if (value != null) { if (!retain) { - contentMap.remove(timestampField); + removeAtPath(contentMap, timestampField); } contentMap.put("@timestamp", value); } } /** - * Copies the value of {@code timestampField} to {@code @timestamp} in the ObjectNode. - * If {@code retain} is false, the source field is removed after copying. - * Does nothing when {@code timestampField} is blank or not present in the document. + * Copies the value at {@code timestampField} to a top-level {@code @timestamp} in the ObjectNode. + * The field may be a {@code /}-delimited path into nested objects (e.g. {@code @metadata/event_time}). + * If {@code retain} is false, the source field is removed after copying, pruning any ancestor + * objects that become empty. Does nothing when {@code timestampField} is blank or absent. */ private void applyTimestamp(final ObjectNode node, final String timestampField, final boolean retain) { if (StringUtils.isBlank(timestampField)) { return; } - final JsonNode value = node.get(timestampField); + final JsonNode value = nodeAtPath(node, timestampField); if (value != null && !value.isNull()) { if (!retain) { - node.remove(timestampField); + removeAtPath(node, timestampField); } node.set("@timestamp", value); } @@ -1069,13 +1104,16 @@ private void applyTimestamp(final ObjectNode node, final String timestampField, * Stops as soon as both fields have been found to avoid scanning the rest of the document. * Returns a two-element array: {@code [id, docIndex]}. * Used for NDJSON Index/Create when neither suppression nor field-removal requires a Map parse. + *

+ * {@code idFieldName} and {@code indexFieldName} are the already-decoded flat field names (any + * escaped slash/backslash resolved by the caller), compared directly against the raw JSON keys. */ private String[] extractIdAndIndex( final String rawJson, - final String idField, final String flowFileIdAttribute, - final String indexField, final String fallbackIndex) throws IOException { - final boolean needId = StringUtils.isNotBlank(idField); - final boolean needIndex = StringUtils.isNotBlank(indexField); + final String idFieldName, final String flowFileIdAttribute, + final String indexFieldName, final String fallbackIndex) throws IOException { + final boolean needId = StringUtils.isNotBlank(idFieldName); + final boolean needIndex = StringUtils.isNotBlank(indexFieldName); String id = needId ? null : (StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null); String docIndex = fallbackIndex; @@ -1092,11 +1130,11 @@ private String[] extractIdAndIndex( if (foundId && foundIndex) { break; } - if (!foundId && idField.equals(p.currentName()) && p.nextToken() != null && isScalarValue(p.currentToken())) { + if (!foundId && idFieldName.equals(p.currentName()) && p.nextToken() != null && isScalarValue(p.currentToken())) { final String value = p.getText(); id = StringUtils.isNotBlank(value) ? value : (StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null); foundId = true; - } else if (!foundIndex && indexField.equals(p.currentName()) && p.nextToken() != null && isScalarValue(p.currentToken())) { + } else if (!foundIndex && indexFieldName.equals(p.currentName()) && p.nextToken() != null && isScalarValue(p.currentToken())) { final String value = p.getText(); docIndex = StringUtils.isNotBlank(value) ? value : fallbackIndex; foundIndex = true; @@ -1113,12 +1151,13 @@ private String[] extractIdAndIndex( /** * Extracts the document ID from a pre-parsed JsonNode. * Used for JSON Array Index/Create operations where the node is already available. + * The field may be a {@code /}-delimited path into nested objects. */ private String extractId(final JsonNode node, final String idAttribute, final String flowFileIdAttribute) { if (StringUtils.isBlank(idAttribute)) { return StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null; } - final String value = fieldNodeToString(node.get(idAttribute)); + final String value = fieldNodeToString(nodeAtPath(node, idAttribute)); if (StringUtils.isNotBlank(value)) { return value; } @@ -1127,13 +1166,14 @@ private String extractId(final JsonNode node, final String idAttribute, final St /** * Resolves the document ID for Update/Delete/Upsert operations from the already-parsed content Map. + * The field may be a {@code /}-delimited path into nested objects. * Falls back to the FlowFile attribute value when the field is absent from the document. */ private String resolveId(final Map contentMap, final String idAttribute, final String flowFileIdAttribute) { if (StringUtils.isBlank(idAttribute)) { return null; } - final String value = fieldValueToString(contentMap.get(idAttribute)); + final String value = fieldValueToString(valueAtPath(contentMap, idAttribute)); if (StringUtils.isNotBlank(value)) { return value; } @@ -1143,30 +1183,279 @@ private String resolveId(final Map contentMap, final String idAt /** * Extracts the index name from a pre-parsed {@link JsonNode}. * Used for JSON Array Index/Create operations where the node is already available. + * The field may be a {@code /}-delimited path into nested objects. * Falls back to {@code fallbackIndex} when the field is absent or blank. */ private String extractIndex(final JsonNode node, final String indexField, final String fallbackIndex) { if (StringUtils.isBlank(indexField)) { return fallbackIndex; } - final String value = fieldNodeToString(node.get(indexField)); + final String value = fieldNodeToString(nodeAtPath(node, indexField)); return StringUtils.isNotBlank(value) ? value : fallbackIndex; } /** * Resolves the index name from an already-parsed content Map. * Used for Update/Delete/Upsert operations and suppression-enabled Index/Create paths - * where the Map is already available. Falls back to {@code fallbackIndex} when the - * field is absent or blank. + * where the Map is already available. The field may be a {@code /}-delimited path into + * nested objects. Falls back to {@code fallbackIndex} when the field is absent or blank. */ private String resolveIndex(final Map contentMap, final String indexField, final String fallbackIndex) { if (StringUtils.isBlank(indexField)) { return fallbackIndex; } - final String value = fieldValueToString(contentMap.get(indexField)); + final String value = fieldValueToString(valueAtPath(contentMap, indexField)); return StringUtils.isNotBlank(value) ? value : fallbackIndex; } + /** + * Normalizes a configured Identifier/Index/Timestamp Field value into the path form consumed by the + * shared path helpers. In Nested Field Path mode the value is already a {@code /}-delimited path and + * is returned unchanged. In Literal Field Name mode the value names a single field verbatim, so any + * {@code /} or {@code \} is escaped ({@code \} to {@code \\} then {@code /} to {@code \/}) to produce + * a single-segment path that resolves back to that exact field name. + */ + private static String fieldPath(final String configuredValue, final boolean nestedFieldPaths) { + if (configuredValue == null || nestedFieldPaths) { + return configuredValue; + } + return configuredValue.replace("\\", "\\\\").replace("/", "\\/"); + } + + /** + * Whether {@code field} is a {@code /}-delimited nested path rather than a flat field name. + * A {@code /} preceded by a backslash is an escaped literal slash within a single field name + * (see {@link #splitPath(String)}) and does not make the field a nested path. + */ + private static boolean isNestedPath(final String field) { + if (field == null) { + return false; + } + for (int i = 0; i < field.length(); i++) { + final char c = field.charAt(i); + if (c == '\\' && i + 1 < field.length()) { + final char next = field.charAt(i + 1); + if (next == '/' || next == '\\') { + i++; // skip the escaped character + continue; + } + } else if (c == '/') { + return true; + } + } + return false; + } + + /** + * Splits a {@code /}-delimited field path into its decoded segments, ignoring empty segments + * produced by leading, trailing, or repeated slashes. A field name containing no unescaped + * slash yields a single segment equal to the (decoded) name itself. + *

+ * A backslash escapes the following character so it is taken literally: {@code \/} is a slash + * that is part of a field name rather than a separator, and {@code \\} is a literal backslash. + * A backslash before any other character (or at the end of the string) is kept as a literal + * backslash, so field names containing stray backslashes are unaffected. + */ + private static String[] splitPath(final String path) { + final List segments = new ArrayList<>(); + final StringBuilder current = new StringBuilder(path.length()); + for (int i = 0; i < path.length(); i++) { + final char c = path.charAt(i); + if (c == '\\' && i + 1 < path.length()) { + final char next = path.charAt(i + 1); + if (next == '/' || next == '\\') { + current.append(next); + i++; // consume the escaped character + continue; + } + current.append(c); // lone backslash, kept literal + } else if (c == '/') { + if (current.length() > 0) { + segments.add(current.toString()); + current.setLength(0); + } + } else { + current.append(c); + } + } + if (current.length() > 0) { + segments.add(current.toString()); + } + return segments.toArray(new String[0]); + } + + /** + * Decodes a single flat field name, resolving the same backslash escapes as + * {@link #splitPath(String)} ({@code \/} to {@code /}, {@code \\} to {@code \}) without treating + * any slash as a separator. Returns the input unchanged when it contains no backslash. + */ + private static String decodeSegment(final String field) { + if (field.indexOf('\\') < 0) { + return field; + } + final StringBuilder sb = new StringBuilder(field.length()); + for (int i = 0; i < field.length(); i++) { + final char c = field.charAt(i); + if (c == '\\' && i + 1 < field.length()) { + final char next = field.charAt(i + 1); + if (next == '/' || next == '\\') { + sb.append(next); + i++; + continue; + } + } + sb.append(c); + } + return sb.toString(); + } + + /** + * Returns the value at the {@code /}-delimited {@code path} within {@code map}, or {@code null} + * when any segment is missing or an intermediate segment is not a JSON object. A path with no + * slash is a direct top-level lookup (no allocation). + */ + private static Object valueAtPath(final Map map, final String path) { + if (!isNestedPath(path)) { + return map.get(decodeSegment(path)); + } + Object current = map; + for (final String segment : splitPath(path)) { + if (!(current instanceof Map)) { + return null; + } + current = ((Map) current).get(segment); + if (current == null) { + return null; + } + } + return current; + } + + /** + * Returns the {@link JsonNode} at the {@code /}-delimited {@code path} within {@code node}, or + * {@code null} when any segment is missing or an intermediate segment is not a JSON object. + * A path with no slash is a direct field lookup. + */ + private static JsonNode nodeAtPath(final JsonNode node, final String path) { + if (!isNestedPath(path)) { + return node.get(decodeSegment(path)); + } + JsonNode current = node; + for (final String segment : splitPath(path)) { + if (current == null || !current.isObject()) { + return null; + } + current = current.get(segment); + } + return current; + } + + /** + * Removes the leaf at the {@code /}-delimited {@code path} within {@code root}, then prunes any + * ancestor objects that become empty as a result (the transient envelope often used for routing + * metadata, e.g. {@code @metadata}, should not be indexed once its fields are extracted). + * Does nothing when the path is absent. A path with no slash is a direct top-level removal. + */ + private static void removeAtPath(final Map root, final String path) { + if (!isNestedPath(path)) { + root.remove(decodeSegment(path)); + return; + } + removeAtPath(root, splitPath(path), MAP_ACCESSOR); + } + + /** + * Removes the leaf at the {@code /}-delimited {@code path} within {@code root}, then prunes any + * ancestor objects that become empty as a result. Does nothing when the path is absent. + * A path with no slash is a direct field removal. + */ + private static void removeAtPath(final ObjectNode root, final String path) { + if (!isNestedPath(path)) { + root.remove(decodeSegment(path)); + return; + } + removeAtPath(root, splitPath(path), OBJECT_NODE_ACCESSOR); + } + + /** + * Shared descend-and-prune walk for both the {@link Map} and {@link ObjectNode} object models. + * Descends the parent {@code segments}, removes the leaf, then walks back up removing each + * ancestor that the removal left empty. The {@link NodeAccessor} isolates the few model-specific + * operations so the traversal logic exists in exactly one place. + */ + private static void removeAtPath(final T root, final String[] segments, final NodeAccessor accessor) { + final List chain = new ArrayList<>(); + T current = root; + chain.add(current); + for (int i = 0; i < segments.length - 1; i++) { + final T next = accessor.childObject(current, segments[i]); + if (next == null) { + return; + } + current = next; + chain.add(current); + } + accessor.remove(chain.get(chain.size() - 1), segments[segments.length - 1]); + for (int i = chain.size() - 1; i >= 1; i--) { + if (accessor.isEmpty(chain.get(i))) { + accessor.remove(chain.get(i - 1), segments[i - 1]); + } else { + break; + } + } + } + + /** + * The model-specific operations the shared {@link #removeAtPath(Object, String[], NodeAccessor)} + * walk needs: fetching a child only when it is itself an object, removing a key, and testing + * emptiness. + */ + private interface NodeAccessor { + /** The child object at {@code key}, or {@code null} when absent or not itself an object. */ + T childObject(T node, String key); + + void remove(T node, String key); + + boolean isEmpty(T node); + } + + private static final NodeAccessor> MAP_ACCESSOR = new NodeAccessor<>() { + @Override + @SuppressWarnings("unchecked") + public Map childObject(final Map node, final String key) { + final Object next = node.get(key); + return next instanceof Map ? (Map) next : null; + } + + @Override + public void remove(final Map node, final String key) { + node.remove(key); + } + + @Override + public boolean isEmpty(final Map node) { + return node.isEmpty(); + } + }; + + private static final NodeAccessor OBJECT_NODE_ACCESSOR = new NodeAccessor<>() { + @Override + public ObjectNode childObject(final ObjectNode node, final String key) { + final JsonNode next = node.get(key); + return next != null && next.isObject() ? (ObjectNode) next : null; + } + + @Override + public void remove(final ObjectNode node, final String key) { + node.remove(key); + } + + @Override + public boolean isEmpty(final ObjectNode node) { + return node.isEmpty(); + } + }; + /** * Converts a document field value taken from a parsed content Map into its string form for use * as a document ID or index name. Returns {@code null} for absent values and for container diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md index a80d3c4290f3..d2da2f79302a 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md @@ -30,6 +30,62 @@ failed to an errors queue so that only failed FlowFiles can be processed downstr The index, operation and (optional) type fields are configured with default values. The ID (optional unless the operation is "index") can be set as an attribute on the FlowFile(s). +### Field Path Mode + +The **Identifier Field**, **Index Field** and **Timestamp Field** properties each name a field within the document +whose value is extracted (and used as the document ID, index name, or `@timestamp` respectively). The **Field Path +Mode** property controls how those values are interpreted: + +* **Literal Field Name** (the default) — each value is the exact name of a top-level field. A `/` or `\` is just + a character in the field name, so a value like `@metadata/id` matches a top-level field literally named + `@metadata/id`. +* **Nested Field Path** — each value is a `/`-delimited path into nested objects, as described below. + +#### Literal Field Name (default) + +For the document: + +```json +{ + "@metadata/id": "abc", + "message": "Hello, world" +} +``` + +an Identifier Field of `@metadata/id` selects `abc` — the top-level field whose name is literally +`@metadata/id`. The same value against `{"@metadata": {"id": "abc"}}` selects nothing, because there is no +top-level field with that exact name. + +#### Nested Field Path + +The remainder of this section applies when **Field Path Mode** is set to **Nested Field Path**. For the document: + +```json +{ + "@metadata": { + "id": "abc", + "index": "my-index" + }, + "message": "Hello, world" +} +``` + +* the Identifier Field path `@metadata/id` selects `abc` +* the Index Field path `@metadata/index` selects `my-index` + +#### Field names that contain a slash + +Because `/` separates path segments, a field whose name literally contains a `/` must have that slash escaped as `\/`, +and a literal backslash must be escaped as `\\`. For example, for the document `{"a/b": "abc"}` the path `a\/b` selects +the top-level field named `a/b`. + +#### Removal and pruning + +When the corresponding **Retain** property is set to `false`, the field is removed from the document body after its +value is extracted. For a nested path, any parent object that is left empty by the removal is also pruned. For example, +extracting and removing `@metadata/id` from `{"@metadata": {"id": "abc"}, "message": "Hello, world"}` leaves +`{"message": "Hello, world"}`, with the now-empty `@metadata` object removed. + ### Dynamic Templates Index and Create operations can use Dynamic Templates. The Dynamic Templates property must be parsable as a JSON object. diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java index b93fe826720c..c61efe132fbd 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java @@ -43,6 +43,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -903,6 +904,47 @@ private static String docContent(final IndexOperationRequest item) { return item.getFields() == null ? "" : item.getFields().toString(); } + /** Switches the processor to Nested Field Path mode (the field properties default to Literal Field Name). */ + private void useNestedFieldPaths() { + runner.setProperty(PutElasticsearchJson.FIELD_PATH_MODE, FieldReferenceMode.NESTED_PATH.getValue()); + } + + @Test + void testDocumentFieldNamesWithSlashAndBackslashPassThroughRaw() { + // Only the configured Identifier/Index/Timestamp Field property values are parsed as nested + // paths; field names in the document body are never interpreted. With no extraction configured, + // the raw bytes -- including field names that contain "/" and "\" -- are passed through unchanged. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.assertValid(); + + final List ops = captureOperations(); + final String doc = "{\"a/b\":\"x\",\"c\\\\d\":\"y\",\"msg\":\"hello\"}"; + runner.enqueue(doc + "\n"); + runner.run(); + + runner.assertTransferCount(AbstractPutElasticsearch.REL_SUCCESSFUL, 1); + assertEquals(doc, docContent(ops.getFirst()), "document body passed through byte-for-byte"); + } + + @Test + void testUnrelatedSlashFieldRetainedWhenExtractingId() { + // Even on the parse-and-re-serialize path (here forced by removing the extracted id field), a + // document field whose name contains a "/" is not the configured path and is left untouched. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "doc_id"); + runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"doc_id\":\"1\",\"a/b\":\"x\",\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("1", ops.getFirst().getId()); + final String content = docContent(ops.getFirst()); + assertTrue(content.contains("\"a/b\""), "unrelated slash-named field retained"); + assertFalse(content.contains("doc_id"), "only the configured field is removed"); + } + @Test void testIndexFieldNdjsonExtractionRetainedByDefault() { runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); @@ -1110,4 +1152,388 @@ void testIndexFieldNullValueFallsBackConsistently() { runner.run(); assertEquals("test_index", ops.getFirst().getIndex(), "NDJSON null -> fallback (not literal \"null\")"); } + + // ------------------------------------------------------------------------- + // Field Path Mode: Literal Field Name (the default) + // ------------------------------------------------------------------------- + + @Test + void testFieldPathModeDefaultsToLiteral() { + assertEquals(FieldReferenceMode.LITERAL.getValue(), PutElasticsearchJson.FIELD_PATH_MODE.getDefaultValue()); + } + + @Test + void testLiteralModeTreatsSlashAsLiteralFieldName() { + // Default (Literal Field Name) mode: "@metadata/id" names one top-level field, so a document whose + // id is nested under @metadata is NOT matched -- this is the pre-nested-path behavior preserved on upgrade. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "@metadata/id"); + runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n"); + runner.run(); + + assertNull(ops.getFirst().getId(), "literal mode does not descend into @metadata"); + assertTrue(docContent(ops.getFirst()).contains("@metadata"), "nested object untouched in literal mode"); + } + + @Test + void testLiteralModeMatchesFieldNameContainingSlash() { + // In literal mode the value names the field verbatim, including a "/", with no escaping required. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}\n"); + runner.run(); + + assertEquals("docs-x", ops.getFirst().getIndex(), "literal field named a/b matched verbatim"); + assertFalse(docContent(ops.getFirst()).contains("a/b"), "literal field removed"); + } + + // ------------------------------------------------------------------------- + // Nested field paths (e.g. @metadata/index) + // ------------------------------------------------------------------------- + + @Test + void testNestedIndexFieldNdjsonPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\"},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("docs-2026", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertFalse(content.contains("@metadata"), "now-empty parent should be pruned"); + assertTrue(content.contains("hello")); + } + + @Test + void testNestedIndexFieldJsonArrayPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.JSON_ARRAY.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("[{\"@metadata\":{\"index\":\"docs-a\"},\"msg\":\"x\"}]"); + runner.run(); + + assertEquals("docs-a", ops.getFirst().getIndex()); + assertFalse(docContent(ops.getFirst()).contains("@metadata")); + } + + @Test + void testNestedIndexFieldSingleJsonPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-single\"},\"msg\":\"x\"}"); + runner.run(); + + assertEquals("docs-single", ops.getFirst().getIndex()); + assertFalse(ops.getFirst().getFields().containsKey("@metadata")); + } + + @Test + void testNestedIndexFieldKeepsParentWithRemainingSiblings() { + useNestedFieldPaths(); + // @metadata still holds another field after the index leaf is removed, so it must NOT be pruned + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\",\"keep\":\"yes\"},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("docs-2026", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertTrue(content.contains("@metadata"), "parent kept because it still has siblings"); + assertTrue(content.contains("keep")); + assertFalse(content.contains("\"index\"")); + } + + @Test + void testNestedIndexFieldRetainedByDefault() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\"},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("docs-2026", ops.getFirst().getIndex()); + assertTrue(docContent(ops.getFirst()).contains("@metadata"), "nested field retained by default"); + } + + @Test + void testNestedIdentifierFieldNdjsonPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "@metadata/id"); + runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n"); + runner.run(); + + assertEquals("abc-123", ops.getFirst().getId()); + assertFalse(docContent(ops.getFirst()).contains("@metadata")); + } + + @Test + void testNestedTimestampFieldNdjsonPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "@metadata/event_time"); + runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"event_time\":\"2026-01-02T03:04:05Z\"},\"msg\":\"x\"}\n"); + runner.run(); + + final String content = docContent(ops.getFirst()); + assertTrue(content.contains("@timestamp")); + assertTrue(content.contains("2026-01-02T03:04:05Z")); + assertFalse(content.contains("@metadata"), "empty parent pruned after extracting nested timestamp"); + } + + @Test + void testNestedTimestampFieldSingleJsonPrunesEmptyParent() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "meta/ts"); + runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"meta\":{\"ts\":\"2026-05-31T00:00:00Z\"},\"msg\":\"x\"}"); + runner.run(); + + final Map fields = ops.getFirst().getFields(); + assertEquals("2026-05-31T00:00:00Z", fields.get("@timestamp")); + assertFalse(fields.containsKey("meta"), "empty parent pruned"); + } + + @Test + void testNestedIndexFieldMissingLeafFallsBack() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"other\":\"x\"},\"msg\":\"y\"}\n"); + runner.run(); + + assertEquals("test_index", ops.getFirst().getIndex(), "missing nested leaf -> fall back to Index property"); + } + + @Test + void testNestedIndexFieldConsistentAcrossFormats() { + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "@metadata/index"); + List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "Single JSON nested path"); + runner.clearTransferState(); + + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.JSON_ARRAY.getValue()); + ops = captureOperations(); + runner.enqueue("[{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}]"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "JSON Array nested path"); + runner.clearTransferState(); + + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}\n"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "NDJSON nested path"); + } + + @Test + void testDeeplyNestedIndexFieldPrunesAllEmptyAncestors() { + useNestedFieldPaths(); + // A 3-level path (a/b/c): removing the leaf must recursively prune b and then a, + // since each ancestor becomes empty in turn. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b/c"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a\":{\"b\":{\"c\":\"docs-deep\"}},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("docs-deep", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertFalse(content.contains("\"a\""), "all empty ancestor objects pruned recursively"); + assertFalse(content.contains("\"b\"")); + assertFalse(content.contains("\"c\"")); + assertTrue(content.contains("hello")); + } + + @Test + void testDeeplyNestedIndexFieldKeepsAncestorWithRemainingSibling() { + useNestedFieldPaths(); + // A 3-level path (a/b/c) where the middle object 'b' retains a sibling after the leaf is + // removed: pruning must stop at 'b' and leave 'a' intact. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b/c"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a\":{\"b\":{\"c\":\"docs-deep\",\"keep\":\"yes\"}},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("docs-deep", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertTrue(content.contains("\"keep\""), "sibling stops the recursive pruning"); + assertTrue(content.contains("\"b\"")); + assertTrue(content.contains("\"a\"")); + assertFalse(content.contains("\"c\"")); + } + + @Test + void testNestedIdentifierFieldUpdateOperationResolvesAndPrunes() { + useNestedFieldPaths(); + // Update/Delete/Upsert resolve the ID from the parsed content Map via resolveId, a different + // code path than the Index/Create streaming/JsonNode extraction. Verify the nested path is + // honored and the now-empty parent is pruned from the doc body. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_OP, IndexOperationRequest.Operation.Update.getValue().toLowerCase()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "@metadata/id"); + runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n"); + runner.run(); + + assertEquals("abc-123", ops.getFirst().getId()); + assertFalse(ops.getFirst().getFields().containsKey("@metadata"), "empty parent pruned after extracting nested id"); + assertTrue(ops.getFirst().getFields().containsKey("msg")); + } + + // ------------------------------------------------------------------------- + // Escaped path separators (e.g. a\/b addresses the flat field named "a/b") + // ------------------------------------------------------------------------- + + @Test + void testEscapedSlashIndexFieldTreatedAsFlatNameAcrossFormats() { + useNestedFieldPaths(); + // "a\/b" must resolve the top-level field literally named "a/b" on every path, + // including the NDJSON raw-bytes streaming scan. + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b"); + List ops = captureOperations(); + runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "Single JSON escaped-slash flat field"); + runner.clearTransferState(); + + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.JSON_ARRAY.getValue()); + ops = captureOperations(); + runner.enqueue("[{\"a/b\":\"docs-x\",\"msg\":\"m\"}]"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "JSON Array escaped-slash flat field"); + runner.clearTransferState(); + + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + ops = captureOperations(); + runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}\n"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "NDJSON escaped-slash flat field (raw streaming path)"); + } + + @Test + void testEscapedSlashIndexFieldRemovedWhenRetainFalse() { + useNestedFieldPaths(); + // The escaped-slash field is flat, so removal must delete the top-level "a/b" key and must + // NOT descend into a nonexistent "a" object. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a/b\":\"docs-x\",\"keep\":\"y\"}\n"); + runner.run(); + + assertEquals("docs-x", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertFalse(content.contains("a/b"), "escaped-slash flat field removed"); + assertTrue(content.contains("keep")); + } + + @Test + void testNestedPathWithEscapedSlashSegment() { + useNestedFieldPaths(); + // "a\/b/c": the first segment is the literal field "a/b", then nested field "c". + // Removing the leaf prunes the now-empty "a/b" object. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b/c"); + runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a/b\":{\"c\":\"docs-x\"},\"msg\":\"m\"}\n"); + runner.run(); + + assertEquals("docs-x", ops.getFirst().getIndex()); + final String content = docContent(ops.getFirst()); + assertFalse(content.contains("a/b"), "now-empty object named \"a/b\" pruned"); + assertTrue(content.contains("msg")); + } + + @Test + void testEscapedBackslashFieldName() { + useNestedFieldPaths(); + // "a\\b" addresses the flat field literally named "a\b" (single backslash). + runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\\\b"); + final List ops = captureOperations(); + runner.enqueue("{\"a\\\\b\":\"docs-x\",\"msg\":\"m\"}"); + runner.run(); + assertEquals("docs-x", ops.getFirst().getIndex(), "\\\\ resolves to a single literal backslash in the field name"); + } + + @Test + void testEscapedSlashIdentifierFieldUpdateOperation() { + useNestedFieldPaths(); + // Update resolves the ID from the parsed Map via resolveId; verify the escaped-slash flat + // field name is honored and removed. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_OP, IndexOperationRequest.Operation.Update.getValue().toLowerCase()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "a\\/b"); + runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"a/b\":\"id-1\",\"msg\":\"m\"}\n"); + runner.run(); + + assertEquals("id-1", ops.getFirst().getId()); + assertFalse(ops.getFirst().getFields().containsKey("a/b"), "escaped-slash flat id field removed"); + assertTrue(ops.getFirst().getFields().containsKey("msg")); + } }