ON CONFLICT DO NOTHING in YugabyteDB

In a previous post, UPSERT in YugabyteDB, I mentioned that the INSERT ... ON CONFLICT statement processes data row by row and may not be scalable. Instead, using standard SQL clauses like WITH, NOT IN(), and RETURNING is advisable to query for existing data and define what needs to be inserted with declarative logic. This post discusses a similar approach for INSERT ... ON CONFLICT DO NOTHING.

Here is an example where I have an events table containing one million values with id from one to one million:

create table events ( id bigint primary key, value float);
insert into events select generate_series(1,1000000), random();

I have 14 new values (id from 1001 to 1014) to insert, but my input also included id from 1 to 4 that already exists. Let's try the ON CONFLICT DO NOTHING:

yugabyte=# explain (analyze, dist, costs off, summary)
insert into events values
 (1001,random()), (1002,random()), (1003,random()), (1004,random()),
 (1007,random()), (1008,random()), (1009,random()), (1010,random()),
 (1011,random()), (1012,random()), (1013,random()), (1014,random()),
 (   1,random()), (   2,random()), (   3,random()), (   4,random())
on conflict do nothing
;
                                 QUERY PLAN
----------------------------------------------------------------------------
 Insert on events (actual time=26.726..26.726 rows=0 loops=1)
   Conflict Resolution: NOTHING
   Tuples Inserted: 12
   Conflicting Tuples: 4
   ->  Values Scan on "*VALUES*" (actual time=0.003..0.072 rows=16 loops=1)
         Storage Table Read Requests: 16
         Storage Table Read Execution Time: 9.096 ms
         Storage Table Write Requests: 12.000
         Storage Flush Requests: 12
         Storage Flush Execution Time: 15.577 ms
 Planning Time: 0.055 ms
 Execution Time: 26.768 ms
 Storage Read Requests: 16
 Storage Read Execution Time: 9.096 ms
 Storage Write Requests: 12.000
 Storage Flush Requests: 12
 Storage Flush Execution Time: 15.577 ms
 Storage Execution Time: 24.673 ms
 Peak Memory Usage: 8 kB

I have a read request for each of the 16 rows in the input (Storage Table Read Requests: 16) and a write request for each of the 12 row that doesn't conflict and is finally inserted (Storage Flush Requests: 12). Those weren't batched as each one has been flushed (Storage Flush Requests: 12). This multiplies with the network latency and the total took 25 milliseconds.

With a Distributed SQL database where the read and write requests are network calls, they should be sent by batches.

I'll write the logic with Common Table Expressions (CTE):

the input values are built with VALUES if literals are passed or SELECT if they exist in another table. I call this values_to_insert
the list of keys that are ON CONFLICT as they already exists in the target table. I use a simple IN(values_to_insert) to get them and call it conflict
the insert removes those conflicting key from the input value. I use a simple NOT IN(conflict)
I enabled Batched Nested Loop Join to get arrays of values in this IN() clause, setting yb_bnl_batch_size to 1024
I also count the inserted keys and the conflicted ones

Here is the query:

yugabyte=# explain (analyze, dist, costs off, summary)
/*+ Set (yb_bnl_batch_size 1024) */
with
-- the values to insert
values_to_insert(id,values) as ( values
 (1001,random()), (1002,random()), (1003,random()), (1004,random()),
 (1007,random()), (1008,random()), (1009,random()), (1010,random()),
 (1011,random()), (1012,random()), (1013,random()), (1014,random()),
 (   1,random()), (   2,random()), (   3,random()), (   4,random())
),
-- the keys that alredy exists
conflict as (
 select id from events
  where id in (select id from values_to_insert)
),
-- the insert of new key only
inserted as (
 insert into events
 select * from values_to_insert
  where id not in ( select id from conflict where id is not null)
  returning id
)
-- feedback
select ( select count(*) inserted  from inserted)
     , ( select count(*) conflicts from conflict)
;

                                                           QUERY PLAN
---------------------------------------------------------------------------------------------------------------------------------
 Result (actual time=1.349..1.350 rows=1 loops=1)
   CTE values_to_insert
     ->  Values Scan on "*VALUES*" (actual time=0.003..0.009 rows=16 loops=1)
   CTE conflict
     ->  YB Batched Nested Loop Join (actual time=1.189..1.198 rows=4 loops=1)
           Join Filter: (events.id = values_to_insert.id)
           ->  HashAggregate (actual time=0.021..0.024 rows=16 loops=1)
                 Group Key: values_to_insert.id
                 ->  CTE Scan on values_to_insert (actual time=0.000..0.012 rows=16 loops=1)
           ->  Index Scan using events_pkey on events (actual time=1.079..1.083 rows=4 loops=1)
                 Index Cond: (id = ANY (ARRAY[(values_to_insert.id)::bigint, ($2)::bigint, ($3)::bigint, ..., ($1024)::bigint]))
                 Storage Table Read Requests: 1
                 Storage Table Read Execution Time: 0.911 ms
   CTE inserted
     ->  Insert on events events_1 (actual time=1.258..1.329 rows=12 loops=1)
           Storage Table Write Requests: 12.000
           ->  CTE Scan on values_to_insert values_to_insert_1 (actual time=1.218..1.226 rows=12 loops=1)
                 Filter: (NOT (hashed SubPlan 3))
                 Rows Removed by Filter: 4
                 SubPlan 3
                   ->  CTE Scan on conflict (actual time=1.191..1.202 rows=4 loops=1)
                         Filter: (id IS NOT NULL)
   InitPlan 5 (returns $1029)
     ->  Aggregate (actual time=1.342..1.343 rows=1 loops=1)
           ->  CTE Scan on inserted (actual time=1.260..1.336 rows=12 loops=1)
   InitPlan 6 (returns $1030)
     ->  Aggregate (actual time=0.003..0.003 rows=1 loops=1)
           ->  CTE Scan on conflict conflict_1 (actual time=0.000..0.000 rows=4 loops=1)
 Planning Time: 1.494 ms
 Execution Time: 5.862 ms
 Storage Read Requests: 1
 Storage Read Execution Time: 0.911 ms
 Storage Write Requests: 12.000
 Storage Flush Requests: 1
 Storage Flush Execution Time: 3.561 ms
 Storage Execution Time: 4.472 ms
 Peak Memory Usage: 4760 kB

The performance is now scalable. To check for conflicts I have only one read request because they are batched: one read for up to 1024 rows in the input. I have twelve write requests, one per row to be inserted, but they are sent by batches, here only one flush request.

If you have duplicates in the input and want to pick only the first one like 'ON CONFLICT DO NOTHING' does, you can replace select * from values_to_insert by select distinct on (id) * from values_to_insert.

This is efficient when inserting few rows into a large table. For the opposite, inserting a lot of rows on a nearly empty table, you may prefer to omit the where id in (select id from values_to_insert) so that you fetch all existing keys rather than sending the new keys to get only the conflicting ones.

Blog

ON CONFLICT DO NOTHING in YugabyteDB

Franck Pachot

Join Our Newsletter. No Spam, Only the good stuff.

Related