Forum Discussion
Can parallel pipeline runs upsert into the same Lakehouse table without commit conflicts?
Thanks a lot, this was very helpful. Your prediction was right, and I could reproduce it.
OPTIMIZE without a partition: After I ran OPTIMIZE on the unpartitioned table, it had merged several _db_id values into shared files. Parallel upserts then failed with ConcurrentDeleteDeleteException (not ConcurrentDeleteReadException), and the Copy sink did not retry it on its own.
What I did: The Copy sink refuses a partition column that is also a key column (KeyColumnIntersectsWithPartitionColumn). If I drop _db_id from the key, the sink matches across partitions and silently overwrites other databases' rows (my databases share the same IDs). So I added a second column with the same value (_db_partition, an additional column in the Copy source) and partitioned by that column. The key stays _db_id + primary key.
What I tested:
- 3 databases loaded one after another into the partitioned table: each kept all its rows, one partition and one file per database.
- OPTIMIZE on a partitioned incremental table compacted the files within each partition (34 files down to 5) and mixed no databases.
- 4 databases in parallel re-upserted already loaded windows into those compacted files: no errors, no duplicates, and all row counts matched.
What I'm doing from here:
- OPTIMIZE and VACUUM only as the last step of the orchestrating pipeline, never on a separate schedule. Not tested: OPTIMIZE running at the same time as an ingest of the same database.
- Keeping the Copy retries.
- Your staging pattern stays as the fallback if conflicts show up in production.
Thanks, I found the same note in the Lakehouse connector docs ("Partition columns cannot overlap with key columns"), so partitioning by _db_id is not an option for us.
As suggested, I tested parallel upserts in a non-production setup.
Setup
- One generic ingest pipeline loops over the tables (sequential ForEach). Each iteration runs a Copy activity from DB2 into a Lakehouse table: Upsert with keys _db_id + primary key, unpartitioned tables, 2 retries.
- To simulate 10 databases, a separate test pipeline calls the ingest pipeline through 10 Invoke pipeline activities without dependencies. Each uses a different _db_id and the same DB2 connection.
Test 1: two parallel runs, small tables. The Copy activities overlapped on the same table, and the MERGE commits landed 7 to 21 seconds apart. Both runs succeeded.
Test 2: ten parallel runs, small table (about 35 rows per run). My first attempt was misleading: the notebook activities queued for Spark sessions, so the Copy activities barely overlapped. With a shared session tag, all 10 Copy activities started within about 2 seconds. All runs succeeded, with no errors and no Copy retries. Sink write time was 11 to 18 seconds, compared to about 9 seconds for a single run, with no staircase of growing durations.
Test 3: ten parallel runs, larger full-load table (57,512 rows per run). A high concurrency session holds at most 5 notebooks, so the runs started in two groups of 5. Most databases already had rows in the table, so this was mostly updates.
- All 10 runs succeeded, with no errors and no Copy retries. Each database ended up with exactly 57,512 rows.
- DESCRIBE HISTORY shows 10 consecutive MERGE versions. The 5 commits of one group landed within 1.5 seconds, roughly 0.35 seconds apart, as if committed one after another. Sink write time stayed at 11 to 14 seconds.
- The table had one file per database afterwards, so the databases did not rewrite each other's files. I suspect that is part of why there were no conflicts, but I have not confirmed it.
What this does not show: how the Copy sink handles conflicts internally (operationParameters, readVersion and isBlindAppend are still NULL), or what happens after OPTIMIZE, which could merge several databases into one file. As you said, a passing test is no guarantee under higher load.