Late doesn't mean queued: the watermark's silent drop contract

A watermark is a state boundary, not a waiting room

withWatermark("event_time", "10 minutes") does not hold every row for ten minutes. Structured Streaming tracks the greatest event_time it has observed and derives the watermark from that value:

The input data advances the watermark; the wall clock does not. If a source stalls, the watermark freezes. If a burst of fresh events arrives, it can push the watermark forward past older stragglers. Records within the threshold are guaranteed to be processed. Records older than it are outside the guarantee; they are dropped from the watermarked stateful result once older than the watermark.

That contract matters in append-mode stateful operations: windowed aggregations, dropDuplicatesWithinWatermark, and outer stream-stream joins. There is no automatic dead-letter queue, warning, or exception for a late record that misses cleaned-up state. If the data must be inspected, you need a separate quarantine or reconciliation path.

from pyspark.sql.functions import window, col

hourly = (
    spark.readStream.table("bronze_clicks")
    .withWatermark("event_time", "10 minutes")
    .groupBy(window(col("event_time"), "1 hour"))
    .count()
)

Illustrative scenario — all times and figures are illustrative

The query uses a 10-minute watermark. Events stamped 10:02, 10:05, and 10:08 arrive. Then an out-of-order fresh burst includes an event at 10:35, moving the maximum observed event time to 10:35 and the watermark to 10:25. A straggler stamped 10:20 arrives after that boundary.

Primary reference: Databricks docs: apply watermarks to control data processing thresholds

The first run is uniquely dangerous

When a stateful query first starts against an existing Delta table, it consumes the table's initial snapshot. By default, those files are processed in last-modified order, not event-time order. A recently rewritten file full of old events can therefore arrive after newer files have already advanced the watermark. The old rows now look late and can be mass-dropped on the first run.

Enable withEventTimeOrder on the stream before it begins:

from pyspark.sql.functions import window, col

stream = (
    spark.readStream
    .option("withEventTimeOrder", "true")
    .table("bronze_clicks")
    .withWatermark("event_time", "10 minutes")
    .groupBy(window(col("event_time"), "1 hour"))
    .count()
)

The option divides the initial snapshot's event-time range into buckets. Each micro-batch processes one bucket in event-time order. maxFilesPerTrigger and maxBytesPerTrigger still constrain each batch approximately. The feature is available in Databricks Runtime 11.3 LTS and above and has been generally available since February 2025.

# Cluster-wide switch for all applicable streaming queries
spark.databricks.delta.withEventTimeOrder.enabled true

Constraints that change the recovery plan

• You cannot change withEventTimeOrder while the initial snapshot is actively processing without deleting the checkpoint.
• You cannot downgrade to a runtime that lacks the feature until the initial snapshot completes; otherwise wait, or delete the checkpoint and restart.
• It is not supported when the event-time column is generated and non-projection transformations sit between the Delta source and the watermark.
• It is not supported when more than one Delta source sits behind a watermark in the query.

Initial snapshot processing can be slower because each micro-batch scans the snapshot for its event-time bucket. Mitigate that cost by using a real Delta source column as event time so data skipping can apply, and partitioning the table by event time where that layout fits the workload.

Two production gotchas

Use this when

• Stateful streaming operations need bounded state: append-mode windowed aggregations, deduplication, or stream-stream joins.
• The source has known, bounded lateness.
• You need protection from unbounded state growth, rising latency, and memory errors.

Avoid this when

• You need a data-quality or completeness gate. A watermark drops; it does not quarantine.
• You need audit-grade counts.
• The delay would be tiny compared with the source's real-world lateness.
• You are considering multipleWatermarkPolicy=max but cannot accept loss from the slowest stream.

“My batch recount over the source table disagrees with my streaming output — is the stream wrong?”

Probably incomplete, not wrong. A batch query reads the source rows that exist now. A stateful stream with a watermark applies a time-bounded processing contract: records that arrive beyond that boundary may be dropped. The disagreement often appears after restarts, backfills, or a first run that consumed the initial snapshot in file modification order.

Prove it by reconciling source and output per event-time window. If missing events cluster in the first windows after a restart or backfill, the watermark is the prime suspect.

-- Late-drop audit: source events vs. streamed output, per window
WITH src AS (
  SELECT date_trunc('hour', event_time) AS hr, COUNT(*) AS source_events
  FROM bronze_clicks
  GROUP BY 1
),
out AS (
  SELECT date_trunc('hour', window.start) AS hr, SUM(cnt) AS streamed_events
  FROM click_hourly_counts
  GROUP BY 1
)
SELECT s.hr,
       s.source_events,
       COALESCE(o.streamed_events, 0) AS streamed_events,
       s.source_events - COALESCE(o.streamed_events, 0) AS missing_events
FROM src s
LEFT JOIN out o USING (hr)
ORDER BY s.hr;

This audit assumes click_hourly_counts retains one or more count rows that can be summed per hourly window. Adapt the output-side aggregation if your sink writes one final row per window or uses a different count column.