Hi @gkapri,

Thanks for the detailed writeup. I can see from the thread that you have already tried the literal variable approach that works in a notebook but does not work inside a Lakeflow Spark Declarative Pipeline (SDP). Let me explain what is happening and how to address it.

WHY PARTITION PRUNING DOES NOT WORK WITH readStream IN SDP

The root cause is how Structured Streaming reads Delta tables versus how batch reads work. These are fundamentally different processing models:

1. Batch reads (spark.read) evaluate the full query plan at execution time, including partition filters. When you filter on a partition column with a literal value, Spark can push that filter down into the scan and skip entire partition directories. This is why your notebook test worked.

2. Streaming reads (spark.readStream) process the Delta transaction log incrementally. The stream tracks which files have already been processed using a checkpoint, and on each micro-batch it picks up NEW files from the log. The key point is that readStream does not apply your downstream filters at the file-selection stage. Instead, it reads all new files from the transaction log first, and then your filter is applied after the data is already read into memory. This is by design, because the streaming engine needs to maintain exactly-once processing guarantees by tracking every file.

3. Non-deterministic functions in SDP add another layer. Inside a Lakeflow Spark Declarative Pipeline (SDP), expressions like current_date() are evaluated at pipeline plan time. Even converting to a literal variable with spark.sql() may not resolve the issue because the SDP execution model handles variable evaluation differently than an interactive notebook.

So to summarize: the 37 million rows you see being read is the stream reading all new files from the bronze table's transaction log, and then your processing_date filter is applied after the fact, resulting in only 24,722 rows being kept.

WHAT ABOUT AUTO CDC (APPLY CHANGES)?

I noticed from the follow-up comments that you and @Anish_2 are using AUTO CDC (formerly APPLY CHANGES) for the silver layer, which requires a streaming source. This means you cannot simply switch to a batch read (spark.read) to get partition pruning. This is a real constraint.

Here are your options:

OPTION 1: LET THE STREAM HANDLE INCREMENTALITY (RECOMMENDED)

Instead of filtering by processing_date to limit the data, rely on the streaming checkpoint to handle incrementality for you. This is actually the intended design pattern for streaming pipelines:

import dlt
from pyspark.sql.functions import col

@Dlt.table()
def silver_table():
return spark.readStream.table("LIVE.bronze_table")

The stream will only process NEW files that arrived since the last pipeline run. After the initial backfill (which will read all historical data once), subsequent runs will only pick up incremental data. This eliminates the need for the date filter entirely because the checkpoint ensures you never reprocess old data.

If your concern is the initial backfill reading too much data, you can control the rate with:

@Dlt.table()
def silver_table():
return (
spark.readStream
.option("maxFilesPerTrigger", 100)
.table("LIVE.bronze_table")
)

This throttles how many files are processed per micro-batch during the initial catch-up.

OPTION 2: USE skipChangeCommits WITH A TARGETED STREAM

If your bronze table receives updates or deletes and you only want appends, you can combine streaming with skipChangeCommits:

@Dlt.table()
def silver_table():
return (
spark.readStream
.option("skipChangeCommits", "true")
.table("LIVE.bronze_table")
)

This skips any commits that contain updates or deletes, processing only append commits.

OPTION 3: USE A MATERIALIZED VIEW FOR THE FILTERED LAYER

If you truly need partition pruning with a date filter, use a materialized view for that transformation step, since materialized views use batch reads:

CREATE OR REFRESH MATERIALIZED VIEW silver_filtered
AS
SELECT *
FROM LIVE.bronze_table
WHERE processing_date >= current_date() - INTERVAL 2 DAYS

Materialized views process data using batch reads, so partition pruning works as expected. The trade-off is that materialized views fully recompute on each refresh (though Databricks does apply incremental refresh optimizations where possible).

However, if you need to feed this into an AUTO CDC target, you would need an additional streaming table layer reading from this materialized view.

OPTION 4: USE startingTimestamp TO LIMIT INITIAL LOAD

If your main concern is the initial pipeline run reading too much historical data, you can set a starting point for the stream:

@Dlt.table()
def silver_table():
return (
spark.readStream
.option("startingTimestamp", "2026-02-01")
.table("LIVE.bronze_table")
)

This tells the stream to only process changes from that timestamp forward, skipping all earlier data. After the initial run, the checkpoint takes over and only new data is processed.

ADDITIONAL PERFORMANCE TIPS

1. Enable auto-compaction on your bronze table. If your bronze streaming table creates many small files, queries (and downstream streams) will be slower. Lakeflow Spark Declarative Pipelines (SDP) enables auto-optimization by default, but verify it is working by checking the file count:

DESCRIBE DETAIL your_catalog.your_schema.bronze_table

2. Consider liquid clustering instead of partitioning. For new tables, liquid clustering (CLUSTER BY) often outperforms traditional partitioning because it handles data skipping more flexibly:

CREATE OR REFRESH STREAMING TABLE bronze_table
CLUSTER BY (processing_date)
AS SELECT ...

3. Run OPTIMIZE periodically on the bronze table if you see many small files accumulating, or enable predictive optimization at the catalog or schema level.

KEY TAKEAWAY

The fundamental insight is that streaming reads and batch reads use different mechanisms for data access. Streaming reads track the transaction log incrementally and do not apply partition pruning on downstream filters. The recommended pattern for streaming pipelines is to rely on checkpointing for incrementality rather than date-based filters. Your date filter still works for correctness (it correctly filters the output), but it cannot reduce the amount of data read at the source scan level in a streaming context.

RELEVANT DOCUMENTATION

Delta Lake as a streaming source:
https://docs.databricks.com/en/structured-streaming/delta-lake.html

Lakeflow Spark Declarative Pipelines (SDP) streaming tables:
https://docs.databricks.com/en/ldp/concepts.html

AUTO CDC (APPLY CHANGES) reference:
https://docs.databricks.com/en/ldp/cdc.html

Data skipping and partition pruning:
https://docs.databricks.com/en/delta/data-skipping.html

Liquid clustering:
https://docs.databricks.com/en/delta/clustering.html

File size tuning and auto-compaction:
https://docs.databricks.com/en/delta/tune-file-size.html

Hope this helps you and @Anish_2! The short answer is that this is expected behavior for streaming reads, and the best approach is to let the streaming checkpoint handle incrementality rather than using date filters to limit the scan.

* This reply used an agent system I built to research and draft this response based on the wide set of documentation I have available and previous memory. I personally review the draft for any obvious issues and for monitoring system reliability and update it when I detect any drift, but there is still a small chance that something is inaccurate, especially if you are experimenting with brand new features.