Add more utility functions (#117)

README.md

@@ -289,6 +289,9 @@ The Expression Language supports the following functions:
 * `coalesce(value, valueIfNull)`: Returns `value` if it is not `null`, otherwise returns `valueIfNull`.
 * `replace(input, regex, replacement)`: Replaces each substring of `input` that matches the `regex` regular expression with `replacement`. See [Java's replaceAll](https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/lang/String.html#replaceAll(java.lang.String,java.lang.String)).
 * `str(input)`: Converts `input` to a string.
+* `toJson(input)`: Converts `input` to a JSON string.
+* `fromJson(input)`: Parse `input` as JSON.
+* `split(input, separatorExpression)`: Split the input to a list of strings, this is internally using the String.split() function. An empty input corresponds to an empty list. The input is convered to a String using the str() function.
 * `now()`: Returns the current timestamp.
 * `timestampAdd(input, delta, unit)`: Returns a timestamp formed by adding `delta` in `unit` to the `input` timestamp. 
   * `input` a timestamp to add to.
@@ -302,6 +305,7 @@ The Expression Language supports the following functions:
 * `filter(collection, expression)`: Returns a new collection containing only the elements of `collection` for which `expression` is `true`. The current element is available under the `record` variable. An example is fn:filter(value.queryResults, "fn:toDouble(record.similarity) >= 0.5")
 For all methods, if a parameter is not in the right type, a conversion will be done using the rules described in [Type conversions](#type-conversions).
 For instance, you can do `fn:timestampAdd('2022-10-02T01:02:03Z', '42', 'hours'.bytes)`
+* `unpack(input, fieldsList)`: Returns a map containing the elements of `input`, for each field in the `fieldList` you will see an entry in the map. If the input is a string it is converted to a list using the `split()` function with the ',' separator
 
 When a function returns a timestamp, its type is `INSTANT`.
 

streaming-ai/src/main/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluator.java

@@ -37,6 +37,20 @@ public JstlEvaluator(String expression, Class<? extends T> type) {
 
   @SneakyThrows
   private void registerFunctions() {
+    this.expressionContext
+        .getFunctionMapper()
+        .mapFunction("fn", "toJson", JstlFunctions.class.getMethod("toJson", Object.class));
+    this.expressionContext
+        .getFunctionMapper()
+        .mapFunction("fn", "fromJson", JstlFunctions.class.getMethod("fromJson", Object.class));
+    this.expressionContext
+        .getFunctionMapper()
+        .mapFunction(
+            "fn", "split", JstlFunctions.class.getMethod("split", Object.class, Object.class));
+    this.expressionContext
+        .getFunctionMapper()
+        .mapFunction(
+            "fn", "unpack", JstlFunctions.class.getMethod("unpack", Object.class, Object.class));
     this.expressionContext
         .getFunctionMapper()
         .mapFunction("fn", "uppercase", JstlFunctions.class.getMethod("uppercase", Object.class));

streaming-ai/src/main/java/com/datastax/oss/streaming/ai/jstl/JstlFunctions.java

@@ -17,6 +17,7 @@
 
 import com.datastax.oss.streaming.ai.TransformContext;
 import com.datastax.oss.streaming.ai.jstl.predicate.JstlPredicate;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.el.ELException;
 import java.lang.reflect.Array;
 import java.math.BigDecimal;
@@ -26,14 +27,20 @@
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.el.util.MessageFactory;
 
 /** Provides convenience methods to use in jstl expression. All functions should be static. */
 @Slf4j
 public class JstlFunctions {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
   @Setter private static Clock clock = Clock.systemUTC();
 
   public static String uppercase(Object input) {
@@ -44,6 +51,21 @@ public static String lowercase(Object input) {
     return input == null ? null : toString(input).toLowerCase();
   }
 
+  public static String toJson(Object input) throws Exception {
+    return MAPPER.writeValueAsString(input);
+  }
+
+  public static Object fromJson(Object input) throws Exception {
+    if (input == null) {
+      return null;
+    }
+    String s = toString(input);
+    if (s.isEmpty()) {
+      return null;
+    }
+    return MAPPER.readValue(s, Object.class);
+  }
+
   public static Object toDouble(Object input) {
     if (input == null) {
       return null;
@@ -58,6 +80,58 @@ public static Object toInt(Object input) {
     return JstlTypeConverter.INSTANCE.coerceToInteger(input);
   }
 
+  public static List<String> split(Object input, Object separatorExpression) {
+    if (input == null) {
+      return null;
+    }
+    String s = toString(input);
+    if (s.isEmpty()) {
+      // special case, split would return a list with an empty string
+      return List.of();
+    }
+    String separator = toString(separatorExpression);
+    return Stream.of(s.split(separator)).collect(Collectors.toList());
+  }
+
+  public static Map<String, Object> unpack(Object input, Object fieldsExpression) {
+    if (input == null) {
+      return null;
+    }
+    String fields = toString(fieldsExpression);
+    Map<String, Object> result = new HashMap<>();
+    List<Object> values;
+    if (input instanceof String) {
+      String s = (String) input;
+      if (s.isEmpty()) {
+        values = List.of();
+      } else {
+        values = Stream.of(s.split(",")).collect(Collectors.toList());
+      }
+    } else if (input instanceof List) {
+      values = (List<Object>) input;
+    } else if (input instanceof Collection) {
+      values = new ArrayList<>((Collection<Object>) input);
+    } else if (input.getClass().isArray()) {
+      values = new ArrayList<>(Array.getLength(input));
+      for (int i = 0; i < Array.getLength(input); i++) {
+        values.add(Array.get(input, i));
+      }
+    } else {
+      throw new IllegalArgumentException(
+          "fn:unpack cannot unpack object of type " + input.getClass().getName());
+    }
+    List<String> headers = Stream.of(fields.split(",")).collect(Collectors.toList());
+    for (int i = 0; i < headers.size(); i++) {
+      String header = headers.get(i);
+      if (i < values.size()) {
+        result.put(header, values.get(i));
+      } else {
+        result.put(header, null);
+      }
+    }
+    return result;
+  }
+
   public static List<Object> filter(Object input, String expression) {
     if (input == null) {
       return null;

streaming-ai/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlFunctionsTest.java

@@ -29,6 +29,7 @@
 import java.time.LocalTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.testng.annotations.DataProvider;
@@ -393,6 +394,72 @@ void testCast() {
     assertEquals(null, JstlFunctions.toInt(null));
   }
 
+  @Test
+  void testSplit() {
+    assertEquals(List.of("1", "2"), JstlFunctions.split("1,2", ","));
+    assertEquals(List.of(), JstlFunctions.split("", ","));
+    assertEquals(null, JstlFunctions.split(null, ","));
+  }
+
+  @Test
+  void testUnpack() {
+    assertEquals(map("field1", "1", "field2", "2"), JstlFunctions.unpack("1,2", "field1,field2"));
+    assertEquals(map("field1", null, "field2", null), JstlFunctions.unpack("", "field1,field2"));
+    assertEquals(null, JstlFunctions.unpack(null, "field1,field2"));
+
+    assertEquals(
+        map("field1", "1", "field2", "2", "field3", null),
+        JstlFunctions.unpack("1,2", "field1,field2,field3"));
+    assertEquals(map("field1", "1", "field2", null), JstlFunctions.unpack("1", "field1,field2"));
+
+    assertEquals(
+        map("field1", "1", "field2", "2"),
+        JstlFunctions.unpack(JstlFunctions.split("1:2", ":"), "field1,field2"));
+
+    assertEquals(
+        map("field1", 1f, "field2", 2f), JstlFunctions.unpack(List.of(1f, 2f), "field1,field2"));
+  }
+
+  @Test
+  void testToJson() throws Exception {
+    assertEquals("{\"field1\":1}", JstlFunctions.toJson(Map.of("field1", 1)));
+    assertEquals("null", JstlFunctions.toJson(null));
+    assertEquals("\"\"", JstlFunctions.toJson(""));
+    assertEquals("[1,2,3]", JstlFunctions.toJson(List.of(1, 2, 3)));
+  }
+
+  @Test
+  void testFromJson() throws Exception {
+    assertEquals(Map.of("field1", 1), JstlFunctions.fromJson("{\"field1\":1}"));
+    assertEquals(null, JstlFunctions.fromJson(null));
+    assertEquals(null, JstlFunctions.fromJson("null"));
+    assertEquals(null, JstlFunctions.fromJson(""));
+    assertEquals("", JstlFunctions.fromJson("\"\""));
+    assertEquals(List.of(1, 2, 3), JstlFunctions.fromJson("[1,2,3]"));
+  }
+
+  private static Map<String, Object> map(
+      String key, Object nullableValue, String key2, Object nullableValue2) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(key, nullableValue);
+    map.put(key2, nullableValue2);
+    return map;
+  }
+
+  private static Map<String, Object> map(
+      String key,
+      Object nullableValue,
+      String key2,
+      Object nullableValue2,
+      String key3,
+      Object nullableValue3) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(key, nullableValue);
+    map.put(key2, nullableValue2);
+    map.put(key3, nullableValue3);
+    return map;
+  }
+
   @Test
   void testFilterQueryResults() {