acquire.sql

UPDATE queued_jobs
SET status='in-progress',
	updated_at=now()
FROM (
	SELECT id AS inner_id
	FROM queued_jobs
	WHERE status='queued'
		AND name = $1
		AND run_after <= now()
	LIMIT 1
	FOR UPDATE
) find_job
WHERE queued_jobs.id = find_job.inner_id 
	AND status='queued'
RETURNING id, status, data

app_code.go

	rows, err := acquireStmt.Query(name)
	if err != nil {
		err = dberror.GetError(err)
		return nil, err
	}
	defer rows.Close()
	count := 0
	scanned := false
	for rows.Next() {
		count += 1
		if !scanned {
			rows.Scan(args(qj, &bt)...)
			scanned = true
		}
	}
	if count == 0 {
		return nil, sql.ErrNoRows
	}
	if count > 1 {
		panic(fmt.Sprintf("Too many rows affected by Acquire: %d", count))
	}
	err = rows.Err()
	if err != nil {
		return nil, err
	}

expected-explain.txt

2016-04-22 22:30:33.173 UTC [30516-571aa5d0.7734-70286] 00000 shyp_api@shyp_jobs LOG:  duration: 3.734 ms  plan:
        Query Text: -- queued_jobs.Acquire
        UPDATE queued_jobs
        SET status='in-progress',
                updated_at=now()
        FROM (
                SELECT id AS inner_id
                FROM queued_jobs
                WHERE status='queued'
                        AND name = $1
                        AND run_after <= now()
                LIMIT 1
                FOR UPDATE
        ) find_job
        WHERE queued_jobs.id = find_job.inner_id
                AND status='queued'
        RETURNING id,
                name,
                attempts,
                run_after,
                expires_at,
                status,
                data,
                created_at,
                updated_at
        Update on queued_jobs  (cost=0.41..8.53 rows=1 width=120) (actual time=3.730..3.733 rows=1 loops=1)
          ->  Nested Loop  (cost=0.41..8.53 rows=1 width=120) (actual time=3.688..3.690 rows=1 loops=1)
                ->  Subquery Scan on find_job  (cost=0.00..0.08 rows=1 width=56) (actual time=3.672..3.673 rows=1 loops=1)
                      ->  Limit  (cost=0.00..0.07 rows=1 width=22) (actual time=3.662..3.662 rows=1 loops=1)
                            ->  LockRows  (cost=0.00..2935.47 rows=42743 width=22) (actual time=3.661..3.661 rows=1 loops=1)
                                  ->  Seq Scan on queued_jobs queued_jobs_1  (cost=0.00..2508.04 rows=42743 width=22) (actual time=1.362..1.375 rows=5 loops=1)
                                        Filter: ((status = 'queued'::job_status) AND (name = $1) AND (run_after <= now()))
                                        Rows Removed by Filter: 1
                ->  Index Scan using queued_jobs_pkey on queued_jobs  (cost=0.41..8.44 rows=1 width=80) (actual time=0.012..0.013 rows=1 loops=1)
                      Index Cond: (id = find_job.inner_id)
                      Filter: (status = 'queued'::job_status)

funky-explain.txt

2016-04-22 22:31:20.648 UTC [30521-571aa5d0.7739-208291] 00000 shyp_api@shyp_jobs LOG:  duration: 67.522 ms  plan:
        Query Text: -- queued_jobs.Acquire
        UPDATE queued_jobs
        SET status='in-progress',
                updated_at=now()
        FROM (
                SELECT id AS inner_id
                FROM queued_jobs
                WHERE status='queued'
                        AND name = $1
                        AND run_after <= now()
                LIMIT 1
                FOR UPDATE
        ) find_job
        WHERE queued_jobs.id = find_job.inner_id
                AND status='queued'
        RETURNING id,
                name,
                attempts,
                run_after,
                expires_at,
                status,
                data,
                created_at,
                updated_at
        Update on queued_jobs  (cost=0.75..16.83 rows=1 width=120) (actual time=3.011..67.515 rows=2 loops=1)
          ->  Nested Loop  (cost=0.75..16.83 rows=1 width=120) (actual time=2.974..67.458 rows=2 loops=1)
                Join Filter: (queued_jobs.id = find_job.inner_id)
                Rows Removed by Join Filter: 475
                ->  Index Scan using queued_jobs_pkey on queued_jobs  (cost=0.38..8.39 rows=1 width=80) (actual time=0.011..1.326 rows=477 loops=1)
                      Filter: (status = 'queued'::job_status)
                      Rows Removed by Filter: 1
                ->  Subquery Scan on find_job  (cost=0.38..8.42 rows=1 width=56) (actual time=0.137..0.138 rows=1 loops=477)
                      ->  Limit  (cost=0.38..8.41 rows=1 width=22) (actual time=0.136..0.136 rows=1 loops=477)
                            ->  LockRows  (cost=0.38..8.41 rows=1 width=22) (actual time=0.136..0.136 rows=1 loops=477)
                                  ->  Index Scan using find_queued_job on queued_jobs queued_jobs_1  (cost=0.38..8.40 rows=1 width=22) (actual time=0.134..0.135 rows=2 loops=477)
                                        Index Cond: ((name = $1) AND (run_after <= now()))
                                        Filter: (status = 'queued'::job_status)

schema.txt

shyp_jobs=# \d queued_jobs;
                   Table "public.queued_jobs"
   Column   |           Type           |       Modifiers
------------+--------------------------+------------------------
 id         | uuid                     | not null
 name       | text                     | not null
 attempts   | smallint                 | not null
 run_after  | timestamp with time zone | not null
 expires_at | timestamp with time zone |
 created_at | timestamp with time zone | not null default now()
 updated_at | timestamp with time zone | not null default now()
 status     | job_status               | not null
 data       | jsonb                    | not null
Indexes:
    "queued_jobs_pkey" PRIMARY KEY, btree (id)
    "find_queued_job" btree (name, run_after) WHERE status = 'queued'::job_status
Check constraints:
    "queued_jobs_attempts_check" CHECK (attempts >= 0)
Foreign-key constraints:
    "queued_jobs_name_fkey" FOREIGN KEY (name) REFERENCES jobs(name)