Watermarking without aggregation with Structured Streaming

[peay]

Hello,

I am trying to use watermarking without aggregation, to filter out records that 
are just too late, instead of appending them to the output. My understanding is 
that aggregation is required for `withWatermark` to have any effect. Is that 
correct?

I am looking for something along the lines of

```
df.withWatermark("ts", ...).filter(F.col("ts") <F.getCurrentWatermark())
```

Is there any way to get the watermark value to achieve that?

Thanks!