Feature: Early Window Expiration with Triggers

[gwaramadze]

Early Window Expiration with Triggers

This PR adds support for early window expiration via callback triggers for tumbling and hopping windows.

Feature Overview

Windows can now be expired early based on custom logic by providing before_update or after_update callbacks when defining a window. When these callbacks return True, the window is immediately expired rather than waiting for its natural expiration time.

API

Two new optional parameters added to tumbling_window() and hopping_window():

• before_update: Callback invoked before updating the window state

  • Receives the current aggregated value (or collection) before the new value is applied
  • Useful for preventing a value from being included if a threshold would be exceeded

• after_update: Callback invoked after updating the window state

  • Receives the updated aggregated value (or collection) after the new value is applied
  • Useful for expiring a window once a threshold is reached

Both callbacks receive: (aggregated, value, key, timestamp, headers) -> bool

Examples

After Update - Expire when sum exceeds threshold:

def trigger_on_sum(aggregated, value, key, timestamp, headers):
    return aggregated > 100  # Expire when sum exceeds 100

sdf = app.dataframe(topic)
sdf = sdf.tumbling_window(duration_ms=60000, after_update=trigger_on_sum).sum().final()

Before Update - Prevent value from exceeding threshold:

def trigger_before_exceeding(aggregated, value, key, timestamp, headers):
    return (aggregated + value) > 100  # Expire before sum would exceed 100

sdf = app.dataframe(topic)
sdf = sdf.tumbling_window(duration_ms=60000, before_update=trigger_before_exceeding).sum().final()

With collect operations:

def trigger_on_count(aggregated, value, key, timestamp, headers):
    return len(aggregated) >= 10  # Expire when 10 items collected

sdf = app.dataframe(topic)
sdf = sdf.tumbling_window(duration_ms=60000, after_update=trigger_on_count).collect().final()

Implementation Details

• Triggered windows are moved to the expired windows list and deleted from state
• Collection values are not immediately deleted; normal expiration handles cleanup
• This means triggered windows can be "resurrected" if new data arrives before natural expiration
• Works with both aggregation operations (.sum(), .mean(), etc.) and .collect()
• Fully compatible with overlapping hopping windows