In your scenario using Medallion Architecture with Delta tables as both streaming source and sink, it is important to understand Spark Structured Streaming behavior and performance characteristics, especially with joins and memory usage. Here is a direct, actionable analysis based on your detailed setup and observed metrics.

How Delta Table as Source Works in Streaming

When you use a Delta table as a streaming source (spark.readStream.format("delta").table("table_name")), Spark Structured Streaming tracks new data files appended to the Delta log. On initial start, unless you specify "startingVersion", Spark reads a full snapshot of the table (all rows). For every subsequent trigger, only new files/data are processed as micro-batch increments.

Key behavior:

• On the first trigger, the read covers the entire table.

• On subsequent triggers, it reads only new appended files (new data) since the last checkpoint.

Why Joins Appear to Process All Rows Each Trigger

What you observed:

• On the initial run, df1-df3 have all current rows, π1, π2, π3. Join+write occur.

• After appending new rows, and triggering, you see df1-df3 now have π1+n, π2+m, π3+k rows, always including all past data.

This happens because:

• Each trigger’s streaming DataFrame by default represents all unprocessed data (since last checkpoint), but should not include all rows from the physical table every time, unless you configured "availableNow=true" -- in which case, you get a bounded (batch-like) execution and a full scan occurs.

• If you run with .trigger(availableNow=True), the micro-batch will process the full available data each time; this is intended for one-off refreshes, not continuous streaming.

• For normal .trigger(processingTime='interval'), each trigger sees just new data, leading to much lower processing/memory.

Memory Growth: The Real Concern with Joins

Structured Streaming joins (especially inner/outer) build in-memory state for joined keys. If you do not set watermark or retention duration, the in-memory state grows unbounded, because Spark assumes all historic data may still match a late-arriving record.

• The state store (see "numRowsTotal", "memoryUsedBytes" in your metrics) accumulates old data.

• Each symmetric hash join instance holds all seen join keys until it can safely evict data, which only happens with event time watermarks.

Without Watermarks:

• Memory usage increases with each new batch, as old keys are never "timed out" or removed.

• This can easily lead to OOM (Out Of Memory) errors or excessive state store growth, especially in large tables.

With Watermarks:

• Watermarks let Spark know “data older than this is safe to drop from memory,” so the state store remains bounded.

• For append-only tables and inner joins, set watermarks on the event-time column and ideally design joins so only a recent window of keys must be kept in memory.

Recommendations for Your Scenario

1. Use streaming mode (not availableNow) for continuous ingest

• Only use availableNow=True for batch catch-up; for continuous, classic streaming mode is more memory-efficient.

2. Always set watermarks in streaming joins

• Example:

   

  python

  df1 = df1.withWatermark("event_time_col", "30 minutes") df2 = df2.withWatermark("event_time_col", "30 minutes") result = df1.join(df2, ...)

  Adjust the watermark according to your use case latency/late-arrival tolerances.

3. Monitor state store metrics

• Regularly check memoryUsedBytes and numRowsTotal for each stateful operator. Unbounded growth signals missing watermark or join design issues.

4. Partition/cluster source Delta tables on key columns (for large tables)

• This optimizes both streaming read and joins.

5. For batch use, use availableNow and checkpointing

• But be aware this will re-scan all available data, so it’s not for true streaming.

Relevant Documentation & Key Options

• [Delta streaming source documentation, start options, and watermarks]

• Databricks: "Best Practices for Using Structured Streaming"

Summary Table: Streaming Join Behavior

If you continue joins without watermarks, expect state store memory to grow linearly with data size, which is unsustainable. Set watermarks and optimize triggers for production.