Issue with Upsert Table Action in Metadata-Driven Pipeline to Lakehouse Table

Hello parasol54 - 

Here is an example of how you can perform upserts on lakehouse tables within a metadata-driven pipline.

• Run a notebook which contains this code below. 

Create variables which define the paths (parameters for each component are passed to the notebook from the data pipeline):

# build absolute paths to lakehouse storage locations
source__path = f"{source__protocol}://{source__storage_account}@{source__endpoint}/{source__container}/{source__directory}/"
destination__path = f"{destination__protocol}://{destination__storage_account}@{destination__endpoint}/{destination__container}/{destination__directory}/"

Define a function to perform the upsert.

def process_table(source__lakehouse_path, source__table_name, destination__lakehouse_path, destination__table_name, unique_id):
    # load the tables
    source__table = DeltaTable.forPath(source__lakehouse_path+source__table_name)
    destination__table = DeltaTable.forPath(destination__lakehouse_path+destination__table_name)
    # convert the source table to a dataframe
    df = source__table.toDF()
    # define a list of columns names
    column_names = df.columns
    # perform the merge (upsert) operation
    destination__table.alias('destination') \
    .merge(
        df.alias('updates'),
        'destination.{0} = updates.{0}'.format(unique_id)
    )\
    .whenMatchedUpdate(set = {col: f"updates.{col}" for col in column_names}) \
    .whenNotMatchedInsert(values = {col: f"updates.{col}" for col in column_names})\
    .execute()

Invoke the function for each row in the metadata/control/watermark table (json_array)

for i in json_array:
    source__table_name = i["source__table"]
    destination__table_name = i["destination__table"]
    unique_id = i["unique_id"]
    process_table(source__path, source__table_name, destination__path, destination__table_name, unique_id)

Hi parasol54 

In Power BI Dataflow Gen 2, 

Warehouse table = Lakehouse Delta table

Here you can see the Fabric SQL database = Demo Semantic Model = Demo Warehouse = Lakehouse. All four are the same and give you desired results.

so you are getting unified data. 

There is only overwrite mode of delta tables. It means there is no upsert needed. The reason is In Memory engine of Linux Engine aka Windows Engine.

You have to understand the whole concept of apache spark, Delta lakes and how they work.

Hi Bhavesh,

Thanks for the explanation! Just to clarify — is the “Upsert” option also unsupported when the destination is a Warehouse table (not a Lakehouse Delta table)?

I understand that Lakehouse relies on Delta Lake versioning and doesn’t support upsert through the Copy activity — but will this limitation still apply if I use Warehouse as the destination?

Ultimately, I’m trying to implement metadata-driven upserts — would that work better in the Warehouse environment?

Regards