Implement generic Table for basic CRUD operations & concurrent safe dequeuing

[VojtechVitek]

Concurrent dequeue/update pattern

• Using a shared lock using PostgreSQL's FOR UPDATE SKIP LOCKED pattern
• Concurrent dequeue/process example:

  pgkit/tests/table_test.go

  Lines 193 to 200 in d081629

  	reviews, err := db.Reviews.DequeueForProcessing(ctx, 10)
  	require.NoError(t, err, "dequeue reviews")
  	for i, review := range reviews {
  	go worker.ProcessReview(ctx, review)
  	ids[i] = append(ids[i], review.ID)
  	}

• Dequeue code:

  pgkit/tests/tables_test.go

  Lines 24 to 47 in d081629

  	func (t *reviewsTable) DequeueForProcessing(ctx context.Context, limit uint64) ([]*Review, error) {
  	var dequeued []*Review
  	where := sq.Eq{
  	"status": ReviewStatusPending,
  	"deleted_at": nil,
  	}
  	orderBy := []string{
  	"created_at ASC",
  	}
  	err := t.LockForUpdates(ctx, where, orderBy, limit, func(reviews []*Review) {
  	now := time.Now().UTC()
  	for _, review := range reviews {
  	review.Status = ReviewStatusProcessing
  	review.ProcessedAt = &now
  	}
  	dequeued = reviews
  	})
  	if err != nil {
  	return nil, fmt.Errorf("lock for updates: %w", err)
  	}
  	return dequeued, nil
  	}

• Worker processing code:

  pgkit/tests/worker_test.go

  Lines 24 to 64 in d081629

  	func (w *Worker) ProcessReview(ctx context.Context, review *Review) (err error) {
  	w.wg.Add(1)
  	defer w.wg.Done()
  	defer func() {
  	// Always update review status to "approved", "rejected" or "failed".
  	noCtx := context.Background()
  	err = w.DB.Reviews.LockForUpdate(noCtx, sq.Eq{"id": review.ID}, []string{"id DESC"}, func(update *Review) {
  	now := time.Now().UTC()
  	update.ProcessedAt = &now
  	if err != nil {
  	update.Status = ReviewStatusFailed
  	return
  	}
  	update.Status = review.Status
  	})
  	if err != nil {
  	log.Printf("failed to save review: %v", err)
  	}
  	}()
  	// Simulate long-running work.
  	select {
  	case <-ctx.Done():
  	return ctx.Err()
  	case <-time.After(1 * time.Second):
  	}
  	// Simulate external API call to an LLM.
  	if rand.Intn(2) == 0 {
  	return fmt.Errorf("failed to process review: <some underlying error>")
  	}
  	review.Status = ReviewStatusApproved
  	if rand.Intn(2) == 0 {
  	review.Status = ReviewStatusRejected
  	}
  	now := time.Now().UTC()
  	review.ProcessedAt = &now
  	return nil
  	}

Data models with strongly typed *data.Record type (using generics):

• CRUD usage:

  pgkit/tests/table_test.go

  Lines 24 to 45 in d081629

  	// Create.
  	err := db.Accounts.Save(ctx, account)
  	require.NoError(t, err, "Create failed")
  	require.NotZero(t, account.ID, "ID should be set")
  	require.NotZero(t, account.CreatedAt, "CreatedAt should be set")
  	require.NotZero(t, account.UpdatedAt, "UpdatedAt should be set")
  	// Check count.
  	count, err := db.Accounts.Count(ctx, nil)
  	require.NoError(t, err, "FindAll failed")
  	require.Equal(t, uint64(1), count, "Expected 1 account")
  	// Read from DB & check for equality.
  	accountCheck, err := db.Accounts.GetByID(ctx, account.ID)
  	require.NoError(t, err, "FindByID failed")
  	require.Equal(t, account.ID, accountCheck.ID, "account ID should match")
  	require.Equal(t, account.Name, accountCheck.Name, "account name should match")
  	// Update.
  	account.Name = "Updated account"
  	err = db.Accounts.Save(ctx, account)
  	require.NoError(t, err, "Save failed")

• DB transaction across multiple tables:

  pgkit/tests/table_test.go

  Lines 63 to 83 in d081629

  	err := db.BeginTx(ctx, func(tx *Database) error {
  	// Create account.
  	account := &Account{Name: "Complex Transaction Account"}
  	err := tx.Accounts.Save(ctx, account)
  	require.NoError(t, err, "Create account failed")
  	articles := []*Article{
  	{Author: "First", AccountID: account.ID},
  	{Author: "Second", AccountID: account.ID},
  	{Author: "Third", AccountID: account.ID},
  	}
  	// Save articles (3x insert).
  	err = tx.Articles.Save(ctx, articles...)
  	require.NoError(t, err, "Save failed")
  	for _, article := range articles {
  	require.NotZero(t, article.ID, "ID should be set")
  	require.NotZero(t, article.CreatedAt, "CreatedAt should be set")
  	require.NotZero(t, article.UpdatedAt, "UpdatedAt should be set")
  	}

[david-littlefarmer]

One comment, looks good, it would be nice to also have paginated methods.

[VojtechVitek]

Exactly, great point about the pagination 👍 I'd love to see both LIMIT/OFFSET and cursor-based pagination defined as a base data model methods. This needs some more thoughts, so I skipped it for now.

[shunkakinoki]

Make where sq.Sqlizer a non-nil clause so that it doesn't check for all conditions?

[VojtechVitek]

Thank you. I think it's a totally valid point to worry about performance of the nil case, which effectively queries all table data:

SELECT COUNT(1) FROM table

Imho, I think we should keep it as is & provide a warning about this in the comment.

And also, I think it'd be great to provide a faster alternative that returns "estimated count" via a query such as

-- Planner’s estimate of live tuples in the table. Updated by VACUUM, ANALYZE, certain DDL.  
SELECT reltuples::BIGINT AS estimated_count 
FROM pg_class 
WHERE relname = 'table_name';

or

-- Estimated live rows tracked by the statistics collector (monitoring view) for each table.  
SELECT n_live_tup
  FROM pg_stat_user_tables
 WHERE relname = 'table_name';

^^ need to test these first

[pkieltyka]

what is the need for us to have a Count() method on the table..? I am nervous for offering this feature to developers, because it means they'll use it, and counting a table is one of the slowest things you can do.. can we just remove this method entirely..?

[VojtechVitek]

Imho there's nothing wrong with SELECT COUNT(1) as long as we hit the index properly, e.g.:

SELECT COUNT(1) FROM jobs WHERE status = 2;

But I hear you, this should never get "abused" for wrong purposes, such as counting number of total pages for LIMIT OFFSET pagination.

[VojtechVitek]

@klaidliadon @david-littlefarmer what do you use COUNT(1) or COUNT(*) for in builder and marketplace-api? I see several use of COUNT in these codebases. Are there any legit use-cases?

[david-littlefarmer]

Yes, we are counting the orders (listings/offers) so we can show the numbers on top of the page on FE. How many listings are there as total count and as filtered count.

[klaidliadon]

In the Builder we use count using foreign keys and/or indexed values.
There are a few exceptions for tables that have a really low row count.