Struct WatermarkDynamicFilter 

pub struct WatermarkDynamicFilter { /* private fields */ }

Dynamic filter that drops rows older than the current watermark.

Holds a shared watermark atomic (same as super::watermark_udf::WatermarkUdf) and a monotonic generation counter that increments on each watermark advance. The generation lets downstream consumers detect stale state without comparing full watermark values.

Implementations

impl WatermarkDynamicFilter

pub fn new( watermark_ms: Arc<AtomicI64>, generation: Arc<AtomicU64>, time_column: String, ) -> Self

Creates a new watermark filter.

Arguments

• watermark_ms - Shared atomic holding the current watermark in epoch milliseconds. Values < 0 mean “uninitialized”.
• generation - Monotonic counter incremented on each advance.
• time_column - Name of the event-time column in record batches.

pub fn advance_watermark(&self, new_ms: i64)

Advances the watermark if new_ms exceeds the current value.

On a successful advance the generation counter is incremented. No-op when new_ms <= current.

pub fn generation(&self) -> u64

Returns the current generation (monotonically increasing).

pub fn watermark_ms(&self) -> i64

Returns the current watermark in epoch milliseconds.

pub fn filter_batch( &self, batch: &RecordBatch, ) -> Result<Option<RecordBatch>, DataFusionError>

Filters a record batch, keeping only rows where time_column >= watermark.

Returns Ok(None) when all rows are filtered out. When watermark < 0 (uninitialized), all rows pass through.

Handles both Int64 (epoch millis) and Timestamp(Millisecond, _) columns.

Errors

Returns an error if the time column is missing or has an unsupported type.

Trait Implementations

impl Debug for WatermarkDynamicFilter

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.