add sourcePath intrinsic function

Author: cwensel

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

@@ -141,3 +141,7 @@ Use this when there are many files with overlapping but not consistent field nam
 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
+
+`sourcePath`:: Add the URI of the data currently being processed.
+Def:::
+- `^sourcePath{} -> to_field` - Assign the current URI to `to_field`

tessellate-main/src/main/java/io/clusterless/tessellate/operation/SourcePathFunction.java

@@ -52,10 +52,11 @@ public void prepare(FlowProcess flowProcess, OperationCall<Tuple> operationCall)
     public void operate(FlowProcess flowProcess, FunctionCall<Tuple> functionCall) {
         String sourcePath = flowProcess.getFlowProcessContext().getSourcePath();
 
-        if (sourcePath == null)
+        if (sourcePath == null) {
             sourcePath = defaultPath;
-        else if (fileNameOnly)
+        } else if (fileNameOnly) {
             sourcePath = getFileName(sourcePath);
+        }
 
         functionCall.getContext().set(0, sourcePath);
 
@@ -65,8 +66,9 @@ else if (fileNameOnly)
     protected String getFileName(String sourcePath) {
         int i = sourcePath.lastIndexOf(File.separatorChar);
 
-        if (i == -1)
+        if (i == -1) {
             return sourcePath;
+        }
 
         return sourcePath.substring(i + 1);
     }

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

@@ -23,6 +23,7 @@ private static void add(IntrinsicBuilder intrinsicBuilder) {
 
     static {
         add(new TsidIntrinsic());
+        add(new SourcePathIntrinsic());
         add(new FixedWidthIntrinsic());
         add(new EnsureIntrinsic());
         add(new ToJsonIntrinsic());

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

@@ -0,0 +1,37 @@
+/*
+ * 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.tuple.Fields;
+import io.clusterless.tessellate.operation.SourcePathFunction;
+import io.clusterless.tessellate.parser.ast.Operation;
+
+public class SourcePathIntrinsic extends IntrinsicBuilder {
+
+    public SourcePathIntrinsic() {
+        super("sourcePath");
+    }
+
+    @Override
+    public Result create(Fields currentFields, Operation operation) {
+        Fields toFields = fieldsParser().asFields(operation.results());
+
+        if (toFields.isNone()) {
+            toFields = new Fields("source_path", String.class);
+        }
+
+        if (toFields.size() != 1) {
+            throw new IllegalArgumentException("results may only have one field");
+        }
+
+        SourcePathFunction function = new SourcePathFunction(toFields);
+
+        return new Result(Fields.NONE, function, toFields);
+    }
+}

tessellate-main/src/test/java/io/clusterless/tessellate/pipeline/PipelineTest.java

@@ -350,7 +350,8 @@ void awsS3AccessLogWithTransforms(@PathForResource("/data/aws-s3-access-log.txt"
                         "httpStatusString|Integer",// coerce: "httpStatusString|String"
                         "requestID->",// discard: "httpStatusString|String"
                         "200=>code|Integer", // insert
-                        "=> _empty|Integer" // insert
+                        "=> _empty|Integer", // insert
+                        "^sourcePath{} +> source_path"
                 ))
                 .withSink(Sink.builder()
                         .withOutput(output)
@@ -377,7 +378,7 @@ void awsS3AccessLogWithTransforms(@PathForResource("/data/aws-s3-access-log.txt"
         CascadingTesting.validateEntries(
                 pipeline.flow().openSink(),
                 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 -> assertEquals(merged.source().schema().declared().size() + 1 + 1 + 1 + 1, l, "wrong tuple size"),
                 l -> {
                 }
         );