feat(workflow-operator): let Filter combine predicates with AND - #8484
feat(workflow-operator): let Filter combine predicates with AND#8484gupta-sahil01 wants to merge 1 commit into
Conversation
The Filter operator hard-coded OR semantics, so users needing AND had to chain multiple Filter operators. Add a PredicateCombinator enum exposed as a descriptor property, defaulting to OR so existing workflows are unaffected — a stored desc without the field deserializes to OR, so no migration is needed. Empty predicate lists keep dropping every tuple in both modes; a bare forall would have flipped that to passing everything.
Automated Reviewer SuggestionsBased on the
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #8484 +/- ##
============================================
- Coverage 93.94% 93.94% -0.01%
- Complexity 4827 4833 +6
============================================
Files 1209 1210 +1
Lines 49705 49717 +12
Branches 6074 6075 +1
============================================
+ Hits 46695 46706 +11
Misses 1522 1522
- Partials 1488 1489 +1
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 419 | 0.256 | 24,036/31,032/31,032 us | 🔴 +14.7% / 🔴 +92.3% |
| 🟢 | bs=100 sw=10 sl=64 | 925 | 0.565 | 105,845/138,018/138,018 us | 🟢 -7.6% / 🔴 +27.0% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,086 | 0.663 | 924,025/990,868/990,868 us | ⚪ within ±5% / 🟢 -7.6% |
Baseline details
Latest main ded7ba1 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 419 tuples/sec | 453 tuples/sec | 755.28 tuples/sec | -7.5% | -44.5% |
| bs=10 sw=10 sl=64 | MB/s | 0.256 MB/s | 0.276 MB/s | 0.461 MB/s | -7.2% | -44.5% |
| bs=10 sw=10 sl=64 | p50 | 24,036 us | 20,949 us | 12,957 us | +14.7% | +85.5% |
| bs=10 sw=10 sl=64 | p95 | 31,032 us | 32,247 us | 16,134 us | -3.8% | +92.3% |
| bs=10 sw=10 sl=64 | p99 | 31,032 us | 32,247 us | 20,333 us | -3.8% | +52.6% |
| bs=100 sw=10 sl=64 | throughput | 925 tuples/sec | 950 tuples/sec | 980.1 tuples/sec | -2.6% | -5.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.565 MB/s | 0.58 MB/s | 0.598 MB/s | -2.6% | -5.6% |
| bs=100 sw=10 sl=64 | p50 | 105,845 us | 103,451 us | 101,894 us | +2.3% | +3.9% |
| bs=100 sw=10 sl=64 | p95 | 138,018 us | 149,307 us | 108,718 us | -7.6% | +27.0% |
| bs=100 sw=10 sl=64 | p99 | 138,018 us | 149,307 us | 122,482 us | -7.6% | +12.7% |
| bs=1000 sw=10 sl=64 | throughput | 1,086 tuples/sec | 1,101 tuples/sec | 1,011 tuples/sec | -1.4% | +7.4% |
| bs=1000 sw=10 sl=64 | MB/s | 0.663 MB/s | 0.672 MB/s | 0.617 MB/s | -1.3% | +7.4% |
| bs=1000 sw=10 sl=64 | p50 | 924,025 us | 905,258 us | 996,422 us | +2.1% | -7.3% |
| bs=1000 sw=10 sl=64 | p95 | 990,868 us | 986,338 us | 1,037,670 us | +0.5% | -4.5% |
| bs=1000 sw=10 sl=64 | p99 | 990,868 us | 986,338 us | 1,072,152 us | +0.5% | -7.6% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,477.77,200,128000,419,0.256,24036.19,31031.70,31031.70
1,100,10,64,20,2161.79,2000,1280000,925,0.565,105845.27,138018.28,138018.28
2,1000,10,64,20,18413.24,20000,12800000,1086,0.663,924025.35,990867.74,990867.74|
/request-review @parshimers |
What changes were proposed in this PR?
The
Filteroperator hard-codes OR semantics — a tuple passes as soon as any onepredicate matches. Users who need AND (e.g.
age > 30 AND country = "US") have tochain multiple
Filteroperators, which adds workflow clutter and an extra hop percondition.
This PR makes the combination configurable:
PredicateCombinatorenum (OR→"any (OR)",AND→"all (AND)"),following the
@JsonValue/@JsonCreatorpattern already used byComparisonTypein the same package.predicateCombinatorproperty onSpecializedFilterOpDesc, defaulting toOR. The UI dropdown renders automatically from the JSON schema, so there are nofrontend changes.
SpecializedFilterOpExecnow selectsforallfor AND andexistsfor OR.Backward compatibility: the default is
OR, and a stored descriptor without thefield deserializes to
OR, so existing workflows behave exactly as before and nomigration is needed.
One behavioral note for reviewers: an empty predicate list continues to drop every
tuple in both modes. A bare
forallreturnstrueon an empty list, which wouldhave silently flipped an empty Filter from "drop everything" to "pass everything", so
the AND branch is guarded with
predicates.nonEmpty &&. There is a test covering this.Nested predicate trees (e.g.
(A AND B) OR C) are intentionally out of scope — theyneed more extensive UI work and are noted as a follow-up in the issue.
Any related issues, documentation, discussions?
Closes #6939
Operator documentation updated in the same PR:
docs/reference/operators/data-cleaning/filter.md.How was this PR tested?
New and updated unit tests, all run locally via
sbt "WorkflowOperator/testOnly *filter*"— 67 tests across 7 suites, all passing.PredicateCombinatorSpec(new): wire names viagetName, case-insensitivefromString, rejection of an unknown name, and the Jackson round-trip.SpecializedFilterOpDescSpec: the property defaults toOR; a hand-writtendescriptor JSON with no
predicateCombinatorfield deserializes toOR(this is the backward-compatibility claim above); an explicit
ANDsurvives theround-trip through the polymorphic
LogicalOpbase;operatorInfoadvertises thenew behavior.
SpecializedFilterOpExecSpec: with two predicates where only one matches, ORkeeps the tuple and AND drops it; AND keeps it when both match; AND with an empty
predicate list drops everything; AND and OR agree on a single predicate. A
beforeblock resets the shared descriptor between tests so a mode set in onetest cannot leak into the ones after it.
The empty-list guard was additionally verified by mutation: removing
predicates.nonEmpty &&from the executor makes exactly one test fail, confirmingthe case is genuinely covered rather than incidentally passing.
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)