derive watermark for temporal filters with condition ts >/>= expr(now())

derive watermark for temporal filters with condition ts >/>= expr(now()) #2928