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>

Keep only rows where time_column >= watermark. Returns Ok(None) when nothing survives; passes through untouched while the watermark is uninitialised (< 0).

Errors

DataFusionError::Plan when time_column is missing or isn’t a Timestamp(_) (propagated from filter_batch_by_timestamp).

Trait Implementations

impl Debug for WatermarkDynamicFilter

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

Formats the value using the given formatter.