Hello ksukhani & ssrithar 

As you can imagine, I can't provide even a simplified version of the codes, but I did my best below in trying to show you how the structure looks like. I think this should help well enough in the context of this post.

Bronze tables

These are created in similar fashion, relying on PySpark for parametrization, dynamic logic for file ingestion, ETL and clean-up, and finally the writing of the tables which is done as below.

df.write.mode("overwrite").format("delta").option("delta.enableChangeDataFeed", "true").partitionBy("Key").saveAsTable(table_name)

The overwrite is there because the input files have timestamps, so if say you already have data from 01.01.2026, when you upload the file for 01.02.2026 you don't need the previous data anymore. Thus, you need to overwrite the existing data in the lakehouse.

Silver MLV

Below is the structure/skeleton for the MLV, as mentioned at the beginning.

sql_statement = f"""

CREATE OR REPLACE MATERIALIZED LAKE VIEW mlv_nane
TBLPROPERTIES (delta.enableChangeDataFeed=true)

AS

SELECT 
    columns 
FROM (
    SELECT columns FROM table WHERE conditions
    UNION ALL
    SELECT columns FROM table WHERE conditions
    UNION ALL
    SELECT columns FROM table WHERE conditions
    UNION ALL
    SELECT columns FROM table WHERE conditions
) AS base_table

LEFT JOIN (SELECT DISTINCT columns FROM table) AS join_1 
    ON base_table.id = join_1.id

LEFT JOIN (SELECT DISTINCT columns FROM table) AS join_2 
    ON base_table.id = join_2.id

LEFT JOIN (SELECT DISTINCT columns FROM table) AS join_3 
    ON base_table.id = join_3.id

LEFT JOIN (SELECT DISTINCT columns FROM table) AS join_4 
    ON base_table.id = join_4.id

LEFT JOIN (SELECT DISTINCT columns FROM table) AS join_5 
    ON base_table.id = join_5.id

"""

spark.sql(sql_statement)

Testing

In the notebook I created for testing, the logic is something like this:

from pyspark.sql import Row

mock_row = Row( columns and values )

mock_df = spark.createDataFrame([mock_row])

mock_df.write.format("delta").option("delta.enableChangeDataFeed", "true").mode("append").saveAsTable(table_name)

Note: table_name from Testing is the same table_name from the Bronze layer.