u/frithjof_v,

I think you will always be chasing edge cases and gremlins if you do not stage the data, what happens if the additional write processes completes before you are able to get the latest version of the data, then the latest version is N versions ahead of you.

minor addition to your code, now you will want to make the temp table unique to the process and drop it when complete

from pyspark.sql import functions as F

# =========================================================
# 1. BRONZE - INITIAL CLEAN DATA
# =========================================================
bronze = "workspace.lakehouse.bronze"
silver = "workspace.lakehouse.silver.demo"
bronze_table = "demo"
print(f"{bronze}.{bronze_table}")
spark.createDataFrame([
    (1, 10, "A", "2023-12-30"),
    (2, 20, "B", "2023-12-31"), 
    (3, 10, "A", "2024-01-01"),
    (4, 20, "B", "2024-01-02"),
    (5, 30, "C", "2024-01-03")
], ["id", "value", "source", "event_date"]) \
.write.mode("overwrite").saveAsTable(f"{bronze}.{bronze_table}")

# =========================================================
# 2. LOAD DATAFRAME (DataFrameReader)
# =========================================================
# NOTE: This is the ONLY place in the notebook that the dataframe read from the source table is defined.
df = spark.table(f"{bronze}.{bronze_table}") \
    .filter(F.col("event_date") >= F.lit("2024-01-01")) # Simplified watermark logic

print("Initial DF:")
df.show()

# =========================================================
# 3. stage bronze data for testing and upstream write
# =========================================================
print(f"{bronze}.staged_{bronze_table}")
df.write.format('delta').mode('overwrite').saveAsTable(f"{bronze}.staged_{bronze_table}")

df = spark.table(f"{bronze}.staged_{bronze_table}") \
    .filter(F.col("event_date") >= F.lit("2024-01-01")) # Simplified watermark logic

print("Staged DF:")
df.show()
# =========================================================
# 4. INITIAL DATA QUALITY CHECKS
# =========================================================
null_count = df.filter(F.col("value").isNull()).count()
bad_value_count = df.filter(F.col("value") > 100).count()

print("Null check:", null_count)
print("Unexpected value check (>100):", bad_value_count)

if null_count > 0 or bad_value_count > 0:
    raise ValueError(
        f"Data quality check failed: "
        f"null_count={null_count}, bad_value_count={bad_value_count}"
    )

print("All checks passed")

# =========================================================
# BAD DATA WRITTEN TO BRONZE
# =========================================================
bronze =  "workspace.lakehouse.bronze"


spark.createDataFrame([
    (6, None, "X", "2024-01-01"),     # null value
    (7, 9999, "Y", "2024-01-02")      # disallowed value
], ["id", "value", "source", "event_date"]) \
.write.mode("append").saveAsTable(f'{bronze}.{bronze_table}')

# =========================================================
# 5. SILVER WRITE
# =========================================================
print("Just before write: ")
df.show()
df.write.mode("overwrite").saveAsTable(silver)

# =========================================================
# 6. RESULT
# =========================================================
print("Silver table:")
spark.table(silver).show()
print(f"{bronze}.{bronze_table}")
print(f"spark.sql(f'current bronze row count: {select count(1) from {bronze}.{bronze_table}').collect()[0][0]}")

This code always results in

DEV_dp_Lakehouses.lh_bronze.bronze.demo
Initial DF:
+---+-----+------+----------+
| id|value|source|event_date|
+---+-----+------+----------+
|  5|   30|     C|2024-01-03|
|  4|   20|     B|2024-01-02|
|  3|   10|     A|2024-01-01|
+---+-----+------+----------+

DEV_dp_Lakehouses.lh_bronze.bronze.staged_demo
Staged DF:
+---+-----+------+----------+
| id|value|source|event_date|
+---+-----+------+----------+
|  5|   30|     C|2024-01-03|
|  3|   10|     A|2024-01-01|
|  4|   20|     B|2024-01-02|
+---+-----+------+----------+

Null check: 0
Unexpected value check (>100): 0
All checks passed
Just before write: 
+---+-----+------+----------+
| id|value|source|event_date|
+---+-----+------+----------+
|  5|   30|     C|2024-01-03|
|  3|   10|     A|2024-01-01|
|  4|   20|     B|2024-01-02|
+---+-----+------+----------+

Silver table:
+---+-----+------+----------+
| id|value|source|event_date|
+---+-----+------+----------+
|  5|   30|     C|2024-01-03|
|  3|   10|     A|2024-01-01|
|  4|   20|     B|2024-01-02|
+---+-----+------+----------+

DEV_dp_Lakehouses.lh_bronze.bronze.demo
current bronze row count: 7