I'm attempting to run a DLT pipeline that uses automatic schema evolution against tables that have type widening enabled.

I have code in this notebook that is a list of tables to create/update along with the schema for those tables. This list and spark schema are fed into this load_snapshot_tables function. That load_snapshot_tables function looks like this:

def load_snapshot_tables(source_system_name, source_schema_name, table_name, spark_schema, select_expression):

    @Dlt.table (
        name=table_name,
        comment=f"{source_system_name}_{source_schema_name}.{table_name}_Snapshot",
        table_properties={"delta.enableTypeWidening": "true"},
        cluster_by=["XXX_Snapshot_Date"]
    )
    def create_snapshot_table():

        snapshot_load_df = (
            spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", False)
            .option("cloudFiles.includeExistingFiles", True)
            .option("pathGlobFilter", "*.json.gz")
            .schema(spark_schema)
            .load(f"abfss://YYY@{adl_name}.dfs.core.windows.net/Snapshot/{source_system_name}/{table_name}")
            .selectExpr(
                "CAST(concat(substring(_metadata.file_name, -20,4),'-',substring(_metadata.file_name, -16,2),'-',substring(_metadata.file_name, -14,2)) AS timestamp) AS XXX_Snapshot_Date",
                *select_expression,
                "_metadata.file_name AS XXX_File_Name",
                "_metadata AS XXX_File_Metadata"
            )
        )

        return (snapshot_load_df)

Everything works except type widening. New columns are added based on the schema I pass in. However, when changing data types, the process fails indicating a casting/type issue. Refreshing the tables resolves the errors. But, I don't want to have to refresh the tables. I've referenced URL Type-Widening in my work/research. In this URL, there is a section titled Widening Types with Automatic Schema Evolution. I meet all of the requirements listed there with possibly the only exception being the first bullet (The command uses INSERT or MERGE INTO). I would have assumed behind the scenes INSERT or MERGE INTO is somehow being used here.

I am using the Preview channel for the pipeline.

So, two questions:

1. What am I missing in my python code to make sure type widening is being honored?
2. What would my python code look like if I had to convert it to force it to use INSERT or MERGE INTO?

Thanks in advance!