Backfill project_id on Ci Stages

Backfills project id on p_ci_stages through
pipelines. Destroys orphaned records.

Changelog: changed

db/post_migrate/20241003225057_backfill_null_project_on_ci_stages_records.rb

+# frozen_string_literal: true
+
+class BackfillNullProjectOnCiStagesRecords < Gitlab::Database::Migration[2.2]
+  milestone '17.6'
+  restrict_gitlab_migration gitlab_schema: :gitlab_ci
+  disable_ddl_transaction!
+
+  def up
+    stages_model = define_batchable_model('p_ci_stages', primary_key: :id)
+
+    loop do
+      batch = stages_model.where(project_id: nil).limit(100).to_a
+      break if batch.empty?
+
+      stages_model
+        .where(id: batch.pluck(:id))
+        .where('p_ci_stages.pipeline_id = p_ci_pipelines.id')
+        .where('p_ci_stages.partition_id = p_ci_pipelines.partition_id')
+        .update_all('project_id = p_ci_pipelines.project_id FROM p_ci_pipelines')
+
+      stages_model
+        .where(id: batch.pluck(:id))
+        .where(project_id: nil)
+        .delete_all
+    end
+  end
+
+  def down
+    # no-op
+    # Not reversable
+  end
+end

db/schema_migrations/20241003225057

+fd3f1a27204fe2952e2f83a94f92737edfd1ff7b446e5db98b052bb03e3373d3
\ No newline at end of file

spec/migrations/db/post_migrate/20241003225057_backfill_null_project_on_ci_stages_records_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe BackfillNullProjectOnCiStagesRecords, migration: :gitlab_ci, feature_category: :database do
+  let(:pipelines_table) { table(:p_ci_pipelines, primary_key: :id, database: :ci) }
+  let(:stages_table) { table(:p_ci_stages, primary_key: :id, database: :ci) }
+
+  let!(:pipeline_with_project) { pipelines_table.create!(project_id: 42, partition_id: 100) }
+  let!(:stage_to_be_backfilled) do
+    stages_table.create!(name: 'backfilled', pipeline_id: pipeline_with_project.id, partition_id: 100)
+  end
+
+  let!(:stage_with_project_id) { stages_table.create!(name: 'standard', project_id: 11, partition_id: 100) }
+  let!(:stage_orphaned) { stages_table.create!(name: 'deleted', partition_id: 100) }
+
+  describe '#up' do
+    it 'backfills applicable records' do
+      expect { migrate! }
+      .to change { stage_to_be_backfilled.reload.project_id }.from(nil).to(pipeline_with_project.project_id)
+      .and not_change { stage_with_project_id.reload.project_id }.from(11)
+    end
+
+    it 'deletes orphaned records' do
+      expect { migrate! }.to change { stages_table.count }.by(-1)
+
+      expect { stage_orphaned.reload }.to raise_error(ActiveRecord::RecordNotFound)
+    end
+  end
+end