Refactor Postgres migrator

quickwit/Makefile

@@ -28,6 +28,7 @@ migrations-lint:
 	@echo "Linting Postgres migrations with diesel-guard"
 	@(command -v diesel-guard >/dev/null || (echo "diesel-guard is not installed. Please install using 'cargo install diesel-guard'." && exit 1))
 	@diesel-guard check quickwit-metastore/migrations/postgresql/
+	@diesel-guard check quickwit-metastore/migrations/postgresql_deferred/
 
 unused-deps:
 	@echo "Checking for unused dependencies"

quickwit/quickwit-metastore/README.md

@@ -53,3 +53,13 @@ You can then use the following commands to apply/revert your postgresql migratio
 sqlx migrate run  --database-url postgres://quickwit-dev:quickwit-dev@localhost:5432/quickwit-metastore-dev --source migrations/postgresql
 sqlx migrate revert  --database-url postgres://quickwit-dev:quickwit-dev@localhost:5432/quickwit-metastore-dev --source migrations/postgresql
 ```
+
+## Deferred migrations
+
+`migrations/postgresql_deferred` holds long-running, degrade-gracefully migrations (e.g. `CREATE INDEX CONCURRENTLY`).
+They run in a background task after readiness, elected across pods by a Postgres advisory lock, and share the
+`_sqlx_migrations` table with the required track, so version numbers must be globally unique across both dirs and each
+migration must be idempotent. See `migrations/postgresql_deferred/README.md` for authoring rules.
+```
+sqlx migrate run  --database-url postgres://quickwit-dev:quickwit-dev@localhost:5432/quickwit-metastore-dev --source migrations/postgresql_deferred
+```

quickwit/quickwit-metastore/build.rs

@@ -14,4 +14,5 @@
 
 fn main() {
     println!("cargo:rerun-if-changed=migrations/postgresql");
+    println!("cargo:rerun-if-changed=migrations/postgresql_deferred");
 }

quickwit/quickwit-metastore/migrations/postgresql_deferred/README.md

@@ -0,0 +1,22 @@
+# Deferred PostgreSQL migrations
+
+Long, degrade-gracefully-only migrations (e.g. `CREATE INDEX CONCURRENTLY`) run in a background task after readiness. Both tracks share the `_sqlx_migrations` table.
+
+- Version must be globally unique across both dirs (continue the single sequence).
+- Must be idempotent. A statement that can't run in a transaction (e.g. `CREATE INDEX CONCURRENTLY`) needs `-- no-transaction` as the first line, then the DDL. Each statement auto-commits, so a killed migration must be safe to re-run (`CREATE INDEX CONCURRENTLY` can leave an invalid index, so drop it first). For example:
+
+29_add_foo_index.up.sql
+  ```sql
+  -- no-transaction
+  DROP INDEX CONCURRENTLY IF EXISTS foo_idx;
+  CREATE INDEX CONCURRENTLY IF NOT EXISTS foo_idx ON foo (bar);
+  ```
+
+29_add_foo_index.down.sql
+  ```sql
+  -- no-transaction
+  DROP INDEX CONCURRENTLY IF EXISTS foo_idx;
+  ```
+
+  The `DROP` before the `CREATE` looks like it would drop the index on every run, but it does not: sqlx applies each migration at most once (tracked by version in `_sqlx_migrations`) and never re-runs one that already succeeded. The drop only matters on a retry after a partial failure. If a `CREATE INDEX CONCURRENTLY` is killed midway, Postgres leaves an *invalid* index and sqlx records nothing (the bookkeeping row is written only on success), so the next attempt re-runs the migration. A bare `CREATE INDEX CONCURRENTLY IF NOT EXISTS` would then see that invalid index and skip it forever, so we drop it first to force a clean rebuild.
+- Never depend on an unshipped required migration; required migrations must never depend on a deferred one.

quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs

@@ -61,17 +61,14 @@ use time::OffsetDateTime;
 use tracing::{debug, info, instrument, warn};
 use uuid::Uuid;
 
+use super::QW_POSTGRES_READ_ONLY_ENV_KEY;
 use super::error::convert_sqlx_err;
-use super::migrator::run_migrations;
+use super::migrator::Migrations;
 use super::model::{PgDeleteTask, PgIndex, PgIndexTemplate, PgShard, PgSplit, Splits};
 use super::parquet_model::{InsertableParquetSplit, ParquetSplitRecord, PgParquetSplit};
 use super::pool::TrackedPool;
 use super::split_stream::SplitStream;
 use super::utils::{append_query_filters_and_order_by, establish_connection};
-use super::{
-    QW_POSTGRES_READ_ONLY_ENV_KEY, QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY,
-    QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY,
-};
 use crate::checkpoint::{
     IndexCheckpointDelta, PartitionId, SourceCheckpoint, SourceCheckpointDelta,
 };
@@ -123,8 +120,6 @@ impl PostgresqlMetastore {
             .expect("PostgreSQL metastore config should have been validated");
 
         let read_only = get_bool_from_env(QW_POSTGRES_READ_ONLY_ENV_KEY, false);
-        let skip_migrations = get_bool_from_env(QW_POSTGRES_SKIP_MIGRATIONS_ENV_KEY, false);
-        let skip_locking = get_bool_from_env(QW_POSTGRES_SKIP_MIGRATION_LOCKING_ENV_KEY, false);
 
         let connection_pool = establish_connection(
             connection_uri,
@@ -137,7 +132,7 @@ impl PostgresqlMetastore {
         )
         .await?;
 
-        run_migrations(&connection_pool, skip_migrations, skip_locking).await?;
+        Migrations::from_env(connection_pool.clone()).run().await?;
 
         let metastore = PostgresqlMetastore {
             uri: connection_uri.clone(),

quickwit/quickwit-metastore/src/metastore/postgres/metrics.rs

@@ -12,7 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use quickwit_metrics::{LazyGauge, lazy_gauge};
+use quickwit_metrics::{LazyCounter, LazyGauge, lazy_counter, lazy_gauge};
+
+// Counts deferred-migration apply attempts, labeled by `result` ("success"/"failure").
+pub(super) static DEFERRED_MIGRATIONS_APPLY: LazyCounter = lazy_counter!(
+        name: "deferred_migrations_apply_total",
+        description: "Number of deferred PostgreSQL migration attempts, by result.",
+        subsystem: "metastore",
+);
 
 pub(super) static ACQUIRE_CONNECTIONS: LazyGauge = lazy_gauge!(
         name: "acquire_connections",