whenNotMatchedBySourceUpdate ConcurrentAppendException Partition

marcuskw — Fri, 08 Sep 2023 14:14:10 GMT

ConcurrentAppendException requires a good partitioning strategy, here my logic works without fault for

"whenMatchedUpdate" and "whenNotMatchedInsert" logic.

When using "whenNotMatchedBySourceUpdate" however it seems that the condition doesn't isolate the specific partition in the delta table.

So when merging a dataframe with that logic we meet a ConcurrentAppendException when running in parallel even though the table is set up with "partition" as a constraint.

(

deltaTable.alias('t')

.merge(df.alias('c'), f" t.partition= '{partition}' AND t.id= c.id")

.whenMatchedUpdate( set =

{

"t.id": "c.id"

,"t.name": "c.name"

,"t.partition": "c.partition"

,"t.flag": "c.flag"

}

)

.whenNotMatchedInsert( values =

{

"t.id": "c.id"

,"t.name": "c.name"

,"t.partition": "c.partition"

,"t.flag": "c.flag"

}

)

.whenNotMatchedBySourceUpdate( condition = f"t.partition= '{partition}'",

set ={

,"t.flag": F.lit(True)

}

)

.execute()

)

Have I misunderstood the merge syntax, possibly whenNotMatchedBySourceUpdate scans the whole table and ignores the condition?

Re: whenNotMatchedBySourceUpdate ConcurrentAppendException Partition

marcuskw — Mon, 11 Sep 2023 08:09:46 GMT

https://docs.databricks.com/en/delta/merge.html

"By definition, whenNotMatchedBySource clauses do not have a source row to pull column values from, and so source columns can’t be referenced."

I found out that this was not functioning correctly in runtime 12.2 but works in 13.3.
So it must have been a fix in the Delta Lake version

topic Re: whenNotMatchedBySourceUpdate ConcurrentAppendException Partition in Data Engineering

whenNotMatchedBySourceUpdate ConcurrentAppendException Partition

Re: whenNotMatchedBySourceUpdate ConcurrentAppendException Partition