Cursor based pagination for batched background migrations

What does this MR do and why?

Initial sketch of cursor-based pagination

Most of the example is in a spec, but this demonstrates working cursor-based pagination across a table with composite primary key

Database migration output

Up:

main: == [advisory_lock_connection] object_id: 120820, pg_backend_pid: 13639
main: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: migrating =======
main: -- add_column(:batched_background_migration_jobs, :min_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0018s
main: -- add_column(:batched_background_migration_jobs, :max_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0006s
main: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: migrated (0.0057s) 

main: == [advisory_lock_connection] object_id: 120820, pg_backend_pid: 13639
ci: == [advisory_lock_connection] object_id: 121120, pg_backend_pid: 13641
ci: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: migrating =======
ci: -- add_column(:batched_background_migration_jobs, :min_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0015s
ci: -- add_column(:batched_background_migration_jobs, :max_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0007s
ci: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: migrated (0.0100s) 

ci: == [advisory_lock_connection] object_id: 121120, pg_backend_pid: 13641
main: == [advisory_lock_connection] object_id: 122400, pg_backend_pid: 13644
main: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: migrating ==========
main: -- add_column(:batched_background_migrations, :min_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0013s
main: -- add_column(:batched_background_migrations, :max_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0009s
main: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: migrated (0.0053s) =

main: == [advisory_lock_connection] object_id: 122400, pg_backend_pid: 13644
ci: == [advisory_lock_connection] object_id: 123640, pg_backend_pid: 13646
ci: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: migrating ==========
ci: -- add_column(:batched_background_migrations, :min_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0030s
ci: -- add_column(:batched_background_migrations, :max_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0007s
ci: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: migrated (0.0110s) =

ci: == [advisory_lock_connection] object_id: 123640, pg_backend_pid: 13646
main: == [advisory_lock_connection] object_id: 125020, pg_backend_pid: 13649
main: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: migrating 
main: -- transaction_open?(nil)
main:    -> 0.0000s
main: -- view_exists?(:postgres_partitions)
main:    -> 0.0092s
main: -- index_exists?(:batched_background_migration_jobs, "batched_background_migration_id, (max_cursor->>'ordering_key')", {:where=>"max_cursor->>'ordering_key' is not null", :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value", :algorithm=>:concurrently})
main:    -> 0.0030s
main: -- execute("SET statement_timeout TO 0")
main:    -> 0.0003s
main: -- add_index(:batched_background_migration_jobs, "batched_background_migration_id, (max_cursor->>'ordering_key')", {:where=>"max_cursor->>'ordering_key' is not null", :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value", :algorithm=>:concurrently})
main:    -> 0.0020s
main: -- execute("RESET statement_timeout")
main:    -> 0.0003s
main: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: migrated (0.0276s) 

main: == [advisory_lock_connection] object_id: 125020, pg_backend_pid: 13649
ci: == [advisory_lock_connection] object_id: 126220, pg_backend_pid: 13651
ci: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: migrating 
ci: -- transaction_open?(nil)
ci:    -> 0.0000s
ci: -- view_exists?(:postgres_partitions)
ci:    -> 0.0006s
ci: -- index_exists?(:batched_background_migration_jobs, "batched_background_migration_id, (max_cursor->>'ordering_key')", {:where=>"max_cursor->>'ordering_key' is not null", :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value", :algorithm=>:concurrently})
ci:    -> 0.0030s
ci: -- execute("SET statement_timeout TO 0")
ci:    -> 0.0003s
ci: -- add_index(:batched_background_migration_jobs, "batched_background_migration_id, (max_cursor->>'ordering_key')", {:where=>"max_cursor->>'ordering_key' is not null", :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value", :algorithm=>:concurrently})
ci:    -> 0.0020s
ci: -- execute("RESET statement_timeout")
ci:    -> 0.0003s
ci: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: migrated (0.0194s) 

ci: == [advisory_lock_connection] object_id: 126220, pg_backend_pid: 13651

Down:

ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 55154
ci: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: reverting 
ci: -- transaction_open?(nil)
ci:    -> 0.0000s
ci: -- view_exists?(:postgres_partitions)
ci:    -> 0.0079s
ci: -- indexes(:batched_background_migration_jobs)
ci:    -> 0.0036s
ci: -- execute("SET statement_timeout TO 0")
ci:    -> 0.0005s
ci: -- remove_index(:batched_background_migration_jobs, {:algorithm=>:concurrently, :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value"})
ci:    -> 0.0023s
ci: -- execute("RESET statement_timeout")
ci:    -> 0.0003s
ci: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: reverted (0.0306s) 

ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 55154
main: == [advisory_lock_connection] object_id: 120580, pg_backend_pid: 49747
main: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: reverting 
main: -- transaction_open?(nil)
main:    -> 0.0000s
main: -- view_exists?(:postgres_partitions)
main:    -> 0.0094s
main: -- indexes(:batched_background_migration_jobs)
main:    -> 0.0028s
main: -- execute("SET statement_timeout TO 0")
main:    -> 0.0003s
main: -- remove_index(:batched_background_migration_jobs, {:algorithm=>:concurrently, :name=>"index_migration_jobs_on_migration_id_and_cursor_max_value"})
main:    -> 0.0014s
main: -- execute("RESET statement_timeout")
main:    -> 0.0003s
main: == 20240711174544 IndexBackgroundMigrationJobsOnMigrationIdAndMaxCursor: reverted (0.0377s)

ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 81077
ci: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: reverting ==========
ci: -- remove_column(:batched_background_migrations, :max_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0012s
ci: -- remove_column(:batched_background_migrations, :min_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0004s
ci: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: reverted (0.0098s) =

ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 81077
main: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 86913
main: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: reverting ==========
main: -- remove_column(:batched_background_migrations, :max_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0016s
main: -- remove_column(:batched_background_migrations, :min_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0007s
main: == 20240711180222 AddCursorsToBatchedBackgroundMigrations: reverted (0.0071s) =

main: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 86913
ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 92417
ci: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: reverting =======
ci: -- remove_column(:batched_background_migration_jobs, :max_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0012s
ci: -- remove_column(:batched_background_migration_jobs, :min_cursor, :jsonb, {:null=>false, :default=>{}})
ci:    -> 0.0004s
ci: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: reverted (0.0106s) 

ci: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 92417
main: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 97831
main: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: reverting =======
main: -- remove_column(:batched_background_migration_jobs, :max_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0012s
main: -- remove_column(:batched_background_migration_jobs, :min_cursor, :jsonb, {:null=>false, :default=>{}})
main:    -> 0.0004s
main: == 20240711174515 AddCursorsToBatchedBackgroundMigrationJobs: reverted (0.0057s) 

main: == [advisory_lock_connection] object_id: 120260, pg_backend_pid: 97831

MR acceptance checklist

Please evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

Screenshots or screen recordings

Screenshots are required for UI changes, and strongly recommended for all other merge requests.

Before	After

How to set up and validate locally

Numbered steps to set up and validate the change are strongly suggested.

#452258