add formatFields intrinsic transform function

Apache Parquet doesn't like spaces and some characters in column names, this makes renaming fields a bit easier.

Author: cwensel

tessellate-main/src/main/antora/modules/reference/pages/transforms.adoc

@@ -2,14 +2,14 @@
 
 == Fields
 
-Input and output files/objects (also referred to as sources and sinks) are made of both rows and columns. Or tuples and fields.
+Input and output files/objects (also referred to as sources and sinks) are made of both rows and columns.
+Or tuples and fields.
 
 A tuple has a set of fields, and a field has an optional xref:types.adoc[type] (and any associated metadata).
 
-Data files, or objects, have paths and names. Field values can be parsed from the paths and embedded in the tuple stream
-as fields. This is common when data has been partitioned into files where common values (like month and/or day) can
-be embedded in the path name to help select relevant files (push down predicates are applied to path values by many
-query engines).
+Data files, or objects, have paths and names.
+Field values can be parsed from the paths and embedded in the tuple stream as fields.
+This is common when data has been partitioned into files where common values (like month and/or day) can be embedded in the path name to help select relevant files (push down predicates are applied to path values by many query engines).
 
 Declared fields in a pipeline have the following format: `<field_name>|<field_type>`, where `<field_name>` is a string, or an ordinal (number representing the position).
 The `<field_name>` may be quoted by a single quote (`'`) .
@@ -20,7 +20,8 @@ The actual supported types and associated metadata are described in xref:types.a
 
 == Transforms
 
-Transforms manipulate the tuple stream. They are applied to every tuple in the tuple stream.
+Transforms manipulate the tuple stream.
+They are applied to every tuple in the tuple stream.
 
 Insert literal:: Insert a literal value into a field.
 Coerce field:: Transform a field, in every tuple.
@@ -49,14 +50,16 @@ For example:
 - `1689820455 pass:[=>] time|DateTime|yyyyMMdd` - convert the long value to a date time using the format `yyyyMMdd` and assign the result to the field `time`.
 - `ratio +> ratio|Double` - Coerces the string field "ratio" to a double, `null` ok.
 - `ratio|Double` - Same as above, coerces the string field "ratio" to a double, `null` ok.
-- `name +> firstName|String` - assigns the value of the field "name" to the field "firstName" as a string. The field `name` is retained.
-- `name pass:[->] firstName|String` - assigns the value of the field "name" to the field "firstName" as a string. The field `name` is discarded (dropped from the tuple stream).
+- `name +> firstName|String` - assigns the value of the field "name" to the field "firstName" as a string.
+The field `name` is retained.
+- `name pass:[->] firstName|String` - assigns the value of the field "name" to the field "firstName" as a string.
+The field `name` is discarded (dropped from the tuple stream).
 - `password pass:[->]` - discards the field `password` from the tuple stream.
 
 ==== Expressions
 
-Expressions are applied to incoming fields and the results are assigned to a new field. Expressions can have zero or
-more field arguments.
+Expressions are applied to incoming fields and the results are assigned to a new field.
+Expressions can have zero or more field arguments.
 
 There are two types of expression:
 
@@ -65,8 +68,7 @@ There are two types of expression:
 
 NOTE: Many more expression types are planned, including native support for regular expressions and JSON paths.
 
-Current only `intrinsic` functions are supported. `intrinsic` functions are built-in functions, with optional
-parameters
+Current only `intrinsic` functions are supported. `intrinsic` functions are built-in functions, with optional parameters
 
 No arguments:: `^intrinsic{} +> new_field|type`
 No arguments, with parameters:: `^intrinsic{param1:value1, param2:value2} +> new_field|type`
@@ -84,14 +86,16 @@ Built-in functions on fields can be applied to one or more fields in every tuple
 `tsid`:: create a unique id as a long or string (using https://github.com/f4b6a3/tsid-creator)
 Def:::
 `^tsid{node:...,nodeCount:...,epoch:...,format:...,counterToZero:...} +> intoField|type`
-`type`:::: must be `string` or `long`, defaults to `long`. When `string`, the `format` is honored.
+`type`:::: must be `string` or `long`, defaults to `long`.
+When `string`, the `format` is honored.
 Params:::
 `node`:::: The node id, defaults to a random int.
 * If a string is provided, it is hashed to an int.
 * `SIP_HASHER.hashString(s, StandardCharsets.UTF_8).asInt() % nodeCount;`
 `nodeCount`:::: The number of nodes, defaults to `1024`
 `epoch`:::: - The epoch, defaults to `Instant.parse("2020-01-01T00:00:00.000Z").toEpochMilli()`
-`format`:::: The format, defaults to `null`. Example: `K%S` where `%S` is a placeholder.
+`format`:::: The format, defaults to `null`.
+Example: `K%S` where `%S` is a placeholder.
 Placeholders:::::
 - `%S`: canonical string in upper case
 - `%s`: canonical string in lower case
@@ -118,3 +122,15 @@ Def:::
 `fromJson`:: converts a json string to a row/tuple
 Def:::
 - `^fromJson{} -> to_field1 + to_field2` - the fields names must match the json properties at the root node
+
+`formatFields`:: reformats all the argument fields to a new format.
+This is especially useful for remaining compatible with Apache Parquet.
+Def:::
+- `^formatFields{format:...} ->` - replace ALL fields with the formatted result
+- `^formatFields{format:...} +>` - append ALL fields with the formatted result
+Params:::
+`format`:::: The format string.
+- `lowerUnderscore` - converts to lower case and replaces `/\ .-` with underscores.
+The default.
+- `upperUnderscore` - converts to upper case and replaces `/\ .-` with underscores
+- `camelCase` - converts to camel case

tessellate-main/src/main/java/io/clusterless/tessellate/parser/ast/IntrinsicParams.java

@@ -25,46 +25,46 @@ public Map<String, String> params() {
         return params;
     }
 
-    public String getString(String param) {
-        return params.get(param);
+    public Optional<String> getString(String param) {
+        return Optional.ofNullable(params.get(param));
     }
 
-    public Boolean getBoolean(String param) {
+    public Optional<Boolean> getBoolean(String param) {
         String value = params.get(param);
 
         if (value == null) {
-            return null;
+            return Optional.empty();
         }
 
-        return Boolean.parseBoolean(value);
+        return Optional.of(Boolean.parseBoolean(value));
     }
 
     public Integer getInteger(String param, Function<String, Integer> otherwise) {
         try {
-            return getInteger(param);
+            return getInteger(param).orElse(null);
         } catch (NumberFormatException e) {
-            return otherwise.apply(getString(param));
+            return otherwise.apply(getString(param).orElse(null));
         }
     }
 
-    public Integer getInteger(String param) {
+    public Optional<Integer> getInteger(String param) {
         String value = params.get(param);
 
         if (value == null) {
-            return null;
+            return Optional.empty();
         }
 
-        return Integer.parseInt(value);
+        return Optional.of(Integer.parseInt(value));
     }
 
-    public Long getLong(String param) {
+    public Optional<Long> getLong(String param) {
         String value = params.get(param);
 
         if (value == null) {
-            return null;
+            return Optional.empty();
         }
 
-        return Long.parseLong(value);
+        return Optional.of(Long.parseLong(value));
     }
 
     @Override

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Intrinsics.java

@@ -26,6 +26,7 @@ private static void add(IntrinsicBuilder intrinsicBuilder) {
         add(new FixedWidthIntrinsic());
         add(new ToJsonIntrinsic());
         add(new FromJsonIntrinsic());
+        add(new FormatFieldsIntrinsic());
     }
 
     public static Map<String, IntrinsicBuilder> builders() {

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Transformer.java

@@ -128,7 +128,7 @@ private IntrinsicBuilder.Result create(PipelineContext context, Operation operat
                 throw new IllegalArgumentException("unknown intrinsic function: " + intrinsic.name());
             }
 
-            IntrinsicBuilder.Result result = intrinsicBuilder.create(operation);
+            IntrinsicBuilder.Result result = intrinsicBuilder.create(context.currentFields, operation);
 
             context.log.info("transform {}: from: {}, to: {}, having: {}", intrinsicBuilder.name(), result.arguments(), result.results(), ((Intrinsic) operation.exp()).params());
 

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/FixedWidthIntrinsic.java

@@ -23,12 +23,12 @@ public FixedWidthIntrinsic() {
     }
 
     @Override
-    public Result create(Operation operation) {
+    public Result create(Fields currentFields, Operation operation) {
         Fields toFields = fieldsParser().asFields(operation.results());
         Intrinsic intrinsic = operation.exp();
 
-        Integer width = intrinsic.params().getInteger(WIDTH);
-        Integer insertAt = intrinsic.params().getInteger(INSERT_AT);
+        Integer width = intrinsic.params().getInteger(WIDTH).orElse(null);
+        Integer insertAt = intrinsic.params().getInteger(INSERT_AT).orElse(null);
 
         // if not given, we will insert at the end
         if (insertAt == null) {

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/FormatFieldsIntrinsic.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (c) 2023-2025 Chris K Wensel <[email protected]>. All Rights Reserved.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+package io.clusterless.tessellate.pipeline.intrinsic;
+
+import cascading.operation.Identity;
+import cascading.tuple.Fields;
+import clusterless.commons.util.Strings;
+import io.clusterless.tessellate.parser.ast.Intrinsic;
+import io.clusterless.tessellate.parser.ast.Operation;
+import io.clusterless.tessellate.util.StringHelper;
+
+import java.lang.reflect.Type;
+import java.util.function.BiFunction;
+
+public class FormatFieldsIntrinsic extends IntrinsicBuilder {
+
+    public static final String FORMAT = "format";
+    public static final String REGEX = "[/\\\\ .-]";
+
+    public FormatFieldsIntrinsic() {
+        super("formatFields", FORMAT);
+    }
+
+    @Override
+    public Result create(Fields currentFields, Operation operation) {
+        Fields fromFields = fieldsParser().asFields(operation.arguments());
+
+        // doesn't make sense to make an empty json object
+        if (fromFields.isNone()) {
+            fromFields = currentFields;
+        }
+
+        Intrinsic intrinsic = operation.exp();
+        String format = intrinsic.params().getString(FORMAT).orElse("lowerUnderscore");
+
+        BiFunction<Comparable, Type, Comparable> function;
+
+        switch (format) {
+            case "camelCase":
+                function = FormatFieldsIntrinsic::camelCase;
+                break;
+            case "upperUnderscore":
+                function = FormatFieldsIntrinsic::upperUnderscore;
+                break;
+            case "lowerUnderscore":
+                function = FormatFieldsIntrinsic::lowerUnderscore;
+                break;
+            default:
+                throw new IllegalStateException("Unexpected value: " + format);
+        }
+
+        Fields toFields = fromFields.rename(function);
+
+        Identity identity = new Identity(toFields);
+
+        return new Result(fromFields, identity, toFields);
+    }
+
+    public static Comparable<?> lowerUnderscore(Comparable<?> comparable, Type type) {
+        String string = comparable.toString().replaceAll(REGEX, "_");
+
+        if (!string.contains("_")) {
+            return Strings.camelToLowerUnderscore(StringHelper.convertConsecutiveUpperCase(string));
+        }
+
+        return string.toLowerCase();
+    }
+
+    public static Comparable<?> upperUnderscore(Comparable<?> comparable, Type type) {
+        String string = comparable.toString().replaceAll(REGEX, "_");
+
+        if (!string.contains("_")) {
+            return Strings.camelToUpperUnderscore(StringHelper.convertConsecutiveUpperCase(string));
+        }
+
+        return string.toUpperCase();
+    }
+
+    public static Comparable<?> camelCase(Comparable<?> comparable, Type type) {
+        String string = comparable.toString().replaceAll(REGEX, "");
+        return Strings.lowerUnderscoreToCamelCase(lowerUnderscore(string, type).toString());
+    }
+}

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/FromJsonIntrinsic.java

@@ -22,7 +22,7 @@ public FromJsonIntrinsic() {
     }
 
     @Override
-    public Result create(Operation operation) {
+    public Result create(Fields currentFields, Operation operation) {
         Fields fromFields = fieldsParser().asFields(operation.arguments());
 
         // doesn't make sense to make an empty json object

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/IntrinsicBuilder.java

@@ -70,7 +70,7 @@ public FieldsParser fieldsParser() {
         return FieldsParser.INSTANCE;
     }
 
-    public abstract Result create(Operation operation);
+    public abstract Result create(Fields currentFields, Operation operation);
 
     protected static void requireParam(Object value, String message) {
         if (value == null) {

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/ToJsonIntrinsic.java

@@ -19,7 +19,7 @@ public ToJsonIntrinsic() {
     }
 
     @Override
-    public Result create(Operation operation) {
+    public Result create(Fields currentFields, Operation operation) {
         Fields fromFields = fieldsParser().asFields(operation.arguments());
 
         // doesn't make sense to make an empty json object

tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/TsidIntrinsic.java

@@ -32,20 +32,20 @@ public TsidIntrinsic() {
     }
 
     @Override
-    public Result create(Operation operation) {
+    public Result create(Fields currentFields, Operation operation) {
         // default to long if not provided
         Fields toFields = fieldsParser().asFields(operation.results(), Long.TYPE);
         Intrinsic intrinsic = operation.exp();
 
-        Integer nodeCount = intrinsic.params().getInteger(NODE_COUNT);
+        Integer nodeCount = intrinsic.params().getInteger(NODE_COUNT).orElse(null);
         Integer node = intrinsic.params().getInteger(NODE, s -> {
             requireParam(nodeCount, "nodeCount param required when providing a string node value");
             return Math.abs(SIP.hashString(s, StandardCharsets.UTF_8).asInt()) % nodeCount;
         });
 
-        Long epoch = intrinsic.params().getLong(EPOCH);
-        String format = intrinsic.params().getString(FORMAT);
-        Boolean counterZero = intrinsic.params().getBoolean(COUNTER_TO_ZERO);
+        Long epoch = intrinsic.params().getLong(EPOCH).orElse(null);
+        String format = intrinsic.params().getString(FORMAT).orElse(null);
+        Boolean counterZero = intrinsic.params().getBoolean(COUNTER_TO_ZERO).orElse(null);
 
         TsidFunction function = new TsidFunction(toFields, node, nodeCount, epoch, format, counterZero);