The Art of PostgreSQL/

...

/

Putting Concurrency to the Test

Putting Concurrency to the Test

Perform a few tests to learn more about how concurrency works in SQL.

We'll cover the following...

When we benchmark the concurrency properties of the two statements, we quickly realize that the activity table is badly designed. The unique constraint includes a timestamptz field, which in PostgreSQL is only precise down to the microsecond.

Fixing the table

This kind of made-up unique constraint means we now have these errors to deal with:

Error: Database error 23505: duplicate key value violates unique constraint "activity_messageid_datetime_action_key"
DETAIL: Key (messageid, datetime, action)=(1, 2022-06-01 10:17:39.079724+00, rt) already exists.

The best course of action here is to do this:

Press + to interact
alter table tweet.activity
drop constraint activity_messageid_datetime_action_key;

Comparing the concurrency scaling

Now, we can properly compare the concurrency scaling of the insert and the update-based versions. In case you might be curious about it, here’s the testing code that’s been used:

Press + to interact
(defpackage #:concurrency
  (:use #:cl #:appdev)
  (:import-from #:lparallel
                #:*kernel*
                #:make-kernel #:make-channel
                #:submit-task #:receive-result
                #:kernel-worker-index)
  (:import-from #:cl-postgres-error
                #:database-error)
  (:export      #:*connspec*
                #:concurrency-test))
(in-package #:concurrency)
(defparameter *connspec* '("appdev" "dim" nil "localhost"))
(defparameter *insert-rt*
  "insert into tweet.activity(messageid, action) values($1, 'rt')")
(defparameter *update-rt*
  "update tweet.message set rts = coalesce(rts, 0) + 1 where messageid = $1")
(defun concurrency-test (workers retweets messageid
                         &optional (connspec *connspec*))
  (format t "Starting benchmark for updates~%")
  (with-timing (rts seconds)
      (run-workers workers retweets messageid *update-rt* connspec)
    (format t "Updating took ~f seconds, did ~d rts~%" seconds rts))
  (format t "~%")
  (format t "Starting benchmark for inserts~%")
  (with-timing (rts seconds)
      (run-workers workers retweets messageid *insert-rt* connspec)
    (format t "Inserting took ~f seconds, did ~d rts~%" seconds rts)))
(defun run-workers (workers retweets messageid sql
                    &optional (connspec *connspec*))
  (let* ((*kernel* (lparallel:make-kernel workers))
         (channel  (lparallel:make-channel)))
    (loop repeat workers
       do (lparallel:submit-task channel #'retweet-many-times
                                 retweets messageid sql connspec))
    (loop repeat workers sum (lparallel:receive-result channel))))
(defun retweet-many-times (times messageid sql
                           &optional (connspec *connspec*))
  (pomo:with-connection connspec
    (pomo:query
     (format nil "set application_name to 'worker ~a'"
             (lparallel:kernel-worker-index)))
    (loop repeat times sum (retweet messageid sql))))
(defun retweet (messageid sql)
  (handler-case
      (progn
        (pomo:query sql messageid)
        1)
    (database-error (c)
      (format t "Error: ~a~%" c)
      0)))

Performing Test

For the first test, it’s a ...