feat(workflow-operator): let Filter combine predicates with AND

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/PredicateCombinator.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.filter;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+public enum PredicateCombinator {
+    OR("any (OR)"),
+
+    AND("all (AND)");
+
+    private final String name;
+
+    private PredicateCombinator(String name) {
+        this.name = name;
+    }
+
+    // use the name string instead of enum string in JSON
+    @JsonValue
+    public String getName() {
+        return this.name;
+    }
+
+    // Handle custom deserialization for enum
+    @JsonCreator
+    public static PredicateCombinator fromString(String value) {
+        for (PredicateCombinator combinator : PredicateCombinator.values()) {
+            if (combinator.name.equalsIgnoreCase(value)) {
+                return combinator;
+            }
+        }
+        throw new IllegalArgumentException("Unknown predicate combinator: " + value);
+    }
+}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala

@@ -20,6 +20,7 @@
 package org.apache.texera.amber.operator.filter
 
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp}
@@ -29,9 +30,14 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
 class SpecializedFilterOpDesc extends FilterOpDesc {
 
   @JsonProperty(value = "predicates", required = true)
-  @JsonPropertyDescription("multiple predicates in OR")
+  @JsonPropertyDescription("multiple predicates to be combined")
   var predicates: List[FilterPredicate] = List.empty
 
+  @JsonProperty(required = true, defaultValue = "any (OR)")
+  @JsonSchemaTitle("Combine Predicates With")
+  @JsonPropertyDescription("how to combine multiple predicates")
+  var predicateCombinator: PredicateCombinator = PredicateCombinator.OR
+
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -53,7 +59,7 @@ class SpecializedFilterOpDesc extends FilterOpDesc {
   override def operatorInfo: OperatorInfo = {
     OperatorInfo(
       "Filter",
-      "Performs a filter operation using OR between multiple predicates",
+      "Performs a filter operation using AND or OR between multiple predicates",
       OperatorGroupConstants.CLEANING_GROUP,
       List(InputPort()),
       List(OutputPort()),

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExec.scala

@@ -25,5 +25,11 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
 class SpecializedFilterOpExec(descString: String) extends FilterOpExec {
   private val desc: SpecializedFilterOpDesc =
     objectMapper.readValue(descString, classOf[SpecializedFilterOpDesc])
-  setFilterFunc((tuple: Tuple) => desc.predicates.exists(_.evaluate(tuple)))
+  setFilterFunc((tuple: Tuple) =>
+    desc.predicateCombinator match {
+      case PredicateCombinator.AND =>
+        desc.predicates.nonEmpty && desc.predicates.forall(_.evaluate(tuple))
+      case _ => desc.predicates.exists(_.evaluate(tuple))
+    }
+  )
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/PredicateCombinatorSpec.scala

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.filter
+
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PredicateCombinatorSpec extends AnyFlatSpec with Matchers {
+
+  "PredicateCombinator" should "map each constant to its wire name via getName" in {
+    PredicateCombinator.OR.getName shouldBe "any (OR)"
+    PredicateCombinator.AND.getName shouldBe "all (AND)"
+    PredicateCombinator.values() should have length 2
+  }
+
+  "PredicateCombinator.fromString" should "resolve names case-insensitively" in {
+    PredicateCombinator.fromString("any (OR)") shouldBe PredicateCombinator.OR
+    PredicateCombinator.fromString("ALL (AND)") shouldBe PredicateCombinator.AND
+  }
+
+  it should "reject an unknown name" in {
+    intercept[IllegalArgumentException](PredicateCombinator.fromString("xor"))
+  }
+
+  "PredicateCombinator" should "round-trip through Jackson using its name" in {
+    objectMapper.writeValueAsString(PredicateCombinator.AND) shouldBe "\"all (AND)\""
+    objectMapper.readValue(
+      "\"any (OR)\"",
+      classOf[PredicateCombinator]
+    ) shouldBe PredicateCombinator.OR
+  }
+}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala

@@ -36,6 +36,7 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers {
     "advertise the name, Cleaning group, and reconfiguration support" in {
     val info = (new SpecializedFilterOpDesc).operatorInfo
     info.userFriendlyName shouldBe "Filter"
+    info.operatorDescription should include("AND or OR")
     info.operatorGroupName shouldBe OperatorGroupConstants.CLEANING_GROUP
     info.inputPorts should have length 1
     info.outputPorts should have length 1
@@ -46,6 +47,10 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers {
     (new SpecializedFilterOpDesc).predicates shouldBe empty
   }
 
+  "SpecializedFilterOpDesc.predicateCombinator" should "default to OR" in {
+    (new SpecializedFilterOpDesc).predicateCombinator shouldBe PredicateCombinator.OR
+  }
+
   "SpecializedFilterOpDesc.getPhysicalOp" should
     "wire SpecializedFilterOpExec and carry port identities" in {
     val op = new SpecializedFilterOpDesc
@@ -70,4 +75,27 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers {
     restored shouldBe a[SpecializedFilterOpDesc]
     restored.asInstanceOf[SpecializedFilterOpDesc].predicates shouldBe empty
   }
+
+  "SpecializedFilterOpDesc" should
+    "deserialize to OR when predicateCombinator is absent" in {
+    val restored =
+      objectMapper.readValue(
+        """{"operatorType":"Filter","predicates":[]}""",
+        classOf[LogicalOp]
+      )
+    restored
+      .asInstanceOf[SpecializedFilterOpDesc]
+      .predicateCombinator shouldBe PredicateCombinator.OR
+  }
+
+  "SpecializedFilterOpDesc" should
+    "round-trip an explicit AND combinator through the polymorphic base" in {
+    val op = new SpecializedFilterOpDesc
+    op.predicateCombinator = PredicateCombinator.AND
+    val restored =
+      objectMapper.readValue(objectMapper.writeValueAsString(op), classOf[LogicalOp])
+    restored
+      .asInstanceOf[SpecializedFilterOpDesc]
+      .predicateCombinator shouldBe PredicateCombinator.AND
+  }
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExecSpec.scala

@@ -61,6 +61,10 @@ class SpecializedFilterOpExecSpec extends AnyFlatSpec with BeforeAndAfter {
     .add(new Attribute("long", AttributeType.LONG), Long.MaxValue)
     .build()
 
+  before {
+    opDesc.predicateCombinator = PredicateCombinator.OR
+  }
+
   it should "open and close" in {
     opDesc.predicates = List()
     val opExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
@@ -135,4 +139,65 @@ class SpecializedFilterOpExecSpec extends AnyFlatSpec with BeforeAndAfter {
         opExec.close()
       })
   }
+
+  it should "keep a tuple under OR when only one predicate matches" in {
+    opDesc.predicates = List(
+      new FilterPredicate("int", ComparisonType.EQUAL_TO, "0"),
+      new FilterPredicate("string", ComparisonType.EQUAL_TO, "world")
+    )
+    opDesc.predicateCombinator = PredicateCombinator.OR
+    val opExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    opExec.open()
+    assert(opExec.processTuple(nonNullTuple, inputPort).nonEmpty)
+    opExec.close()
+  }
+
+  it should "drop a tuple under AND when only one predicate matches" in {
+    opDesc.predicates = List(
+      new FilterPredicate("int", ComparisonType.EQUAL_TO, "0"),
+      new FilterPredicate("string", ComparisonType.EQUAL_TO, "world")
+    )
+    opDesc.predicateCombinator = PredicateCombinator.AND
+    val opExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    opExec.open()
+    assert(opExec.processTuple(nonNullTuple, inputPort).isEmpty)
+    opExec.close()
+  }
+
+  it should "keep a tuple under AND when every predicate matches" in {
+    opDesc.predicates = List(
+      new FilterPredicate("int", ComparisonType.EQUAL_TO, "0"),
+      new FilterPredicate("string", ComparisonType.EQUAL_TO, "hello")
+    )
+    opDesc.predicateCombinator = PredicateCombinator.AND
+    val opExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    opExec.open()
+    assert(opExec.processTuple(nonNullTuple, inputPort).nonEmpty)
+    opExec.close()
+  }
+
+  it should "do nothing under AND when predicates is an empty list" in {
+    opDesc.predicates = List()
+    opDesc.predicateCombinator = PredicateCombinator.AND
+    val opExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    opExec.open()
+    assert(opExec.processTuple(allNullTuple, inputPort).isEmpty)
+    assert(opExec.processTuple(nonNullTuple, inputPort).isEmpty)
+    opExec.close()
+  }
+
+  it should "agree with OR on a single predicate under AND" in {
+    opDesc.predicates = List(new FilterPredicate("string", ComparisonType.EQUAL_TO, "hello"))
+    opDesc.predicateCombinator = PredicateCombinator.AND
+    val andExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    andExec.open()
+    assert(andExec.processTuple(nonNullTuple, inputPort).nonEmpty)
+    andExec.close()
+
+    opDesc.predicateCombinator = PredicateCombinator.OR
+    val orExec = new SpecializedFilterOpExec(objectMapper.writeValueAsString(opDesc))
+    orExec.open()
+    assert(orExec.processTuple(nonNullTuple, inputPort).nonEmpty)
+    orExec.close()
+  }
 }

docs/reference/operators/data-cleaning/filter.md

@@ -19,7 +19,7 @@
 
 ---
 title: "Filter"
-description: "Performs a filter operation using OR between multiple predicates"
+description: "Performs a filter operation using AND or OR between multiple predicates"
 category: "Data Cleaning"
 operator_type: "Filter"
 tags: [data-cleaning]
@@ -31,10 +31,11 @@ tags: [data-cleaning]
 
 | Property | Requirement | Type | Default | Description |
 |----------|-------------|------|---------|-------------|
-| Predicates | ✓ | List<Filter Predicate> | - | Multiple predicates in OR |
+| Predicates | ✓ | List<Filter Predicate> | - | Multiple predicates to be combined |
 | ↳ Attribute | ✓ | String | - |  |
 | ↳ Condition | ✓ | =, >, >=, <, <=, !=, is null,<br>is not null | - |  |
 | ↳ Value |  | String | - |  |
+| Combine Predicates With | ✓ | any (OR), all (AND) | any (OR) | How to combine multiple predicates |
 
 ### Output Ports