Validating a metadata-driven Bronze → Silver incremental load pattern in Fabric

Hi all,

I’m building a metadata-driven Bronze → Silver ingestion framework in Microsoft Fabric and would like feedback on the approach before scaling it to a few hundred tables.

Setup

• Source: Dynamics 365 F&O tables land in a Bronze lakehouse as Delta tables through Fabric Link. Each row has a SinkModifiedOn timestamp and an IsDelete soft-delete flag.
• Config DB: a Fabric SQL Database stores per-table metadata (table name, schema, key column, load type, column list) and the last processed watermark.
• Orchestration: a Data Pipeline runs a Lookup on the config DB, then a ForEach that calls one generic parameterized notebook per table. The notebook returns the new watermark through notebookutils.notebook.exit(). A Script/Stored Procedure activity writes it back to the config DB only if the notebook succeeds.
• Target: a Silver lakehouse with one Delta table per source table.

Parameter cell (values passed from the pipeline)

python

table_name = "generaljournalentry" schema = "d365" key_col = "recid" load_type = "incremental" # full / incremental select_cols = "recid,accountingdate,...,SinkModifiedOn,IsDelete" # "" = all columns last_wm = "" # from config DB; "" = first run

Load logic

python

table_name = "generaljournalentry"

schema = "d365"

key_col = "recid"

load_type = "incremental" # full / incremental

select_cols = "recid,accountingdate,...,SinkModifiedOn,IsDelete" # "" = all columns

last_wm = "" # from config DB; "" = first run

df = spark.read.format("delta").load(src_path)

if select_cols:

df = df.select(*[c.strip() for c in select_cols.split(",")])

new_wm = None

# ---------- FULL ----------

if load_type == "full":

w = Window.partitionBy(key_col).orderBy(F.col("SinkModifiedOn").desc())

df = (df.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")

.withColumn("silver_loaded_on", F.current_timestamp())).cache()

new_wm = df.agg(F.max("SinkModifiedOn")).first()[0]

(df.filter("coalesce(IsDelete, false) = false")

.write.format("delta").mode("overwrite").option("overwriteSchema", "true").save(tgt_path))

# ---------- INCREMENTAL ----------

else:

target_exists = DeltaTable.isDeltaTable(spark, tgt_path)

# Watermark from config DB, with a 1-hour safety overlap

if last_wm:

df = df.filter(F.col("SinkModifiedOn") > F.to_timestamp(F.lit(last_wm)) - F.expr("INTERVAL 1 HOURS"))

# Keep the latest version per key

w = Window.partitionBy(key_col).orderBy(F.col("SinkModifiedOn").desc())

df = (df.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")

.withColumn("silver_loaded_on", F.current_timestamp())).cache()

# New watermark = max of the batch (deleted rows included)

new_wm = df.agg(F.max("SinkModifiedOn")).first()[0]

if new_wm:

if not target_exists:

df.filter("coalesce(IsDelete, false) = false").write.format("delta").save(tgt_path)

else:

(DeltaTable.forPath(spark, tgt_path).alias("t")

.merge(df.alias("s"), f"t.{key_col} = s.{key_col}")

.whenMatchedDelete("coalesce(s.IsDelete, false) = true")

.whenMatchedUpdateAll("s.SinkModifiedOn > t.SinkModifiedOn")

.whenNotMatchedInsertAll("coalesce(s.IsDelete, false) = false")

.execute())

# Return the new watermark to the pipeline

notebookutils.notebook.exit(json.dumps({

"table": f"{schema}.{table_name}",

"new_watermark": str(new_wm or last_wm)

}))

Questions

1. Watermark handling. The watermark is passed in from the config DB and written back from the notebook’s exit value only after a successful run, so a failed run never advances it. Is a 1-hour overlap a sensible safety margin for SinkModifiedOn? Is there a cleaner way to keep the watermark update and the Delta MERGE consistent if the pipeline fails between the two steps?
2. SinkModifiedOn reliability. For D365 data landed through Fabric Link, is SinkModifiedOn reliable as a monotonic incremental column? Are there known cases of records arriving with an older timestamp?
3. Deduplication. I keep the latest version per key with row_number() ordered by SinkModifiedOn. Should I add a tiebreaker column, such as versionnumber, to handle identical timestamps?
4. Orchestration at scale. For around 200+ tables, is ForEach → Notebook activity per table reasonable? Or are notebookutils.notebook.runMultiple or high-concurrency sessions the better choice to avoid repeated Spark startup overhead?
5. Soft deletes. Is physically deleting rows in Silver on IsDelete = true the common approach, or do people usually keep soft-deleted rows in Silver and filter them in Gold?
6. Table maintenance. Any recommended settings for frequently merged Silver tables, such as OPTIMIZE or V-Order schedule, deletion vectors, or partitioning?

Any feedback, gotchas or alternative patterns would be appreciated. Thanks!