feat(windows): add internal filter operators

[Elia-Renzoni]

Summary

This PR introduces an internal collect filter inside event-driven window operators (state, count, sliding, and event-time).

Instead of buffering every row into memory and filtering at emit time the window now evaluates the WHERE condition upon row arrival and only buffers matching rows. Control logic (begin, emit, trigger, count) still evaluates every arriving row to maintain correct execution semantics.

By assigning WHERE predicates directly to WindowPlan.collectCondition, the standalone FilterPlan node is eliminated automatically via PushDownPredicate.

Key benefit: Reduces peak memory consumption from O(all rows) to O(matching rows) without breaking window state/transition semantics or requiring any SQL syntax changes.

Related Issue: #4099

[Elia-Renzoni]

The failure of TestSingleSQLWithEventTime doesn't seem related to this patch. However, the JMX test failure is caused by it. I'll investigate further and update the failing test.

[ngjaying]

@Elia-Renzoni We had just fixed lint errors due to lint version update. Now you can rebase on the latest master to resolve those repo lint errors.

[ngjaying]

We have fixed the merge tool install problem in the ci. Try to rebase on the latest master

Superseded: re-submitting the same review in English.

[ngjaying]

Requesting changes: the P0 below (agg-referencing WHERE pushed below its data dependency, verified end-to-end: base returns rows, this branch returns empty) must be fixed before merge. The other two are required cleanups in the same files.

[ngjaying]

P0: this pushdown moves the predicate below its data dependency and breaks aggregate-referencing WHERE.

In standard SQL, WHERE indeed cannot see aggregates — but eKuiper deliberately deviates here:

1. aggFuncChecker (analyzer.go) only forbids nested aggregates, NoAggFunc inside aggregate queries, and aggregates in GROUP BY. Aggregates in WHERE are allowed;
2. rewrite_planner.go:43 RewriteAggFunctionInWhere is dedicated machinery for exactly this shape: it extracts aggregate calls from Condition into AggFunc output fields $$agg_ref_N and rewrites the original site to bypass(...);
3. The old plan spells this out (see the pre-change expectation in TestExplainAggInWhere): Project > Filter(a > bypass($$agg_ref_0)) > AggFunc > Window — the Filter sits above AggFunc, where the reference is bound.

This PR makes PushDownPredicate swallow the condition unconditionally and return nil (eliminating FilterPlan), while AggFuncPlan has no PushDownPredicate of its own to intercept it. The new plan is Project > AggFunc > Window(collectCondition), and collect is evaluated per input row inside the window, where $$agg_ref_0 can never be bound. collectConditionMatch returns (false, nil) for unbound fields, so rows are silently dropped with no error.

Verified end-to-end with the same rule (CountWindow(2)):

SELECT * FROM demoE2 WHERE temp > avg(temp) GROUP BY CountWindow(2)

• base: returns {temp:27.5} (first window (27.5, 25.5), avg=26.5)
• this branch: returns empty (reproduces the regression)

Suggested fix: refuse to absorb predicates referencing downstream-produced fields ($$agg_ref_* / bypass(...)-wrapped; general principle: never absorb anything that cannot be bound at the new location) in this PushDownPredicate, leaving that part to the upstream FilterPlan. Predicates over pure input columns keep pushing down, so the memory goal is unaffected. Also please cover the SQL above with a runtime regression test — TestExplainAggInWhere only asserts plan shape and cannot catch this class of issue.

[ngjaying]

P2.1: restructure this branch around the invariant below instead of just adding a comment.

There is exactly one correct behavior here (it is also the contract stated in the TestCountWindowCollectCondition comment): a COUNT window's trigger count must observe every row, and narrowing happens only at emit time in nextCountWindow; only non-COUNT windows filter at ingest. The current var filterMatch bool + two append sites reads as if COUNT could double-append, plus the var err shadowing. Make the branches mutually exclusive so the three facts are pinned down:

if o.window.Type == ast.COUNT_WINDOW {
    // Count trigger counts every row; narrowing happens at emit in nextCountWindow.
    inputs = append(inputs, d)
} else if match, err := collectConditionMatch(ctx, d, o.window.CollectCondition, o.name); err != nil {
    o.onError(ctx, err) // eval error: report and drop the row
} else if match {
    inputs = append(inputs, d)
}

Key points: COUNT appends unconditionally; eval error = report + drop (spelled out so nobody later "fixes" it); normal mismatch = silently drop.

[ngjaying]

P2.2: remove both empty-window skips and keep the original emit behavior.

This gate and the len(tsets.Content) == 0 → continue in execProcessingWindow (~L452) are new "skip instead of emit when nothing matched" behavior. Neither is required for the memory goal — not buffering non-matching rows already saves the memory. Skipping emission changes trigger semantics: downstream window count, timing, and records_out_total all change (the forced TestEventWindow metric change 5→2 is the evidence).

Requirement: delete both gates so trigger timing and emission cardinality stay exactly as before (narrowed content is the point of the feature; empty windows still emit).