NIFI-1660 - Enhance the expression language with jsonPath function

nifi-commons/nifi-expression-language/pom.xml

@@ -61,5 +61,13 @@
             <artifactId>hamcrest-all</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.jayway.jsonpath</groupId>
+            <artifactId>json-path</artifactId>
+        </dependency>
+        <dependency>
+	        <groupId>com.fasterxml.jackson.core</groupId>
+	        <artifactId>jackson-databind</artifactId>
+        </dependency>
     </dependencies>
 </project>

nifi-commons/nifi-expression-language/src/main/antlr3/org/apache/nifi/attribute/expression/language/antlr/AttributeExpressionLexer.g

@@ -152,6 +152,7 @@ OR : 'or';
 AND : 'and';
 JOIN : 'join';
 TO_LITERAL : 'literal';
+JSON_PATH : 'jsonPath';
 
 // 2 arg functions
 SUBSTRING	: 'substring';

nifi-commons/nifi-expression-language/src/main/antlr3/org/apache/nifi/attribute/expression/language/antlr/AttributeExpressionParser.g

@@ -75,7 +75,7 @@ tokens {
 // functions that return Strings
 zeroArgString : (TO_UPPER | TO_LOWER | TRIM | TO_STRING | URL_ENCODE | URL_DECODE) LPAREN! RPAREN!;
 oneArgString : ((SUBSTRING_BEFORE | SUBSTRING_BEFORE_LAST | SUBSTRING_AFTER | SUBSTRING_AFTER_LAST | REPLACE_NULL | REPLACE_EMPTY |
-				PREPEND | APPEND | FORMAT | STARTS_WITH | ENDS_WITH | CONTAINS | JOIN) LPAREN! anyArg RPAREN!) |
+				PREPEND | APPEND | FORMAT | STARTS_WITH | ENDS_WITH | CONTAINS | JOIN | JSON_PATH) LPAREN! anyArg RPAREN!) |
 			   (TO_RADIX LPAREN! anyArg (COMMA! anyArg)? RPAREN!);
 twoArgString : ((REPLACE | REPLACE_FIRST | REPLACE_ALL) LPAREN! anyArg COMMA! anyArg RPAREN!) |
 			   (SUBSTRING LPAREN! anyArg (COMMA! anyArg)? RPAREN!);

nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/Query.java

@@ -59,6 +59,7 @@ import org.apache.nifi.attribute.expression.language.evaluation.functions.InEval
 import org.apache.nifi.attribute.expression.language.evaluation.functions.IndexOfEvaluator;
 import org.apache.nifi.attribute.expression.language.evaluation.functions.IsEmptyEvaluator;
 import org.apache.nifi.attribute.expression.language.evaluation.functions.IsNullEvaluator;
+import org.apache.nifi.attribute.expression.language.evaluation.functions.JsonPathEvaluator;
 import org.apache.nifi.attribute.expression.language.evaluation.functions.LastIndexOfEvaluator;
 import org.apache.nifi.attribute.expression.language.evaluation.functions.LengthEvaluator;
 import org.apache.nifi.attribute.expression.language.evaluation.functions.LessThanEvaluator;
@@ -152,6 +153,7 @@ import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpre
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.IS_EMPTY;
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.IS_NULL;
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.JOIN;
+import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.JSON_PATH;
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.LAST_INDEX_OF;
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.LENGTH;
 import static org.apache.nifi.attribute.expression.language.antlr.AttributeExpressionParser.LESS_THAN;
@@ -1352,9 +1354,15 @@ public class Query {
                         "getDelimitedField");
                 }
             }
-            default:
+            case JSON_PATH: {
-                throw new AttributeExpressionLanguageParsingException("Expected a Function-type expression but got " + tree.toString());
+                verifyArgCount(argEvaluators, 1, "jsonPath");
-        }
+                return addToken(new JsonPathEvaluator(toStringEvaluator(subjectEvaluator),
+                        toStringEvaluator(argEvaluators.get(0), "first argument to jsonPath")), "jsonPath");
+            }
+                default:
+                throw new AttributeExpressionLanguageParsingException(
+                        "Expected a Function-type expression but got " + tree.toString());
+            }
     }
 
     public static class Range {

nifi-commons/nifi-expression-language/src/main/java/org/apache/nifi/attribute/expression/language/evaluation/functions/JsonPathEvaluator.java

@@ -0,0 +1,127 @@
+/*
+ * 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.attribute.expression.language.evaluation.functions;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.nifi.attribute.expression.language.evaluation.Evaluator;
+import org.apache.nifi.attribute.expression.language.evaluation.QueryResult;
+import org.apache.nifi.attribute.expression.language.evaluation.StringEvaluator;
+import org.apache.nifi.attribute.expression.language.evaluation.StringQueryResult;
+import org.apache.nifi.attribute.expression.language.evaluation.literals.StringLiteralEvaluator;
+import org.apache.nifi.attribute.expression.language.exception.AttributeExpressionLanguageException;
+
+import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.InvalidJsonException;
+import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.spi.json.JacksonJsonProvider;
+import com.jayway.jsonpath.spi.json.JsonProvider;
+
+
+public class JsonPathEvaluator extends StringEvaluator {
+
+    private static final StringQueryResult EMPTY_RESULT = new StringQueryResult("");
+    private static final Configuration STRICT_PROVIDER_CONFIGURATION = Configuration.builder().jsonProvider(new JacksonJsonProvider()).build();
+    private static final JsonProvider JSON_PROVIDER = STRICT_PROVIDER_CONFIGURATION.jsonProvider();
+
+    private final Evaluator<String> subject;
+    private final Evaluator<String> jsonPathExp;
+    private final JsonPath precompiledJsonPathExp;
+
+    public JsonPathEvaluator(final Evaluator<String> subject, final Evaluator<String> jsonPathExp) {
+        this.subject = subject;
+        this.jsonPathExp = jsonPathExp;
+        // if the search string is a literal, we don't need to evaluate it each
+        // time; we can just
+        // pre-compile it. Otherwise, it must be compiled every time.
+        if (jsonPathExp instanceof StringLiteralEvaluator) {
+            precompiledJsonPathExp = compileJsonPathExpression(jsonPathExp.evaluate(null).getValue());
+        } else {
+            precompiledJsonPathExp = null;
+        }
+
+    }
+
+    @Override
+    public QueryResult<String> evaluate(final Map<String, String> attributes) {
+        final String subjectValue = subject.evaluate(attributes).getValue();
+        if (subjectValue == null || subjectValue.length() == 0) {
+            throw new  AttributeExpressionLanguageException("Subject is empty");
+        }
+        DocumentContext documentContext = null;
+        try {
+            documentContext = validateAndEstablishJsonContext(subjectValue);
+        } catch (InvalidJsonException e) {
+            throw new AttributeExpressionLanguageException("Subject contains invalid JSON: " + subjectValue, e);
+        }
+
+        final JsonPath compiledJsonPath;
+        if (precompiledJsonPathExp != null) {
+            compiledJsonPath = precompiledJsonPathExp;
+        } else {
+            compiledJsonPath = compileJsonPathExpression(jsonPathExp.evaluate(attributes).getValue());
+        }
+
+        Object result = null;
+        try {
+            result = documentContext.read(compiledJsonPath);
+        } catch (Exception e) {
+            // assume the path did not match anything in the document
+            return EMPTY_RESULT;
+        }
+
+        return new StringQueryResult(getResultRepresentation(result, EMPTY_RESULT.getValue()));
+    }
+
+
+    @Override
+    public Evaluator<?> getSubjectEvaluator() {
+        return subject;
+    }
+
+    static DocumentContext validateAndEstablishJsonContext(final String json) {
+        final DocumentContext ctx = JsonPath.using(STRICT_PROVIDER_CONFIGURATION).parse(json);
+        return ctx;
+    }
+
+    static boolean isJsonScalar(final Object obj) {
+        return !(obj instanceof Map || obj instanceof List);
+    }
+
+    static String getResultRepresentation(final Object jsonPathResult, final String defaultValue) {
+        if (isJsonScalar(jsonPathResult)) {
+            return Objects.toString(jsonPathResult, defaultValue);
+        } else if (jsonPathResult instanceof List && ((List<?>) jsonPathResult).size() == 1) {
+            return getResultRepresentation(((List<?>) jsonPathResult).get(0), defaultValue);
+        } else {
+            return JSON_PROVIDER.toJson(jsonPathResult);
+        }
+    }
+
+    static JsonPath compileJsonPathExpression(String exp) {
+        try {
+            return JsonPath.compile(exp);
+        } catch (Exception e) {
+            throw new AttributeExpressionLanguageException("Invalid JSON Path expression: " + exp, e);
+        }
+    }
+
+}
+

nifi-commons/nifi-expression-language/src/test/java/org/apache/nifi/attribute/expression/language/TestQuery.java

@@ -22,6 +22,10 @@ import static org.junit.Assert.assertTrue;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.MatcherAssert.assertThat;
 
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
 import java.text.SimpleDateFormat;
 import java.util.Calendar;
 import java.util.Collections;
@@ -236,6 +240,34 @@ public class TestQuery {
         assertEquals("true", Query.evaluateExpressions("${x:equals(\"${a}\")}", attributes, null));
     }
 
+    @Test
+    public void testJsonPath() {
+        final Map<String, String> attributes = new HashMap<>();
+        attributes.put("json", getResourceAsString("/json/address-book.json"));
+        verifyEquals("${json:jsonPath('$.firstName')}", attributes, "John");
+        verifyEquals("${json:jsonPath('$.address.postalCode')}", attributes, "10021-3100");
+        verifyEquals("${json:jsonPath(\"$.phoneNumbers[?(@.type=='home')].number\")}", attributes, "212 555-1234");
+        verifyEquals("${json:jsonPath('$.phoneNumbers')}", attributes,
+                "[{\"type\":\"home\",\"number\":\"212 555-1234\"},{\"type\":\"office\",\"number\":\"646 555-4567\"}]");
+        verifyEquals("${json:jsonPath('$.missing-path')}", attributes, "");
+        try {
+            verifyEquals("${json:jsonPath('$..')}", attributes, "");
+            Assert.fail("Did not detect bad JSON path expression");
+        } catch (final AttributeExpressionLanguageException e) {
+        }
+        try {
+            verifyEquals("${missing:jsonPath('$.firstName')}", attributes, "");
+            Assert.fail("Did not detect empty JSON document");
+        } catch (AttributeExpressionLanguageException e) {
+        }
+        attributes.put("invalid", "[}");
+        try {
+            verifyEquals("${invlaid:jsonPath('$.firstName')}", attributes, "John");
+            Assert.fail("Did not detect invalid JSON document");
+        } catch (AttributeExpressionLanguageException e) {
+        }
+    }
+
     @Test
     public void testJoin() {
         final Map<String, String> attributes = new HashMap<>();
@@ -1282,4 +1314,25 @@ public class TestQuery {
 
         assertEquals(expectedResult, result.getValue());
     }
+
+    private String getResourceAsString(String resourceName) {
+        Reader reader = new InputStreamReader(new BufferedInputStream(getClass().getResourceAsStream(resourceName)));
+        int n = 0;
+        char[] buf = new char[1024];
+        StringBuilder sb = new StringBuilder();
+        while (n != -1) {
+            try {
+                n = reader.read(buf, 0, buf.length);
+            } catch (IOException e) {
+                throw new RuntimeException("failed to read resource", e);
+            }
+            if (n > 0) {
+                sb.append(buf, 0, n);
+            }
+        }
+        return sb.toString();
+
+
+    }
+
 }

nifi-commons/nifi-expression-language/src/test/resources/json/address-book.json

@@ -0,0 +1,19 @@
+{
+     "firstName": "John", "lastName": "Smith", "age": 25,
+     "address" : {
+         "streetAddress": "21 2nd Street",
+         "city": "New York",
+         "state": "NY",
+         "postalCode": "10021-3100"
+     },
+     "phoneNumbers": [
+         { 
+             "type": "home", 
+             "number": "212 555-1234"
+         },
+         { 
+             "type": "office", 
+             "number": "646 555-4567"
+         }
+     ]
+ }

nifi-docs/src/main/asciidoc/expression-language-guide.adoc

@@ -1215,7 +1215,63 @@ Expressions will provide the following results:
 |=======================================================================================
 
 
+[.function]
+=== jsonPath
 
+*Description*: [.description]#The `jsonPath` function generates a string by evaluating the Subject as JSON and applying a JSON
+  path expression. An empty string is generated if the Subject does not contain valid JSON, the _jsonPath_ is invalid, or the path
+	does not exist in the Subject.  If the evaluation results in a scalar value, the string representation of scalar value is
+	generated.  Otherwise a string representation of the JSON result is generated.  A JSON array of length 1 is special cased
+	when `[0]` is a scalar, the string representation of `[0]` is generated.^1^#
+
+*Subject Type*: [.subject]#String#
+
+*Arguments*:
+	 [.argName]#_jsonPath_# : [.argDesc]#the JSON path expression used to evaluate the Subject.#
+
+*Return Type*: [.returnType]#String#
+
+*Examples*: If the "myJson" attribute is
+
+..........
+{
+  "firstName": "John",
+  "lastName": "Smith",
+  "isAlive": true,
+  "age": 25,
+  "address": {
+    "streetAddress": "21 2nd Street",
+    "city": "New York",
+    "state": "NY",
+    "postalCode": "10021-3100"
+  },
+  "phoneNumbers": [
+    {
+      "type": "home",
+      "number": "212 555-1234"
+    },
+    {
+      "type": "office",
+      "number": "646 555-4567"
+    }
+  ],
+  "children": [],
+  "spouse": null
+}
+..........
+
+.jsonPath Examples
+|===================================================================
+| Expression | Value
+| `${myJson:jsonPath('$.firstName')}` | `John`
+| `${myJson:jsonPath('$.address.postalCode')}` | `10021-3100`
+| `${myJson:jsonPath('$.phoneNumbers[?(@.type=="home")].number')}`^1^ | `212 555-1234`
+| `${myJson:jsonPath('$.phoneNumbers')}` | `[{"type":"home","number":"212 555-1234"},{"type":"office","number":"646 555-4567"}]`
+| `${myJson:jsonPath('$.missing-path')}` | _empty_
+| `${myJson:jsonPath('$.bad-json-path..')}` | _exception bulletin_
+|===================================================================
+
+An empty subject value or a subject value with an invalid JSON document results in an exception bulletin. 
 
 [[numbers]]
 == Mathematical Operations and Numeric Manipulation