Add MySQL support to River

Here, add a new driver `rivermysql` that brings MySQL support to River.

Similar to SQLite, it's unfortunately not quite as good as the Postgres
driver, but it does the job. MySQL has more facilities than SQLite, but
it's still missing some major niceties like `RETURNING`, which for many
queries requires us to write an implementation using two operations --
one that performs the action, and then another that loads back the
result using returned IDs. It also doesn't have listen/notify.

Luckily, _like_ SQLite, any of this nastiness stays cordoned to the
driver layer and doesn't leak into the mainline River code. We've got
good driver testing and basic tests on clients for each driver in place
so we get reasonable assurance that everything works.

In the before times, I would've been fairly concerned at the additional
maintenance burden that supporting another database would bring, but
with the rise of LLMs I think it's more plausible that we can bring
something like this in without much trouble. Incredibly, I was able to
get almost all of this implemented in just one evening whereby my SQLite
driver took me multiple months.

I don't want to bring MySQL in as a hard dependency, so I've made the
MySQL tests disabled by default. Use `RIVER_MYSQL_TESTS_ENABLED=true` to
activate them. Only one CI matrix case runs MySQL tests so that we don't
have to repeat them to exhaustion on every version of Go and Postgres.

See also:
#158

Author: brandur

.github/workflows/ci.yaml

@@ -91,6 +91,12 @@ jobs:
           # latest Postgres version:
           - go-version: "1.25"
             postgres-version: 18
+
+          # Run with MySQL enabled on the latest Go + Postgres to exercise the
+          # MySQL driver in CI. MySQL tests are opt-in via RIVER_MYSQL_TESTS_ENABLED.
+          - go-version: "1.26"
+            postgres-version: 18
+            mysql-enabled: true
       fail-fast: false
     timeout-minutes: 5
 
@@ -128,7 +134,17 @@ jobs:
       - name: Set up database
         run: psql -c "CREATE DATABASE river_test" $ADMIN_DATABASE_URL
 
+      # MySQL is pre-installed on GitHub-hosted Ubuntu runners. Start the
+      # service and configure passwordless root access for MySQL-enabled runs.
+      - name: Start MySQL
+        if: matrix.mysql-enabled
+        run: |
+          sudo systemctl start mysql
+          mysql -u root -proot -e "ALTER USER 'root'@'localhost' IDENTIFIED BY ''; FLUSH PRIVILEGES;"
+
       - name: Test
+        env:
+          RIVER_MYSQL_TESTS_ENABLED: ${{ matrix.mysql-enabled && 'true' || '' }}
         run: make test/race
 
   cli:

Makefile

@@ -27,6 +27,7 @@ generate/migrations: ## Sync changes of pgxv5 migrations to database/sql
 .PHONY: generate/sqlc
 generate/sqlc: ## Generate sqlc
 	cd riverdriver/riverdatabasesql/internal/dbsqlc && sqlc generate
+	cd riverdriver/rivermysql/internal/dbsqlc && sqlc generate
 	cd riverdriver/riverpgxv5/internal/dbsqlc && sqlc generate
 	cd riverdriver/riversqlite/internal/dbsqlc && sqlc generate
 
@@ -101,5 +102,6 @@ verify/migrations: ## Verify synced migrations
 .PHONY: verify/sqlc
 verify/sqlc: ## Verify generated sqlc
 	cd riverdriver/riverdatabasesql/internal/dbsqlc && sqlc diff
+	cd riverdriver/rivermysql/internal/dbsqlc && sqlc diff
 	cd riverdriver/riverpgxv5/internal/dbsqlc && sqlc diff
 	cd riverdriver/riversqlite/internal/dbsqlc && sqlc diff

client.go

@@ -2385,7 +2385,7 @@ type JobListResult struct {
 	LastCursor *JobListCursor
 }
 
-var errJobListParamsMetadataNotSupportedSQLite = errors.New("JobListParams.Metadata is not supported on SQLite")
+var errJobListParamsMetadataNotSupportedMySQLOrSQLite = errors.New("JobListParams.Metadata is not supported on MySQL or SQLite")
 
 // JobList returns a paginated list of jobs matching the provided filters. The
 // provided context is used for the underlying Postgres query and can be used to
@@ -2406,8 +2406,8 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL
 	}
 	params.schema = c.config.Schema
 
-	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
-		return nil, errJobListParamsMetadataNotSupportedSQLite
+	if (c.driver.DatabaseName() == riverdriver.DatabaseNameMySQL || c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite) && params.metadataCalled {
+		return nil, errJobListParamsMetadataNotSupportedMySQLOrSQLite
 	}
 
 	dbParams, err := params.toDBParams()
@@ -2447,8 +2447,8 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 	}
 	params.schema = c.config.Schema
 
-	if c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite && params.metadataCalled {
-		return nil, errJobListParamsMetadataNotSupportedSQLite
+	if (c.driver.DatabaseName() == riverdriver.DatabaseNameMySQL || c.driver.DatabaseName() == riverdriver.DatabaseNameSQLite) && params.metadataCalled {
+		return nil, errJobListParamsMetadataNotSupportedMySQLOrSQLite
 	}
 
 	dbParams, err := params.toDBParams()

go.work

@@ -9,7 +9,10 @@ use (
 	./riverdriver/riverdatabasesql
 	./riverdriver/riverdrivertest
 	./riverdriver/riverpgxv5
+	./riverdriver/rivermysql
 	./riverdriver/riversqlite
 	./rivershared
 	./rivertype
 )
+
+replace github.com/riverqueue/river/riverdriver/rivermysql v0.35.0 => ./riverdriver/rivermysql

job_list_params.go

@@ -346,10 +346,10 @@ func (p *JobListParams) Kinds(kinds ...string) *JobListParams {
 //
 // https://www.postgresql.org/docs/current/functions-json.html
 //
-// This function isn't supported in SQLite due to SQLite not having an
-// equivalent operator to use, so there's no efficient way to implement it. We
-// recommend the use of Where using a condition with a comparison on the `->>`
-// operator instead.
+// This function isn't supported in MySQL or SQLite due to neither having a
+// direct equivalent to Postgres's `@>` operator. We recommend the use of
+// [JobListParams.Where] with a database-specific JSON comparison instead (e.g.
+// `JSON_EXTRACT` for MySQL, `->>` for SQLite).
 func (p *JobListParams) Metadata(json string) *JobListParams {
 	paramsCopy := p.copy()
 	paramsCopy.metadataCalled = true

riverdbtest/riverdbtest.go

@@ -471,7 +471,8 @@ type TestTxOpts struct {
 	// run using TestSchema. This is meant for environments where parallelism
 	// doesn't work as well, like SQLite, which will emit "busy" errors when
 	// multiple clients try to share a schema, even when they're in separate
-	// transactions.
+	// transactions. Also applies to MySQL, where InnoDB deadlocks are common
+	// when multiple transactions are sharing a database.
 	DisableSchemaSharing bool
 
 	// IsTestTxHelper should be set to true for if TestTx is being called from

riverdriver/river_driver_interface.go

@@ -25,6 +25,7 @@ import (
 const AllQueuesString = "*"
 
 const (
+	DatabaseNameMySQL    = "mysql"
 	DatabaseNamePostgres = "postgres"
 	DatabaseNameSQLite   = "sqlite"
 )
@@ -128,6 +129,13 @@ type Driver[TTx any] interface {
 	// API is not stable. DO NOT USE.
 	PoolSet(dbPool any) error
 
+	// SafeIdentifier returns a safely quoted identifier (e.g. a table or
+	// schema name) for use in SQL queries. Each driver quotes using its
+	// native syntax: double quotes for Postgres/SQLite, backticks for MySQL.
+	//
+	// API is not stable. DO NOT USE.
+	SafeIdentifier(ident string) string
+
 	// SQLFragmentColumnIn generates an SQL fragment to be included as a
 	// predicate in a `WHERE` query for the existence of a set of values in a
 	// column like `id IN (...)`. The actual implementation depends on support

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -53,8 +53,9 @@ func New(dbPool *sql.DB) *Driver {
 
 const argPlaceholder = "$"
 
-func (d *Driver) ArgPlaceholder() string { return argPlaceholder }
-func (d *Driver) DatabaseName() string   { return riverdriver.DatabaseNamePostgres }
+func (d *Driver) ArgPlaceholder() string             { return argPlaceholder }
+func (d *Driver) DatabaseName() string               { return riverdriver.DatabaseNamePostgres }
+func (d *Driver) SafeIdentifier(ident string) string { return dbutil.SafeIdentifier(ident) }
 
 func (d *Driver) GetExecutor() riverdriver.Executor {
 	return &Executor{d.dbPool, templateReplaceWrapper{d.dbPool, &d.replacer}, d}

riverdriver/riverdrivertest/driver_client_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/stdlib"
 	"github.com/lib/pq"
@@ -18,6 +19,7 @@ import (
 	"github.com/riverqueue/river/riverdbtest"
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/riverdriver/riverdatabasesql"
+	"github.com/riverqueue/river/riverdriver/rivermysql"
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	"github.com/riverqueue/river/riverdriver/riversqlite"
 	"github.com/riverqueue/river/rivershared/riversharedtest"
@@ -109,6 +111,26 @@ func TestClientWithDriverRiverLibSQL(t *testing.T) {
 	)
 }
 
+func TestClientWithDriverRiverMySQL(t *testing.T) {
+	t.Parallel()
+
+	riversharedtest.SkipIfMySQLNotEnabled(t)
+
+	var (
+		ctx    = context.Background()
+		dbPool = riversharedtest.DBPoolMySQL(ctx, t)
+		driver = rivermysql.New(dbPool)
+	)
+
+	ExerciseClient(ctx, t,
+		func(ctx context.Context, t *testing.T) (riverdriver.Driver[*sql.Tx], string) {
+			t.Helper()
+
+			return driver, riverdbtest.TestSchema(ctx, t, driver, nil)
+		},
+	)
+}
+
 func TestClientWithDriverRiverSQLiteModernC(t *testing.T) {
 	t.Parallel()
 
@@ -519,9 +541,9 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T,
 		})
 
 		listRes, err := client.JobList(ctx, river.NewJobListParams().Metadata(`{"foo":"bar"}`))
-		if bundle.driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
-			t.Logf("Ignoring unsupported JobListResult.Metadata on SQLite")
-			require.EqualError(t, err, "JobListParams.Metadata is not supported on SQLite")
+		if bundle.driver.DatabaseName() == riverdriver.DatabaseNameMySQL || bundle.driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
+			t.Logf("Ignoring unsupported JobListResult.Metadata on %s", bundle.driver.DatabaseName())
+			require.EqualError(t, err, "JobListParams.Metadata is not supported on MySQL or SQLite")
 			return
 		}
 		require.NoError(t, err)
@@ -583,9 +605,9 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T,
 		})
 
 		listRes, err := client.JobListTx(ctx, tx, river.NewJobListParams().Metadata(`{"foo":"bar"}`))
-		if bundle.driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
-			t.Logf("Ignoring unsupported JobListTxResult.Metadata on SQLite")
-			require.EqualError(t, err, "JobListParams.Metadata is not supported on SQLite")
+		if bundle.driver.DatabaseName() == riverdriver.DatabaseNameMySQL || bundle.driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
+			t.Logf("Ignoring unsupported JobListTxResult.Metadata on %s", bundle.driver.DatabaseName())
+			require.EqualError(t, err, "JobListParams.Metadata is not supported on MySQL or SQLite")
 			return
 		}
 		require.NoError(t, err)
@@ -607,9 +629,12 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T,
 
 		listParams := river.NewJobListParams()
 
-		if bundle.driver.DatabaseName() == riverdriver.DatabaseNameSQLite {
+		switch bundle.driver.DatabaseName() {
+		case riverdriver.DatabaseNameSQLite:
 			listParams = listParams.Where("metadata ->> @json_path = @json_val", river.NamedArgs{"json_path": "$.foo", "json_val": "bar"})
-		} else {
+		case riverdriver.DatabaseNameMySQL:
+			listParams = listParams.Where("JSON_UNQUOTE(JSON_EXTRACT(metadata, @json_path)) = @json_val", river.NamedArgs{"json_path": "$.foo", "json_val": "bar"})
+		default:
 			// "bar" is quoted in this branch because `jsonb_path_query_first` needs to be compared to a JSON value
 			listParams = listParams.Where("jsonb_path_query_first(metadata, @json_path) = @json_val", river.NamedArgs{"json_path": "$.foo", "json_val": `"bar"`})
 		}

riverdriver/riverdrivertest/driver_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jackc/pgx/v5/stdlib"
@@ -21,6 +22,7 @@ import (
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/riverdriver/riverdatabasesql"
 	"github.com/riverqueue/river/riverdriver/riverdrivertest"
+	"github.com/riverqueue/river/riverdriver/rivermysql"
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	"github.com/riverqueue/river/riverdriver/riversqlite"
 	"github.com/riverqueue/river/rivershared/riversharedtest"
@@ -203,6 +205,43 @@ func TestDriverRiverLiteLibSQL(t *testing.T) { //nolint:dupl
 		})
 }
 
+func TestDriverRiverMySQL(t *testing.T) {
+	t.Parallel()
+
+	riversharedtest.SkipIfMySQLNotEnabled(t)
+
+	var (
+		ctx    = context.Background()
+		dbPool = riversharedtest.DBPoolMySQL(ctx, t)
+		driver = rivermysql.New(dbPool)
+	)
+
+	riverdrivertest.Exercise(ctx, t,
+		func(ctx context.Context, t *testing.T, opts *riverdbtest.TestSchemaOpts) (riverdriver.Driver[*sql.Tx], string) {
+			t.Helper()
+
+			return driver, riverdbtest.TestSchema(ctx, t, driver, opts)
+		},
+		func(ctx context.Context, t *testing.T) (riverdriver.Executor, riverdriver.Driver[*sql.Tx]) {
+			t.Helper()
+
+			tx, schema := riverdbtest.TestTx(ctx, t, driver, &riverdbtest.TestTxOpts{
+				// Disable schema sharing to reduce InnoDB deadlocks from
+				// parallel tests contending on the same database.
+				DisableSchemaSharing: true,
+			})
+
+			// MySQL has no search_path equivalent, so USE the test schema
+			// database so that unqualified queries resolve correctly.
+			if schema != "" {
+				_, err := tx.ExecContext(ctx, "USE "+schema)
+				require.NoError(t, err)
+			}
+
+			return driver.UnwrapExecutor(tx), driver
+		})
+}
+
 func TestDriverRiverSQLiteModernC(t *testing.T) { //nolint:dupl
 	t.Parallel()