support for regex filter transforms via the ~/regex/ syntax

Author: cwensel

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

@@ -11,7 +11,8 @@ 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).
+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 (`'`) .
 
 `<field_type>` is optional, depending on the use. `<field_type>` further may be formatted as `<type>|<metadata>`.
@@ -20,8 +21,27 @@ 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 either by removing (filtering) a given tuple from the stream or by changing the values in any given tuple in the stream.
+
+=== Filters
+
+A tuple can be retained in a stream if a given predicate expression returns true for any the filter's arguments.
+
+==== Expressions
+
+If the expression is a regular expression, the expression will be matched with every argument individually after being coerced into a String.
+If any value (after coercion) is `null`, an empty string ("") will be passed to the regular expression matcher.
+
+All values must match to retain the tuple.
+
+One argument:: `from_field1 ~/expression/`
+Many arguments:: `from_field1 + from_field2 + from_fieldN ~/expression/`
+
+Note there is no operator after the `~/expression/`, this indicates the statement is a filter.
+
+Note a `/` can be escaped with `//` in any expression.
+
+=== Operators
 
 Insert literal:: Insert a literal value into a field.
 Coerce field:: Transform a field, in every tuple.
@@ -30,8 +50,6 @@ Rename field:: Rename a field, optionally coercing its type.
 Discard field:: Remove a field.
 Apply function:: Apply intrinsic functions against one or more fields.
 
-=== Operators
-
 There are three transform operators:
 
 `pass:[=>]`:: Assign a literal value to a new field.

tessellate-main/src/main/java/io/clusterless/tessellate/factory/TapFactories.java

@@ -99,6 +99,11 @@ public static SourceFactory findSourceFactory(List<URI> uris, Format format, Com
 
     public static SinkFactory findSinkFactory(Sink sinkModel) {
         List<URI> inputUris = sinkModel.uris();
+
+        if (inputUris.isEmpty()) {
+            return null;
+        }
+
         Format format = sinkModel.schema().format();
         Compression compression = sinkModel.schema().compression();
         SinkFactory sinkFactory = findSinkFactory(inputUris, format, compression);

tessellate-main/src/main/java/io/clusterless/tessellate/parser/StatementParser.java

@@ -80,6 +80,25 @@ private static Parser<Void> op(String op) {
                     (Map::entry)
             );
 
+    // Accepts any Java regex, where '/' is escaped as '//'
+    private static final Parser<RegExp> REGEX = Parsers.sequence(
+            Scanners.string("~/"),
+            Parsers.or(
+                            Scanners.string("//"), // the retn() method doesn't work here
+                            Scanners.notChar('/')
+                    )
+                    .many().source(),
+            Scanners.isChar('/'),
+            (unused, regex, unused2) -> new RegExp(regex.replace("//", "/"))
+    );
+
+    private static final Parser<FilterStatement> REGEX_FILTER =
+            Parsers.sequence(
+                    FieldParser.FIELD_LIST.followedBy(Scanners.many(IS_WHITESPACE)),
+                    REGEX.followedBy(EOF),
+                    FilterStatement::new
+            );
+
     private static final Parser<Map<String, String>> PARAMS =
             PARAM_ENTRY.sepBy(PARAM_DELIM).map(l -> l.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
 
@@ -148,18 +167,18 @@ private static Parser<Void> op(String op) {
                     UnaryOperation::new
             );
 
-    public static Parser<Assignment> LITERAL_ASSIGNMENT =
+    public static Parser<AssignmentStatement> LITERAL_ASSIGNMENT =
             Parsers.or(
                     Parsers.sequence(
                             LITERAL_VALUE,
                             ASSIGNMENT.followedBy(Scanners.many(IS_WHITESPACE)),
                             FieldParser.fullFieldDeclaration.followedBy(EOF),
-                            Assignment::new
+                            AssignmentStatement::new
                     ),
                     Parsers.sequence(
                             ASSIGNMENT.followedBy(Scanners.many(IS_WHITESPACE)),
                             FieldParser.fullFieldDeclaration.followedBy(EOF),
-                            Assignment::new
+                            AssignmentStatement::new
                     )
             );
 
@@ -192,22 +211,23 @@ private static Parser<Void> op(String op) {
                     (type, unused) -> type
             );
 
-    public static Parser<Join> JOIN_STATEMENT =
+    public static Parser<JoinStatement> JOIN_STATEMENT =
             Parsers.or(
                     Parsers.sequence(
                             FieldParser.RELATION_LIST.followedBy(Scanners.many(IS_WHITESPACE)),
                             JOIN.followedBy(Scanners.many(IS_WHITESPACE)),
                             Parsers.or(RETAIN, DISCARD).followedBy(Scanners.many(IS_WHITESPACE)),
                             FieldParser.FIELD_LIST.followedBy(EOF),
-                            Join::new
+                            JoinStatement::new
                     ),
                     Parsers.sequence(
                             FieldParser.RELATION_LIST.followedBy(Scanners.many(IS_WHITESPACE)),
                             JOIN.followedBy(EOF),
-                            Join::new
+                            JoinStatement::new
                     ));
 
     public static Parser<Statement> STATEMENTS = Parsers.or(
+            REGEX_FILTER,
             LITERAL_ASSIGNMENT,
             TRANSFORM_COERCE,
             TRANSFORM_DISCARD,

tessellate-main/src/main/java/io/clusterless/tessellate/parser/ast/Assignment.java renamed to tessellate-main/src/main/java/io/clusterless/tessellate/parser/ast/AssignmentStatement.java

@@ -11,18 +11,18 @@
 import com.google.common.base.Joiner;
 import io.clusterless.tessellate.parser.Printer;
 
-public class Assignment implements Statement {
+public class AssignmentStatement implements Statement {
     String literal;
     Op op;
     Field result;
 
-    public Assignment(String literal, Op op, Field result) {
+    public AssignmentStatement(String literal, Op op, Field result) {
         this.literal = literal;
         this.op = op;
         this.result = result;
     }
 
-    public Assignment(Op op, Field result) {
+    public AssignmentStatement(Op op, Field result) {
         this.literal = null;
         this.op = op;
         this.result = result;
@@ -42,8 +42,8 @@ public Field result() {
 
     @Override
     public String toString() {
-        return Joiner.on("")
-                .useForNull("")
+        return Joiner.on(" ")
+                .skipNulls()
                 .join(
                         Printer.literal(literal()),
                         op(),

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

@@ -0,0 +1,47 @@
+/*
+ * 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.parser.ast;
+
+import com.google.common.base.Joiner;
+import io.clusterless.tessellate.parser.Printer;
+
+import java.util.List;
+
+public class FilterStatement implements Statement {
+    List<Field> arguments;
+    Exp exp;
+
+    public FilterStatement(List<Field> arguments, Exp exp) {
+        this.arguments = arguments;
+        this.exp = exp;
+    }
+
+    public List<Field> arguments() {
+        return arguments;
+    }
+
+    public <T extends Exp> T exp() {
+        return (T) exp;
+    }
+
+    @Override
+    public Op op() {
+        return null;
+    }
+
+    @Override
+    public String toString() {
+        return Joiner.on(" ")
+                .skipNulls()
+                .join(
+                        Printer.fields(arguments),
+                        exp()
+                );
+    }
+}

tessellate-main/src/main/java/io/clusterless/tessellate/parser/ast/Join.java renamed to tessellate-main/src/main/java/io/clusterless/tessellate/parser/ast/JoinStatement.java

@@ -10,20 +10,20 @@
 
 import java.util.List;
 
-public class Join implements Statement {
+public class JoinStatement implements Statement {
     private final List<Rel> relations;
     private final JoinType joinType;
     private final Op op;
     private final List<Field> results;
 
-    public Join(List<Rel> relations, JoinType joinType) {
+    public JoinStatement(List<Rel> relations, JoinType joinType) {
         this.relations = relations;
         this.joinType = joinType;
         this.op = new Op();
         this.results = List.of();
     }
 
-    public Join(List<Rel> relations, JoinType joinType, Op op, List<Field> results) {
+    public JoinStatement(List<Rel> relations, JoinType joinType, Op op, List<Field> results) {
         this.relations = relations;
         this.joinType = joinType;
         this.op = op;

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

@@ -17,7 +17,7 @@
 public class Operation implements Statement {
     List<Field> arguments = Collections.emptyList();
     Exp exp;
-    Op op = new Op();
+    Op op;
     List<Field> results = Collections.emptyList();
 
     public Operation(List<Field> arguments, Exp exp, Op op, List<Field> results) {
@@ -66,13 +66,13 @@ public List<Field> results() {
 
     @Override
     public String toString() {
-        return Joiner.on("")
-                .useForNull("")
+        return Joiner.on(" ")
+                .skipNulls()
                 .join(
-                        Printer.fields(arguments),
+                        arguments.isEmpty() ? null : Printer.fields(arguments),
                         exp(),
                         op(),
-                        Printer.fields(results)
+                        results.isEmpty() ? null : Printer.fields(results)
                 );
     }
 }

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

@@ -0,0 +1,26 @@
+/*
+ * 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.parser.ast;
+
+public class RegExp implements Exp {
+    String pattern;
+
+    public RegExp(String pattern) {
+        this.pattern = pattern;
+    }
+
+    public String pattern() {
+        return pattern;
+    }
+
+    @Override
+    public String toString() {
+        return "~/" + pattern + "/";
+    }
+}

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

@@ -11,7 +11,19 @@
 public interface Statement {
     Op op();
 
+    default String opString() {
+        if (op() == null) {
+            return "";
+        }
+
+        return op().op();
+    }
+
     default boolean isJoin() {
-        return this instanceof Join;
+        return this instanceof JoinStatement;
+    }
+
+    default boolean isFilter() {
+        return this instanceof FilterStatement;
     }
 }

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

@@ -29,7 +29,7 @@
 import io.clusterless.tessellate.model.*;
 import io.clusterless.tessellate.options.PipelineOptions;
 import io.clusterless.tessellate.options.PrintOptions;
-import io.clusterless.tessellate.parser.ast.Join;
+import io.clusterless.tessellate.parser.ast.JoinStatement;
 import io.clusterless.tessellate.parser.ast.Rel;
 import io.clusterless.tessellate.parser.ast.Statement;
 import io.clusterless.tessellate.printer.SchemaPrinter;
@@ -288,14 +288,14 @@ private Source findPrimarySource() {
             return source;
         }
 
-        List<Join> joins = pipelineDef.transform().statements(Join.class);
+        List<JoinStatement> joins = pipelineDef.transform().statements(JoinStatement.class);
 
         if (joins.isEmpty()) {
             throw new IllegalStateException("no source defined");
         }
 
         List<String> names = joins.stream()
-                .map(Join::rhsRelations)
+                .map(JoinStatement::rhsRelations)
                 .flatMap(List::stream)
                 .map(Rel::name)
                 .distinct()
@@ -309,10 +309,10 @@ private Source findPrimarySource() {
     }
 
     private Map<String, Source> findSecondarySources() {
-        List<Join> joins = pipelineDef.transform().statements(Join.class);
+        List<JoinStatement> joins = pipelineDef.transform().statements(JoinStatement.class);
 
         List<String> names = joins.stream()
-                .map(Join::lhsRelations)
+                .map(JoinStatement::lhsRelations)
                 .flatMap(List::stream)
                 .map(Rel::name)
                 .distinct()