On this page

• Recompute the recent rows of a streaming table
  • Recompute a window, keep the history
  • Before you start
  • Declare the REPLACE WHERE write
  • Generate the pipeline
  • Read what Lakehouse Plumber wrote
  • Keep the history safe
  • Keep refreshes incremental
  • Read the table downstream
  • What you just did
  • What’s next
    • Syntax and options

Recompute the recent rows of a streaming table

Some data keeps changing for a while after it lands. Orders are corrected, cancelled and re-priced for a few days after they are placed, late records arrive, and an upstream system reprocesses yesterday. After that the rows settle. You want a table that keeps the whole history, where the recent days are recomputed from the source on every refresh and the older days are left alone.

That is what a Databricks REPLACE WHERE flow does, and mode: replace_where generates one.

Recompute a window, keep the history

You give the write a SQL predicate, such as “orders from the last 7 days”. On every refresh the REPLACE WHERE flow deletes the target rows that match the predicate and recomputes them by re-running the source query for that range. Rows that don’t match are left exactly as they are. The source is read in batch, so you process new rows without managing streaming concepts such as watermarks.

A worked example. The predicate is order_date >= date_add(current_date(), -7), and the refresh runs on 2026-06-10, so it covers 2026-06-03 onwards:

BEFORE — target rows
  order_id  order_date  amount
  101       2026-05-20  40
  205       2026-06-05  12
  206       2026-06-06  30
  207       2026-06-08  18

SOURCE — what the bronze table holds now
  order_id  order_date  amount
  101       2026-05-20  45     # corrected, but outside the predicate
  205       2026-06-05  12
  207       2026-06-08  20     # corrected
  208       2026-06-10  25     # new
  # 206 was deleted upstream

AFTER — target rows, via REPLACE WHERE
  order_id  order_date  amount
  101       2026-05-20  40     # untouched: outside the predicate
  205       2026-06-05  12
  207       2026-06-08  20
  208       2026-06-10  25
  # 206 is gone: the rows inside the predicate were recomputed from the source

Inside the predicate the target now matches the source: 207 was corrected, 208 was added and 206 was deleted, with no delete event needed. Outside it, nothing changed, including the correction to order 101. That is the trade-off: once a row is older than the window, the table stops following the source for it. Pick a window that covers how long your data keeps changing.

The batch read is what sets it apart from its streaming sibling, mode: replace:

Aspect	mode: replace_where (REPLACE WHERE)	mode: replace (REPLACE USING)
Rows replaced	Every target row that matches a predicate	The target rows of each key that a batch carries, when the batch’s sequence_by value is newer
Source read	Batch; the predicate’s range is recomputed	Stream; the new rows are applied
Rows outside	Never touched	Keys absent from the batch are never touched

Before you start

You need a Lakehouse Plumber project (see the Get Started course for lhp init) and a source table. This guide reads a bronze orders table that another pipeline lands, with an order_date column for the predicate.

Three rules are worth knowing before you write it:

• The source is read in batch. A streaming source is rejected.

• The target is served by this single flow. No other write action can use its table, and Lakehouse Plumber always creates the table for you.

• Databricks refreshes the flow incrementally only on serverless compute.

Declare the REPLACE WHERE write

The flowgroup has two actions: a load that reads the orders in batch into a view, and a write whose write_target is a streaming_table in mode: replace_where. Create pipelines/orders_recent.yaml:

pipelines/orders_recent.yaml

# Keep a silver table of orders whose last 7 days are recomputed on every refresh.
# Action chain: load (delta, batch read of the bronze orders) -> write (streaming_table, mode: replace_where).
#
# On each refresh the REPLACE WHERE flow deletes the target rows that match the
# predicate (the last 7 days by order_date) and recomputes them from the source.
# Rows outside the predicate (older orders) are left as they are.
pipeline: silver_orders
flowgroup: orders_recent

actions:
  # 1. Load: a batch read of the bronze orders table. A REPLACE WHERE flow reads
  #    its source in batch, so this load must not stream.
  - name: load_orders
    type: load
    readMode: batch
    source:
      type: delta
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: orders
    target: v_orders

  # 2. Write: recompute the last 7 days of orders on every refresh. The change
  #    data feed is turned on so a downstream pipeline can follow the table
  #    (see pipelines/orders_recent_mirror.yaml).
  - name: write_orders_recent
    type: write
    source: v_orders
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders_recent
      mode: replace_where
      table_properties:
        delta.enableChangeDataFeed: "true"
      replace_where_config:
        replace_where: "order_date >= date_add(current_date(), -7)"

• readMode: batch on the load makes v_orders a batch view. A delta load reads in batch by default; it is spelled out here because a streaming view would be rejected.

• mode: replace_where is the switch.

• replace_where_config.replace_where is the predicate, a SQL string that Lakehouse Plumber writes into the generated code as it is. ${...} tokens in it are resolved per environment first.

• table_properties turns on the change data feed. Nothing in this pipeline needs it; a downstream pipeline uses it at the end of this guide.

Generate the pipeline

Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ silver_orders  0 files
✓ validate (0.29s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ silver_orders  1 file
✓ generate (0.31s)
✓ format (0.01s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.3s

validate resolves the tokens and checks the write: that the replace_where_config block holds a non-blank predicate, that the view it reads is a batch view, and that no other write action in the pipeline targets the table.

Read what Lakehouse Plumber wrote

Open generated/dev/silver_orders/orders_recent.py:

generated/dev/silver_orders/orders_recent.py

# Generated by LakehousePlumber
# Pipeline: silver_orders
# FlowGroup: orders_recent

from pyspark import pipelines as dp

# Pipeline Configuration
PIPELINE_ID = "silver_orders"
FLOWGROUP_ID = "orders_recent"


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_orders():
    """Delta source: dev_catalog.bronze.orders"""
    df = spark.read.table("dev_catalog.bronze.orders")

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================


@dp.table(
    name="dev_catalog.silver.orders_recent",
    comment="Streaming table: orders_recent",
    table_properties={
        "delta.enableChangeDataFeed": "true",
        "pipelines.reset.allowed": "false",
    },
    replace_where="order_date >= date_add(current_date(), -7)",
)
def orders_recent():
    """Replace-where flow to dev_catalog.silver.orders_recent"""
    df = spark.read.table("v_orders")

    return df

The target is one @dp.table statement, which Databricks reads as both the streaming table and its REPLACE WHERE flow. There is no dp.create_streaming_table and no flow decorator. The predicate is the replace_where= argument, exactly as you wrote it, and the flow reads its view with spark.read. table_properties holds your change-data-feed setting and one more that you did not write, pipelines.reset.allowed.

Keep the history safe

A full refresh of a REPLACE WHERE flow re-runs the source query for the current predicate only. Without protection it would delete every row older than 7 days. Lakehouse Plumber therefore sets pipelines.reset.allowed to "false" on every REPLACE WHERE table, unless you set it yourself:

• A full refresh of the whole pipeline still runs. The table keeps its older rows, and only the last 7 days are recomputed.

• A full refresh that selects this table fails the update, because the table is not resettable.

When you do want to start the table over, see the reset recipe in the write reference.

Keep refreshes incremental

Databricks can refresh the flow incrementally, processing only what changed inside the predicate instead of recomputing all of it. That needs serverless compute, a deterministic predicate on columns of the source table, and a deterministic query. Two things in your YAML affect it:

• The predicate. A moving lower bound, like the one above, stays eligible. Widening the predicate costs one full recomputation of the window, and so can a moving upper bound.

• Operational metadata. A metadata column computed with a non-deterministic function, such as F.current_timestamp(), makes every refresh recompute the whole window. lhp validate warns about it with LHP-VAL-073.

The write reference explains how to check in the event log which way each refresh ran.

Read the table downstream

Every refresh rewrites rows inside the predicate, so the table is not an append-only source. Suppose a gold pipeline streams it, as it would any other table:

pipelines/orders_recent_copy.yaml (rejected)

pipeline: gold_orders
flowgroup: orders_recent_copy

actions:
  - name: load_orders_recent
    type: load
    readMode: stream
    source:
      type: delta
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders_recent
    target: v_orders_recent

  - name: write_orders_recent_copy
    type: write
    source: v_orders_recent
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${gold_schema}"
      table: orders_recent_copy

On Databricks that stream would fail as soon as a refresh rewrote a row, and with skipChangeCommits it would silently skip the rewritten rows instead. lhp validate stops it first:

$ lhp validate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✗ gold_orders  LHP-VAL-072  Streaming read of a REPLACE WHERE table
✓ silver_orders  0 files
✗ validate (0.32s)
1 validated · 1 failed · 0.3s
gold_orders  LHP-VAL-072  Streaming read of a REPLACE WHERE table
Re-run with details: lhp validate --env dev --show-details

The check covers the whole run, so it catches a reader in another pipeline. In the --show-details panel, the context line names only the pipeline. The message names the read, and the write behind it, as pipeline.flowgroup.action:

gold_orders.orders_recent_copy.load_orders_recent streams from 'dev_catalog.silver.orders_recent', a REPLACE WHERE table written by silver_orders.orders_recent.write_orders_recent. Each refresh rewrites rows, which a streaming read cannot follow (skipChangeCommits silently drops data). Use readMode: batch, or stream its change feed with options.readChangeFeed: "true" (see docs).

A batch read fixes it. To keep a streaming copy instead, follow the table’s change data feed, which records each rewrite as delete and insert events, and apply those events with AUTO CDC. Replace the file with pipelines/orders_recent_mirror.yaml:

pipelines/orders_recent_mirror.yaml

# Mirror the REPLACE WHERE table silver.orders_recent into gold, through its change data feed.
# Action chain: load (delta, change-feed stream of the REPLACE WHERE table) -> write (streaming_table, mode: cdc).
#
# A plain streaming read of a REPLACE WHERE table fails validation (LHP-VAL-072):
# every refresh rewrites rows, which a stream cannot follow. The change feed
# carries each rewrite as delete and insert events, which AUTO CDC applies.
pipeline: gold_orders
flowgroup: orders_recent_mirror

actions:
  # 1. Load: stream the change feed of the REPLACE WHERE table. The where_clause
  #    drops update pre-images, so a stale pre-image can never win the tie-break.
  - name: load_orders_recent_changes
    type: load
    readMode: stream
    source:
      type: delta
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders_recent
      options:
        readChangeFeed: "true"
      where_clause:
        - "_change_type != 'update_preimage'"
    target: v_orders_recent_changes

  # 2. Write: apply the change events onto order_id. Within one commit a key can
  #    have both a delete and an insert; sorting by _change_type after
  #    _commit_version makes the insert win.
  - name: write_orders_recent_mirror
    type: write
    source: v_orders_recent_changes
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${gold_schema}"
      table: orders_recent_mirror
      mode: cdc
      cdc_config:
        keys: ["order_id"]
        sequence_by: ["_commit_version", "_change_type"]
        scd_type: 1
        apply_as_deletes: "_change_type = 'delete'"
        except_column_list: ["_change_type", "_commit_version", "_commit_timestamp"]

• readChangeFeed: "true" streams the table’s change events instead of its rows. Lakehouse Plumber does not report a change-feed read as LHP-VAL-072.

• sequence_by: ["_commit_version", "_change_type"] is required. A rewritten row has a delete and an insert in the same commit. Ordered by _commit_version alone, AUTO CDC can keep the delete and drop the row from the copy without any error; adding _change_type makes the insert win.

• The where_clause drops update pre-images, and except_column_list keeps the change-feed columns out of the copy.

Validate and generate again; both pipelines pass:

$ lhp validate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ gold_orders  0 files
✓ silver_orders  0 files
✓ validate (0.32s)
2 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ gold_orders  1 file
✓ silver_orders  1 file
✓ generate (0.34s)
✓ format (0.01s)
✓ monitoring (0.00s)
2 pipelines generated · 2 files · 0.3s

The mirror’s flow is an AUTO CDC flow over the change feed:

generated/dev/gold_orders/orders_recent_mirror.py (excerpt)

# CDC flow: f_orders_recent_mirror
dp.create_auto_cdc_flow(
    target="dev_catalog.gold.orders_recent_mirror",
    source="v_orders_recent_changes",
    name="f_orders_recent_mirror",
    keys=["order_id"],
    sequence_by=struct("_commit_version", "_change_type"),
    stored_as_scd_type=1,
    except_column_list=["_change_type", "_commit_version", "_commit_timestamp"],
    apply_as_deletes="_change_type = 'delete'",
)

What you just did

A short replace_where_config block compiled to one @dp.table with a replace_where= predicate, a batch read and a reset guard. You declared which rows to keep recomputing; Databricks deletes and recomputes them on each refresh, and Lakehouse Plumber checked the setup around it: a batch source, a single flow, and downstream pipelines that follow the table through its change feed instead of streaming its rows.

What’s next

• Replace by key from a stream instead. When the source resends complete sets of rows per key, use mode: replace (REPLACE USING), covered in Replace all rows that match a grouping key.

• See the field reference. The rules, the reset recipe, the incremental refresh conditions, LHP-VAL-072 and LHP-VAL-073 and backfills are described in the write action reference.

Syntax and options

See Replace rows matching a predicate for all supported settings and defaults.