add ensureFields intrinsic to allow for adding missing fields

cwensel · cwensel · commit 895a7dec7a21 · 2025-05-02T09:03:02.000-07:00
diff --git a/tessellate-main/src/main/antora/modules/reference/pages/transforms.adoc b/tessellate-main/src/main/antora/modules/reference/pages/transforms.adoc
@@ -135,3 +135,9 @@ Params:::
 The default.
 - `upperUnderscore` - converts to upper case and replaces `/\ .-` with underscores
 - `camelCase` - converts to camel case
+
+`ensureFields`:: Add any missing fields/columns.
+Use this when there are many files with overlapping but not consistent field names and the files need to be normalized so they can be processed as a single unit.
+Note any fields not declared as the result fields will be discarded.
+Def:::
+- `^ensureFields{} -> to_field1 + to_field2` - Ensure the results have all declared fields
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Intrinsics.java b/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Intrinsics.java
@@ -24,6 +24,7 @@ private static void add(IntrinsicBuilder intrinsicBuilder) {
     static {
         add(new TsidIntrinsic());
         add(new FixedWidthIntrinsic());
+        add(new EnsureIntrinsic());
         add(new ToJsonIntrinsic());
         add(new FromJsonIntrinsic());
         add(new FormatFieldsIntrinsic());
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Transformer.java b/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/Transformer.java
@@ -54,8 +54,23 @@ private PipelineContext discardAndEval(PipelineContext context) {
 
         if (operation.exp() != null) {
             IntrinsicBuilder.Result result = create(context, operation);
-            // SWAP allows for efficient replacement of arguments with non-argument and results
-            Fields selector = result.arguments().isAll() || (context.currentFields.contains(result.arguments()) && context.currentFields.size() == result.arguments().size()) ? Fields.RESULTS : Fields.SWAP;
+
+            boolean argsIsAll = result.arguments().isAll() ||
+                    containsAll(context.currentFields, result.arguments());
+
+            boolean resultsEqualsDeclared = containsAll(result.function().getFieldDeclaration(), result.results());
+
+            Fields selector;
+
+            if ((argsIsAll && result.results().isNone()) || resultsEqualsDeclared) {
+                selector = Fields.RESULTS;
+            } else if (argsIsAll) {
+                selector = result.results();
+            } else {
+                // SWAP allows for efficient replacement of arguments with non-argument and results
+                selector = Fields.SWAP;
+            }
+
             Pipe pipe = new Each(context.pipe, result.arguments(), result.function(), selector);
 
             Fields currentFields = context.currentFields.subtract(result.arguments()).append(result.results());
@@ -78,6 +93,10 @@ private PipelineContext discardAndEval(PipelineContext context) {
         }
     }
 
+    private static boolean containsAll(Fields lhs, Fields rhs) {
+        return lhs.contains(rhs) && lhs.size() == rhs.size();
+    }
+
     private PipelineContext copyAndEval(PipelineContext context) {
         Operation operation = (Operation) statement;
 
diff --git a/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/EnsureIntrinsic.java b/tessellate-main/src/main/java/io/clusterless/tessellate/pipeline/intrinsic/EnsureIntrinsic.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2023-2025 Chris K Wensel <chris@wensel.net>. 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.Insert;
+import cascading.tuple.Fields;
+import io.clusterless.tessellate.parser.ast.Operation;
+
+public class EnsureIntrinsic extends IntrinsicBuilder {
+
+    public EnsureIntrinsic() {
+        super("ensureFields");
+    }
+
+    @Override
+    public Result create(Fields currentFields, Operation operation) {
+        Fields toFields = fieldsParser().asFields(operation.results());
+
+        if (currentFields.isNone()) {
+            throw new IllegalArgumentException("this intrinsic requires fields to be declared, got: " + currentFields.print());
+        }
+
+        if (toFields.isNone()) {
+            throw new IllegalArgumentException("result fields must be declared");
+        }
+
+        Fields insertFields = toFields.subtract(currentFields);
+
+        Object[] values = new Object[insertFields.size()];
+
+        Insert function = new Insert(insertFields, values);
+
+        return new Result(Fields.ALL, function, toFields);
+    }
+}
diff --git a/tessellate-main/src/test/java/io/clusterless/tessellate/pipeline/PipelineTest.java b/tessellate-main/src/test/java/io/clusterless/tessellate/pipeline/PipelineTest.java
@@ -78,8 +78,8 @@ void noHeaders(@PathForResource("/data/delimited.csv") URI input, @PathForOutput
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(12, l, "wrong length"),
-                l -> assertEquals(5, l, "wrong size"),
+                l -> assertEquals(12, l, "wrong file length"),
+                l -> assertEquals(5, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -93,7 +93,7 @@ void badWidth(@PathForResource("/data/delimited-variable-width.csv") URI input,
                 "^fixedWidth{ width:5, insertAt:3 } ->"
         );
 
-        fixedWidthBase(input, output, transform, List.of());
+        fixedSchemaBase(input, output, transform, List.of(), false, 5, 5);
     }
 
     @Test
@@ -102,7 +102,7 @@ void badWidthWithFields(@PathForResource("/data/delimited-variable-width.csv") U
                 "^fixedWidth{ width:5, insertAt:3 } -> _0+_1+_2+_3+_4"
         );
 
-        fixedWidthBase(input, output, transform, List.of());
+        fixedSchemaBase(input, output, transform, List.of(), false, 5, 5);
     }
 
     @Test
@@ -111,10 +111,28 @@ void badWidthWithFieldsAndTypes(@PathForResource("/data/delimited-variable-width
                 "^fixedWidth{ width:5, insertAt:3 } -> _0|string+_1|string+_2|string+_3|string+_4|string"
         );
 
-        fixedWidthBase(input, output, transform, Field.asField("a|string", "b|string", "c|string", "d|string", "e|string"));
+        fixedSchemaBase(input, output, transform, Field.asField("a|string", "b|string", "c|string", "d|string", "e|string"), false, 5, 5);
     }
 
-    private static void fixedWidthBase(URI input, URI output, Transform transform, List<@NotNull Field> declared) throws IOException {
+    @Test
+    void overlappingSchema(@PathForResource("/data/delimited-header-missing-col.csv") URI input, @PathForOutput URI output) throws IOException {
+        Transform transform = new Transform(
+                "^ensureFields{} -> first|string+second|string+third|string+fourth|string+fifth|string"
+        );
+
+        fixedSchemaBase(input, output, transform, List.of(), true, 13, 5);
+    }
+
+    @Test
+    void overlappingSchemaWithTail(@PathForResource("/data/delimited-header-missing-col.csv") URI input, @PathForOutput URI output) throws IOException {
+        Transform transform = new Transform(
+                "^ensureFields{} -> first|string+second|string+third|string+fourth|string+fifth|string+sixth|string"
+        );
+
+        fixedSchemaBase(input, output, transform, List.of(), true, 13, 6);
+    }
+
+    private static void fixedSchemaBase(URI input, URI output, Transform transform, List<@NotNull Field> declared, boolean embedsSchema, int fileLength, int tupleSize) throws IOException {
         PipelineOptions pipelineOptions = new PipelineOptions();
 
         PipelineDef def = PipelineDef.builder()
@@ -124,7 +142,7 @@ private static void fixedWidthBase(URI input, URI output, Transform transform, L
                         .withSchema(Schema.builder()
                                 .withDeclared(declared)
                                 .withFormat(Format.csv)
-                                .withEmbedsSchema(false)
+                                .withEmbedsSchema(embedsSchema)
                                 .withStrictParsing(declared.isEmpty())
                                 .build())
                         .build())
@@ -147,8 +165,8 @@ private static void fixedWidthBase(URI input, URI output, Transform transform, L
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(5, l, "wrong length"),
-                l -> assertEquals(5, l, "wrong size"),
+                l -> assertEquals(fileLength, l, "wrong file length"),
+                l -> assertEquals(tupleSize, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -189,8 +207,8 @@ void headers(@PathForResource("/data/delimited-header.csv") URI input, @PathForO
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(13, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(5, l, "wrong size"),
+                l -> assertEquals(13, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(5, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -259,8 +277,8 @@ private static void headersPartitionedBase(URI input, URI output, List<SourcePar
 
         CascadingTesting.validateEntries(
                 iterator,
-                l -> assertEquals(13, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(8, l, "wrong size"),
+                l -> assertEquals(13, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(8, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -304,8 +322,8 @@ void awsS3AccessLog(@PathForResource("/data/aws-s3-access-log.txt") URI input, @
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -358,8 +376,8 @@ void awsS3AccessLogWithTransforms(@PathForResource("/data/aws-s3-access-log.txt"
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size() + 1 + 1 + 1, l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size() + 1 + 1 + 1, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -403,8 +421,8 @@ void writeReadParquet(@PathForResource("/data/aws-s3-access-log.txt") URI input,
 
         CascadingTesting.validateEntries(
                 pipelineWrite.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -438,8 +456,8 @@ void writeReadParquet(@PathForResource("/data/aws-s3-access-log.txt") URI input,
 
         CascadingTesting.validateEntries(
                 pipelineRead.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size(), l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -526,8 +544,8 @@ void writeReadParquetPartitioned(@PathForResource("/data/aws-s3-access-log.txt")
 
         CascadingTesting.validateEntries(
                 iterator,
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -598,8 +616,8 @@ void writeReadParquetPartitioned(@PathForResource("/data/aws-s3-access-log.txt")
 
         CascadingTesting.validateEntries(
                 finalIterator,
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -685,8 +703,8 @@ void writeReadParquetPartitionedWithManifests(
 
         CascadingTesting.validateEntries(
                 pipelineWrite.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -729,8 +747,8 @@ void writeReadParquetPartitionedWithManifests(
 
         CascadingTesting.validateEntries(
                 pipelineRead.flow().openSink(),
-                l -> assertEquals(4, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong size"),
+                l -> assertEquals(4, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(merged.source().schema().declared().size() + 3, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -777,8 +795,8 @@ void toJsonAndBack(
 
         CascadingTesting.validateEntries(
                 writeJson.flow().openSink(),
-                l -> assertEquals(13, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(1, l, "wrong size"),
+                l -> assertEquals(13, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(1, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -820,8 +838,8 @@ void toJsonAndBack(
 
         CascadingTesting.validateEntries(
                 readJson.flow().openSink(),
-                l -> assertEquals(13, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(5, l, "wrong size"),
+                l -> assertEquals(13, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(5, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -882,8 +900,8 @@ void headersBadWidth(
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(11, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(5, l, "wrong size"),
+                l -> assertEquals(11, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(5, l, "wrong tuple size"),
                 l -> {
                 }
         );
@@ -983,8 +1001,8 @@ private static void joinTest(URI lhs, URI rhs, URI output, String transform, int
 
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
-                l -> assertEquals(length, l, "wrong length"), // headers are declared so aren't counted
-                l -> assertEquals(size, l, "wrong size"),
+                l -> assertEquals(length, l, "wrong file length"), // headers are declared so aren't counted
+                l -> assertEquals(size, l, "wrong tuple size"),
                 l -> {
                 }
         );
diff --git a/tessellate-main/src/test/resources/data/delimited-header-missing-col.csv b/tessellate-main/src/test/resources/data/delimited-header-missing-col.csv
@@ -0,0 +1,14 @@
+first,second,fourth,fifth
+foo,bar,bin,1
+foo,bar,bin,2
+foo,"bar,bar",bin,3
+foo,"bar"",bar",bin,4
+foo,"bar"""",bar",bin,5
+,"",,6
+,,,7
+foo,,,8
+,"",,9
+f,,,10
+f,,",bin",11
+f,,",bin",
+f,,,
