Function buckets_with_watermarks 

pub fn buckets_with_watermarks<S>(
    stream: S,
    range: Range<u64>,
    bucket_size: u64,
    direction: ScanDirection,
) -> impl Stream<Item = BitmapScanResult<Watermarked<(u64, RoaringBitmap)>>> + Send + 'static
where
    S: Stream<Item = BucketItem> + Send + 'static,

Convenience adapter: wrap a single raw BucketStream into a WatermarkedBucketStream with one Watermark(post_bucket) after each bucket plus one final at the range terminus on EOF.

The DNF eval pipeline budgets and merges leaves itself; this helper is for backend-side consumers (e.g. RocksDB single-dimension scans) that want bucket-level output without the full eval machinery.