fix: add index on GSI base key columns for efficient propagation deletes

crates/storage-postgres/src/bootstrapper.rs

@@ -206,7 +206,11 @@ impl Bootstrapper for PostgresBootstrapper {
 
     async fn run_data_migrations(&self) -> OpResult<()> {
         let pool = self.app_pool(&self.config.data_db).await?;
-        migrations::run_data_migrations(&pool).await
+        migrations::run_data_migrations(&pool).await?;
+        // Programmatic migrations need the catalog pool (to enumerate index
+        // tables) plus the data pool (where the `_ddb_*` tables live).
+        let catalog_pool = self.app_pool(&self.config.catalog_db).await?;
+        migrations::run_data_code_migrations(&catalog_pool, &pool).await
     }
 
     async fn pending_data_migrations(&self) -> OpResult<Vec<String>> {

crates/storage-postgres/src/data/ddl.rs

@@ -229,6 +229,28 @@ impl PostgresEngine {
                 .map_err(|e| StorageError::Internal(e.to_string()))?;
         }
 
+        // Index on base table key columns for delete_index_row_multi lookups.
+        {
+            let mut base_key_cols = vec!["base_pk".to_owned()];
+            for (i, &(_, sk_type)) in base_sks.iter().enumerate() {
+                let col = if i == 0 {
+                    format!("base_{}", sk_column(sk_type))
+                } else {
+                    format!("base_{}", sk_column_n(i, sk_type))
+                };
+                base_key_cols.push(col);
+            }
+            let idx_name = format!("_ddb_{index_id}_base_key_idx");
+            let base_key_idx = format!(
+                "CREATE INDEX \"{idx_name}\" ON {idx_table} ({})",
+                base_key_cols.join(", ")
+            );
+            sqlx::query(&base_key_idx)
+                .execute(&mut **tx)
+                .await
+                .map_err(|e| StorageError::Internal(e.to_string()))?;
+        }
+
         Ok(())
     }
 

crates/storage-postgres/src/migrations.rs

@@ -93,6 +93,102 @@ pub(crate) async fn run_data_migrations(pool: &PgPool) -> OpResult<()> {
     Ok(())
 }
 
+/// Programmatic ("code") data migrations, tracked in `schema_history` alongside
+/// the SQL migrations. Unlike a static `.sql` file, these enumerate the
+/// dynamically-named index tables (`_ddb_<id>`) from the catalog and must run
+/// outside a transaction (they use `CREATE INDEX CONCURRENTLY`), so they cannot
+/// be expressed as SQL in [`DATA_MIGRATIONS`]. Applied by `extenddb migrate`
+/// after the SQL migrations, so the operator controls when the change happens.
+pub(crate) const DATA_CODE_MIGRATIONS: &[&str] = &["003_gsi_base_key_index"];
+
+/// Run programmatic data migrations, skipping already-applied ones.
+///
+/// Needs the catalog pool (to enumerate index tables and their base key schema)
+/// and the data pool (where the `_ddb_*` tables and the `schema_history` ledger
+/// live). Each step is recorded in `schema_history` and skipped on later runs,
+/// exactly like the SQL migrations.
+pub(crate) async fn run_data_code_migrations(
+    catalog_pool: &PgPool,
+    data_pool: &PgPool,
+) -> OpResult<()> {
+    for name in DATA_CODE_MIGRATIONS {
+        if is_migration_applied(data_pool, name).await? {
+            println!("    {name} — already applied, skipping.");
+            continue;
+        }
+        println!("    Applying {name}...");
+        match *name {
+            "003_gsi_base_key_index" => {
+                ensure_gsi_base_key_indexes(catalog_pool, data_pool).await?;
+            }
+            other => {
+                return Err(OpError::Internal(format!(
+                    "Unknown data code migration: {other}"
+                )));
+            }
+        }
+        record_migration(data_pool, name).await?;
+    }
+    Ok(())
+}
+
+/// Create the base-table-key index on every existing GSI/LSI table.
+///
+/// During GSI propagation each index table (`_ddb_<id>`) is looked up back to
+/// its base item via `WHERE base_pk = $1 AND base_sk_* = $2`; without a leading
+/// `(base_pk, base_sk_*)` index that is a sequential scan. New tables get this
+/// index at creation time (see `ddl.rs`); this migration adds it to tables
+/// created before the index existed. `CREATE INDEX CONCURRENTLY IF NOT EXISTS`
+/// is idempotent and does not block concurrent writes.
+async fn ensure_gsi_base_key_indexes(catalog_pool: &PgPool, data_pool: &PgPool) -> OpResult<()> {
+    use extenddb_core::types::{AttributeDefinition, KeySchemaElement};
+    use extenddb_storage::util::{sk_column, sk_column_n};
+
+    // Enumerate every index and its base table key schema from the catalog.
+    let rows: Vec<(String, serde_json::Value, serde_json::Value)> = sqlx::query_as(
+        "SELECT i.index_id, t.key_schema, t.attribute_definitions \
+         FROM indexes i \
+         JOIN tables t ON i.table_id = t.table_id",
+    )
+    .fetch_all(catalog_pool)
+    .await
+    .map_err(|e| OpError::Internal(format!("Enumerate indexes: {e}")))?;
+
+    for (index_id, ks_json, ad_json) in rows {
+        let base_ks: Vec<KeySchemaElement> =
+            serde_json::from_value(ks_json).map_err(|e| OpError::Internal(e.to_string()))?;
+        let attr_defs: Vec<AttributeDefinition> =
+            serde_json::from_value(ad_json).map_err(|e| OpError::Internal(e.to_string()))?;
+
+        let base_sks = crate::data::all_sort_key_info(&base_ks, &attr_defs);
+        let idx_table = crate::data::index_table_name(&index_id);
+        let idx_name = format!("_ddb_{index_id}_base_key_idx");
+
+        let mut base_key_cols = vec!["base_pk".to_owned()];
+        for (i, &(_, sk_type)) in base_sks.iter().enumerate() {
+            let col = if i == 0 {
+                format!("base_{}", sk_column(sk_type))
+            } else {
+                format!("base_{}", sk_column_n(i, sk_type))
+            };
+            base_key_cols.push(col);
+        }
+
+        // CONCURRENTLY cannot run inside a transaction, so execute on the pool.
+        let sql = format!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS \"{}\" ON {} ({})",
+            idx_name,
+            idx_table,
+            base_key_cols.join(", ")
+        );
+        sqlx::query(&sql)
+            .execute(data_pool)
+            .await
+            .map_err(|e| OpError::Internal(format!("Base key index on {idx_table}: {e}")))?;
+    }
+    Ok(())
+}
+
 /// Check if a table exists in the public schema.
 pub(crate) async fn table_exists(pool: &PgPool, name: &str) -> OpResult<bool> {
     let exists: bool = sqlx::query_scalar(
@@ -130,6 +226,12 @@ pub(crate) async fn pending_data_migrations(pool: &PgPool) -> OpResult<Vec<Strin
         }
         pending.push((*filename).to_owned());
     }
+    // Code migrations are tracked in the same data-database ledger.
+    for name in DATA_CODE_MIGRATIONS {
+        if !is_migration_applied(pool, name).await? {
+            pending.push((*name).to_owned());
+        }
+    }
     Ok(pending)
 }
 

tests/test_cli_lifecycle.py

@@ -81,6 +81,9 @@ def test_data_migrations_tracked(self, cli_env):
 
         assert "001_data_schema.sql" in tracked, tracked
         assert "002_gsi_pending.sql" in tracked, tracked
+        # Programmatic (code) migration for the GSI base-key index is tracked in
+        # the same ledger, so `migrate` treats it exactly like a SQL migration.
+        assert "003_gsi_base_key_index" in tracked, tracked
 
     def test_migrate_applies_pending_data_migration(self, cli_env):
         """`extenddb migrate` applies a pending *data* migration on an existing
@@ -161,6 +164,90 @@ def _data_conn():
         finally:
             conn.close()
 
+    def test_migrate_applies_base_key_index_code_migration(self, cli_env):
+        """`extenddb migrate` applies the programmatic GSI base-key-index
+        migration (`003_gsi_base_key_index`) and never re-runs it.
+
+        The base-key index makes GSI-propagation deletes (`WHERE base_pk = $1
+        AND base_sk_* = $2`) index scans instead of sequential scans. New tables
+        get it at creation time; this migration adds it to tables created before
+        the index existed. It is a *code* migration (it enumerates the
+        dynamically-named `_ddb_*` tables and uses `CREATE INDEX CONCURRENTLY`),
+        so this verifies it is wired into the same pending/apply/tracking flow as
+        the SQL migrations: pending is detected, `migrate` refuses without
+        `--yes`, applies with `--yes`, records the ledger row, and is idempotent
+        on a second run. (Index derivation itself is the same logic exercised at
+        table-creation time across the GSI integration suite.)
+        """
+        import psycopg2
+
+        result = _run_extenddb(
+            "init", *_init_args(cli_env),
+            config=cli_env["config_path"],
+            env_override={"EXTENDDB_ADMIN_PASSWORD": "TestPass1!"},
+        )
+        assert result.returncode == 0
+
+        data_db = cli_env["db_name"][: -len("_catalog")]
+
+        def _data_conn():
+            return psycopg2.connect(PG_ADMIN_CONN + "/" + data_db)
+
+        # Simulate a deployment created before 003 existed: drop only its ledger
+        # row so `migrate` sees the code migration as pending again.
+        conn = _data_conn()
+        try:
+            conn.autocommit = True
+            with conn.cursor() as cur:
+                cur.execute(
+                    "DELETE FROM schema_history "
+                    "WHERE filename = '003_gsi_base_key_index'"
+                )
+                cur.execute(
+                    "SELECT EXISTS(SELECT 1 FROM schema_history "
+                    "WHERE filename = '003_gsi_base_key_index')"
+                )
+                assert cur.fetchone()[0] is False
+        finally:
+            conn.close()
+
+        # Without --yes, migrate must refuse while the code migration is pending.
+        refused = _run_extenddb(
+            "migrate", *_pg_args(),
+            config=cli_env["config_path"],
+            check=False,
+        )
+        assert refused.returncode != 0, refused.stdout + refused.stderr
+        assert "003_gsi_base_key_index" in (refused.stdout + refused.stderr)
+
+        # With --yes, migrate applies and records the code migration.
+        applied = _run_extenddb(
+            "migrate", "--yes", *_pg_args(),
+            config=cli_env["config_path"],
+            check=False,
+        )
+        assert applied.returncode == 0, applied.stdout + applied.stderr
+
+        conn = _data_conn()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT EXISTS(SELECT 1 FROM schema_history "
+                    "WHERE filename = '003_gsi_base_key_index')"
+                )
+                assert cur.fetchone()[0] is True
+        finally:
+            conn.close()
+
+        # Idempotent: a second migrate finds nothing pending.
+        again = _run_extenddb(
+            "migrate", "--yes", *_pg_args(),
+            config=cli_env["config_path"],
+            check=False,
+        )
+        assert again.returncode == 0, again.stdout + again.stderr
+        assert "up to date" in again.stdout.lower(), again.stdout
+
     def test_init_serve_status_stop(self, cli_env):
         """Full lifecycle: init → serve → status → stop."""
         # Init