Add option to combine Filter operator predicates with AND instead of only OR #6939

Feature Summary

The Filter operator accepts a list of predicates, but the combination logic is hard-coded to OR. In SpecializedFilterOpExec.scala:28:

setFilterFunc((tuple: Tuple) => desc.predicates.exists(_.evaluate(tuple)))

exists means a tuple passes as soon as any one predicate matches. There is no way to express a conjunction.

As a result, a user who wants age > 30 AND country = "US" cannot do it in a single Filter operator. They have to chain one Filter per predicate:

Source ──> Filter(age > 30) ──> Filter(country = "US") ──> ...

This is awkward for a few reasons:

• The multi-predicate UI is misleading — the properties panel invites you to add several predicates, but silently gives you the opposite semantics of what many users expect from a "filter with conditions" form.
• Workflows get visually noisy: an N-way conjunction needs N operators on the canvas instead of one.
• It costs N operator boundaries (N physical ops, extra serialization/messaging hops between workers) to express something the operator could evaluate inline.
• Mixed intent (A AND B, then OR C) requires a fan-out/Union construction that is far harder to read than the equivalent single form.

Prior related issue: #3455 documented the OR behavior in the operator description. This issue proposes making the behavior configurable rather than only documented.

Proposed Solution or Design

Add a top-level property on SpecializedFilterOpDesc that selects how the predicates are combined, defaulting to OR so existing saved workflows are unaffected.

1. New enum (e.g. common/workflow-operator/.../filter/PredicateCombinator.java, alongside the existing ComparisonType.java), using the same @JsonValue / @JsonCreator pattern so the UI dropdown shows friendly labels:

public enum PredicateCombinator {
    OR("any (OR)"),
    AND("all (AND)");
    ...
}

2. Descriptor — SpecializedFilterOpDesc.scala, following the defaultValue + @JsonSchemaTitle convention already used elsewhere (e.g. ProjectionOpDesc.isDrop):

@JsonProperty(value = "combinator", required = true, defaultValue = "any (OR)")
@JsonSchemaTitle("Match")
@JsonPropertyDescription("whether a tuple must satisfy any predicate (OR) or all predicates (AND)")
var combinator: PredicateCombinator = PredicateCombinator.OR

@JsonProperty(value = "predicates", required = true)
@JsonPropertyDescription("predicates combined according to the selected match mode")
var predicates: List[FilterPredicate] = List.empty

Because the field is initialized to OR, workflow JSON persisted before this change deserializes to the current behavior — no migration needed.

3. Executor — SpecializedFilterOpExec.scala:

setFilterFunc((tuple: Tuple) =>
  desc.combinator match {
    case PredicateCombinator.AND => desc.predicates.forall(_.evaluate(tuple))
    case _                       => desc.predicates.exists(_.evaluate(tuple))
  }
)

Note the empty-list edge case differs between the two (exists on empty is false, forall on empty is true); predicates is required = true, but the behavior should be pinned down by a test either way.

4. Metadata / docs

• Update operatorInfo description in SpecializedFilterOpDesc.scala:56 — it currently hard-codes "Performs a filter operation using OR between multiple predicates".
• Regenerate / update docs/reference/operators/data-cleaning/filter.md, which repeats the same OR wording.

5. Tests

• SpecializedFilterOpExecSpec.scala — add AND cases (all match, some match, none match) next to the existing OR cases.
• SpecializedFilterOpDescSpec.scala — cover the default value and round-trip serialization of the new property.

The frontend needs no change: the properties panel is generated from the operator's JSON schema, so the new field renders as a dropdown automatically.

Possible follow-up (out of scope here): a fully nested predicate tree (groups of AND/OR) would cover mixed expressions like (A AND B) OR C. That is a much larger UI change; a flat all/any toggle covers the common case and is a strict superset of today's behavior. Happy to discuss if committers prefer to go straight to the general form.

I'd like to work on this — please let me know if the approach looks reasonable before I open a PR.

Affected Area

Workflow Engine (Amber)