diff --git a/lib/gitlab/database/migration_helpers.rb b/lib/gitlab/database/migration_helpers.rb
index c3a7b384172ce663465402a0fcbe12d0e22fdac5..82adb0dd369b884af94116fdfda11a47b28d1414 100644
--- a/lib/gitlab/database/migration_helpers.rb
+++ b/lib/gitlab/database/migration_helpers.rb
@@ -292,23 +292,34 @@ def remove_concurrent_index_by_name(table_name, index_name, options = {})
       #                      order of the ALTER TABLE. This can be useful in situations where the foreign
       #                      key creation could deadlock with another process.
       #
-      # rubocop: disable Metrics/ParameterLists
-      def add_concurrent_foreign_key(source, target, column:, on_delete: :cascade, on_update: nil, target_column: :id, name: nil, validate: true, reverse_lock_order: false)
+      def add_concurrent_foreign_key(source, target, column:, **options)
+        options.reverse_merge!({
+          on_delete: :cascade,
+          on_update: nil,
+          target_column: :id,
+          validate: true,
+          reverse_lock_order: false,
+          allow_partitioned: false,
+          column: column
+        })
+
         # Transactions would result in ALTER TABLE locks being held for the
         # duration of the transaction, defeating the purpose of this method.
         if transaction_open?
           raise 'add_concurrent_foreign_key can not be run inside a transaction'
         end
 
-        options = {
-          column: column,
-          on_delete: on_delete,
-          on_update: on_update,
-          name: name.presence || concurrent_foreign_key_name(source, column),
-          primary_key: target_column
-        }
+        if !options.delete(:allow_partitioned) && table_partitioned?(source)
+          raise ArgumentError, 'add_concurrent_foreign_key can not be used on a partitioned ' \
+            'table. Please use add_concurrent_partitioned_foreign_key on the partitioned table ' \
+            'as we need to create foreign keys on each partition and a FK on the parent table'
+        end
 
-        if foreign_key_exists?(source, target, **options)
+        options[:name] ||= concurrent_foreign_key_name(source, column)
+        options[:primary_key] = options[:target_column]
+        check_options = options.slice(:column, :on_delete, :on_update, :name, :primary_key)
+
+        if foreign_key_exists?(source, target, **check_options)
           warning_message = "Foreign key not created because it exists already " \
             "(this may be due to an aborted migration or similar): " \
             "source: #{source}, target: #{target}, column: #{options[:column]}, "\
@@ -321,17 +332,18 @@ def add_concurrent_foreign_key(source, target, column:, on_delete: :cascade, on_
           # validating it. This means we keep the ALTER TABLE lock only for a
           # short period of time. The key _is_ enforced for any newly created
           # data.
+          not_valid_option = 'NOT VALID' unless table_partitioned?(source)
 
           with_lock_retries do
-            execute("LOCK TABLE #{target}, #{source} IN SHARE ROW EXCLUSIVE MODE") if reverse_lock_order
+            execute("LOCK TABLE #{target}, #{source} IN SHARE ROW EXCLUSIVE MODE") if options[:reverse_lock_order]
             execute <<-EOF.strip_heredoc
             ALTER TABLE #{source}
             ADD CONSTRAINT #{options[:name]}
             FOREIGN KEY (#{multiple_columns(options[:column])})
-            REFERENCES #{target} (#{multiple_columns(target_column)})
+            REFERENCES #{target} (#{multiple_columns(options[:target_column])})
             #{on_update_statement(options[:on_update])}
             #{on_delete_statement(options[:on_delete])}
-            NOT VALID;
+            #{not_valid_option};
             EOF
           end
         end
@@ -345,13 +357,12 @@ def add_concurrent_foreign_key(source, target, column:, on_delete: :cascade, on_
         #
         # Note this is a no-op in case the constraint is VALID already
 
-        if validate
+        if options[:validate]
           disable_statement_timeout do
             execute("ALTER TABLE #{source} VALIDATE CONSTRAINT #{options[:name]};")
           end
         end
       end
-      # rubocop: enable Metrics/ParameterLists
 
       def validate_foreign_key(source, column, name: nil)
         fk_name = name || concurrent_foreign_key_name(source, column)
@@ -1239,6 +1250,12 @@ def partition?(table_name)
         end
       end
 
+      def table_partitioned?(table_name)
+        Gitlab::Database::PostgresPartitionedTable
+          .find_by_name_in_current_schema(table_name)
+          .present?
+      end
+
       private
 
       def multiple_columns(columns, separator: ', ')
diff --git a/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers.rb b/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers.rb
index 8849191f356416affafea3bcb0bd38abaefdd88a..7d9c12d776e588da4c6001b8770e07f15b6f29cc 100644
--- a/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers.rb
+++ b/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers.rb
@@ -32,46 +32,75 @@ module ForeignKeyHelpers
         # column - The name of the column to create the foreign key on.
         # on_delete - The action to perform when associated data is removed,
         #             defaults to "CASCADE".
+        # on_update - The action to perform when associated data is updated,
+        #             no default value is set.
         # name - The name of the foreign key.
+        # validate - Flag that controls whether the new foreign key will be
+        #            validated after creation and if it will be added on the parent table.
+        #            If the flag is not set, the constraint will only be enforced for new data
+        #            in the existing partitions. The helper will need to be called again
+        #            with the flag set to `true` to add the foreign key on the parent table
+        #            after validating it on all partitions.
+        #            `validate: false` should be paired with `prepare_partitioned_async_foreign_key_validation`
+        # reverse_lock_order - Flag that controls whether we should attempt to acquire locks in the reverse
+        #                      order of the ALTER TABLE. This can be useful in situations where the foreign
+        #                      key creation could deadlock with another process.
         #
-        def add_concurrent_partitioned_foreign_key(source, target, column:, on_delete: :cascade, name: nil)
+        def add_concurrent_partitioned_foreign_key(source, target, column:, **options)
           assert_not_in_transaction_block(scope: ERROR_SCOPE)
 
-          partition_options = {
-            column: column,
-            on_delete: on_delete,
+          options.reverse_merge!({
+            target_column: :id,
+            on_delete: :cascade,
+            on_update: nil,
+            name: nil,
+            validate: true,
+            reverse_lock_order: false,
+            column: column
+          })
 
-            # We'll use the same FK name for all partitions and match it to
-            # the name used for the partitioned table to follow the convention
-            # used by PostgreSQL when adding FKs to new partitions
-            name: name.presence || concurrent_partitioned_foreign_key_name(source, column),
+          # We'll use the same FK name for all partitions and match it to
+          # the name used for the partitioned table to follow the convention
+          # used by PostgreSQL when adding FKs to new partitions
+          options[:name] ||= concurrent_partitioned_foreign_key_name(source, column)
+          check_options = options.slice(:column, :on_delete, :on_update, :name)
+          check_options[:primary_key] = options[:target_column]
 
-            # Force the FK validation to true for partitions (and the partitioned table)
-            validate: true
-          }
-
-          if foreign_key_exists?(source, target, **partition_options)
+          if foreign_key_exists?(source, target, **check_options)
             warning_message = "Foreign key not created because it exists already " \
               "(this may be due to an aborted migration or similar): " \
-              "source: #{source}, target: #{target}, column: #{partition_options[:column]}, "\
-              "name: #{partition_options[:name]}, on_delete: #{partition_options[:on_delete]}"
+              "source: #{source}, target: #{target}, column: #{options[:column]}, "\
+              "name: #{options[:name]}, on_delete: #{options[:on_delete]}, "\
+              "on_update: #{options[:on_update]}"
 
             Gitlab::AppLogger.warn warning_message
 
             return
           end
 
-          partitioned_table = find_partitioned_table(source)
-
-          partitioned_table.postgres_partitions.order(:name).each do |partition|
-            add_concurrent_foreign_key(partition.identifier, target, **partition_options)
+          Gitlab::Database::PostgresPartitionedTable.each_partition(source) do |partition|
+            add_concurrent_foreign_key(partition.identifier, target, **options)
           end
 
-          with_lock_retries do
-            add_foreign_key(source, target, **partition_options)
+          # If we are to add the FK on the parent table now, it will trigger
+          # the validation on all partitions. The helper must be called one
+          # more time with `validate: true` after the FK is valid on all partitions.
+          return unless options[:validate]
+
+          options[:allow_partitioned] = true
+          add_concurrent_foreign_key(source, target, **options)
+        end
+
+        def validate_partitioned_foreign_key(source, column, name: nil)
+          assert_not_in_transaction_block(scope: ERROR_SCOPE)
+
+          Gitlab::Database::PostgresPartitionedTable.each_partition(source) do |partition|
+            validate_foreign_key(partition.identifier, column, name: name)
           end
         end
 
+        private
+
         # Returns the name for a concurrent partitioned foreign key.
         #
         # Similar to concurrent_foreign_key_name (Gitlab::Database::MigrationHelpers)
diff --git a/spec/lib/gitlab/database/migration_helpers_spec.rb b/spec/lib/gitlab/database/migration_helpers_spec.rb
index 63a470c82a1ffe0c15b4083d5ef8147db9df1a72..1af69f85a54eab35937c5f93ff372ada03f7a9d0 100644
--- a/spec/lib/gitlab/database/migration_helpers_spec.rb
+++ b/spec/lib/gitlab/database/migration_helpers_spec.rb
@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe Gitlab::Database::MigrationHelpers do
+RSpec.describe Gitlab::Database::MigrationHelpers, feature_category: :database do
   include Database::TableSchemaHelpers
   include Database::TriggerHelpers
 
@@ -979,6 +979,65 @@
           end
         end
       end
+
+      context 'when creating foreign key on a partitioned table' do
+        before do
+          model.execute(<<~SQL)
+            CREATE TABLE public._test_source_partitioned_table (
+              id serial NOT NULL,
+              partition_id serial NOT NULL,
+              owner_id bigint NOT NULL,
+              PRIMARY KEY (id, partition_id)
+            ) PARTITION BY LIST(partition_id);
+
+            CREATE TABLE _test_source_partitioned_table_1
+              PARTITION OF public._test_source_partitioned_table
+              FOR VALUES IN (1);
+
+            CREATE TABLE public._test_dest_partitioned_table (
+              id serial NOT NULL,
+              partition_id serial NOT NULL,
+              PRIMARY KEY (id, partition_id)
+            );
+          SQL
+        end
+
+        it 'creates the FK without using NOT VALID', :aggregate_failures do
+          allow(model).to receive(:execute).and_call_original
+
+          expect(model).to receive(:with_lock_retries).and_yield
+
+          expect(model).to receive(:execute).with(
+            "ALTER TABLE _test_source_partitioned_table\n" \
+            "ADD CONSTRAINT fk_multiple_columns\n" \
+            "FOREIGN KEY \(partition_id, owner_id\)\n" \
+            "REFERENCES _test_dest_partitioned_table \(partition_id, id\)\n" \
+            "ON UPDATE CASCADE\n" \
+            "ON DELETE CASCADE\n;\n"
+          )
+
+          model.add_concurrent_foreign_key(
+            :_test_source_partitioned_table,
+            :_test_dest_partitioned_table,
+            column: [:partition_id, :owner_id],
+            target_column: [:partition_id, :id],
+            name: :fk_multiple_columns,
+            on_update: :cascade,
+            allow_partitioned: true
+          )
+        end
+
+        it 'raises an error if allow_partitioned is not set' do
+          expect(model).not_to receive(:with_lock_retries).and_yield
+          expect(model).not_to receive(:execute).with(/FOREIGN KEY/)
+
+          args = [:_test_source_partitioned_table, :_test_dest_partitioned_table]
+          options = { column: [:partition_id, :owner_id], target_column: [:partition_id, :id] }
+
+          expect { model.add_concurrent_foreign_key(*args, **options) }
+            .to raise_error ArgumentError, /use add_concurrent_partitioned_foreign_key/
+        end
+      end
     end
   end
 
@@ -2896,4 +2955,18 @@ def setup
       it { is_expected.to be_falsey }
     end
   end
+
+  describe "#table_partitioned?" do
+    subject { model.table_partitioned?(table_name) }
+
+    let(:table_name) { 'p_ci_builds_metadata' }
+
+    it { is_expected.to be_truthy }
+
+    context 'with a non-partitioned table' do
+      let(:table_name) { 'users' }
+
+      it { is_expected.to be_falsey }
+    end
+  end
 end
diff --git a/spec/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers_spec.rb b/spec/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers_spec.rb
index f0e34476cf28bd1cf3f23e0596e1e401de95dcf2..d5f4afd7ba46db96f8d642efc695efd20a300d07 100644
--- a/spec/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers_spec.rb
+++ b/spec/lib/gitlab/database/partitioning_migration_helpers/foreign_key_helpers_spec.rb
@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe Gitlab::Database::PartitioningMigrationHelpers::ForeignKeyHelpers do
+RSpec.describe Gitlab::Database::PartitioningMigrationHelpers::ForeignKeyHelpers, feature_category: :database do
   include Database::TableSchemaHelpers
 
   let(:migration) do
@@ -16,15 +16,23 @@
   let(:partition_schema) { 'gitlab_partitions_dynamic' }
   let(:partition1_name) { "#{partition_schema}.#{source_table_name}_202001" }
   let(:partition2_name) { "#{partition_schema}.#{source_table_name}_202002" }
+  let(:validate) { true }
   let(:options) do
     {
       column: column_name,
       name: foreign_key_name,
       on_delete: :cascade,
-      validate: true
+      on_update: nil,
+      primary_key: :id
     }
   end
 
+  let(:create_options) do
+    options
+      .except(:primary_key)
+      .merge!(reverse_lock_order: false, target_column: :id, validate: validate)
+  end
+
   before do
     allow(migration).to receive(:puts)
     allow(migration).to receive(:transaction_open?).and_return(false)
@@ -67,12 +75,11 @@
 
         expect(migration).to receive(:concurrent_partitioned_foreign_key_name).and_return(foreign_key_name)
 
-        expect_add_concurrent_fk_and_call_original(partition1_name, target_table_name, **options)
-        expect_add_concurrent_fk_and_call_original(partition2_name, target_table_name, **options)
+        expect_add_concurrent_fk_and_call_original(partition1_name, target_table_name, **create_options)
+        expect_add_concurrent_fk_and_call_original(partition2_name, target_table_name, **create_options)
 
-        expect(migration).to receive(:with_lock_retries).ordered.and_yield
-        expect(migration).to receive(:add_foreign_key)
-          .with(source_table_name, target_table_name, **options)
+        expect(migration).to receive(:add_concurrent_foreign_key)
+          .with(source_table_name, target_table_name, allow_partitioned: true, **create_options)
           .ordered
           .and_call_original
 
@@ -81,6 +88,39 @@
         expect_foreign_key_to_exist(source_table_name, foreign_key_name)
       end
 
+      context 'with validate: false option' do
+        let(:validate) { false }
+        let(:options) do
+          {
+            column: column_name,
+            name: foreign_key_name,
+            on_delete: :cascade,
+            on_update: nil,
+            primary_key: :id
+          }
+        end
+
+        it 'creates the foreign key only on partitions' do
+          expect(migration).to receive(:foreign_key_exists?)
+            .with(source_table_name, target_table_name, **options)
+            .and_return(false)
+
+          expect(migration).to receive(:concurrent_partitioned_foreign_key_name).and_return(foreign_key_name)
+
+          expect_add_concurrent_fk_and_call_original(partition1_name, target_table_name, **create_options)
+          expect_add_concurrent_fk_and_call_original(partition2_name, target_table_name, **create_options)
+
+          expect(migration).not_to receive(:add_concurrent_foreign_key)
+            .with(source_table_name, target_table_name, **create_options)
+
+          migration.add_concurrent_partitioned_foreign_key(
+            source_table_name, target_table_name,
+            column: column_name, validate: false)
+
+          expect_foreign_key_not_to_exist(source_table_name, foreign_key_name)
+        end
+      end
+
       def expect_add_concurrent_fk_and_call_original(source_table_name, target_table_name, options)
         expect(migration).to receive(:add_concurrent_foreign_key)
           .ordered
@@ -100,8 +140,6 @@ def expect_add_concurrent_fk_and_call_original(source_table_name, target_table_n
           .and_return(true)
 
         expect(migration).not_to receive(:add_concurrent_foreign_key)
-        expect(migration).not_to receive(:with_lock_retries)
-        expect(migration).not_to receive(:add_foreign_key)
 
         migration.add_concurrent_partitioned_foreign_key(source_table_name, target_table_name, column: column_name)
 
@@ -110,30 +148,43 @@ def expect_add_concurrent_fk_and_call_original(source_table_name, target_table_n
     end
 
     context 'when additional foreign key options are given' do
-      let(:options) do
+      let(:exits_options) do
         {
           column: column_name,
           name: '_my_fk_name',
           on_delete: :restrict,
-          validate: true
+          on_update: nil,
+          primary_key: :id
         }
       end
 
+      let(:create_options) do
+        exits_options
+          .except(:primary_key)
+          .merge!(reverse_lock_order: false, target_column: :id, validate: true)
+      end
+
       it 'forwards them to the foreign key helper methods' do
         expect(migration).to receive(:foreign_key_exists?)
-          .with(source_table_name, target_table_name, **options)
+          .with(source_table_name, target_table_name, **exits_options)
           .and_return(false)
 
         expect(migration).not_to receive(:concurrent_partitioned_foreign_key_name)
 
-        expect_add_concurrent_fk(partition1_name, target_table_name, **options)
-        expect_add_concurrent_fk(partition2_name, target_table_name, **options)
+        expect_add_concurrent_fk(partition1_name, target_table_name, **create_options)
+        expect_add_concurrent_fk(partition2_name, target_table_name, **create_options)
 
-        expect(migration).to receive(:with_lock_retries).ordered.and_yield
-        expect(migration).to receive(:add_foreign_key).with(source_table_name, target_table_name, **options).ordered
+        expect(migration).to receive(:add_concurrent_foreign_key)
+          .with(source_table_name, target_table_name, allow_partitioned: true, **create_options)
+          .ordered
 
-        migration.add_concurrent_partitioned_foreign_key(source_table_name, target_table_name,
-          column: column_name, name: '_my_fk_name', on_delete: :restrict)
+        migration.add_concurrent_partitioned_foreign_key(
+          source_table_name,
+          target_table_name,
+          column: column_name,
+          name: '_my_fk_name',
+          on_delete: :restrict
+        )
       end
 
       def expect_add_concurrent_fk(source_table_name, target_table_name, options)
@@ -153,4 +204,39 @@ def expect_add_concurrent_fk(source_table_name, target_table_name, options)
       end
     end
   end
+
+  describe '#validate_partitioned_foreign_key' do
+    context 'when run inside a transaction block' do
+      it 'raises an error' do
+        expect(migration).to receive(:transaction_open?).and_return(true)
+
+        expect do
+          migration.validate_partitioned_foreign_key(source_table_name, column_name, name: '_my_fk_name')
+        end.to raise_error(/can not be run inside a transaction/)
+      end
+    end
+
+    context 'when run outside a transaction block' do
+      before do
+        migration.add_concurrent_partitioned_foreign_key(
+          source_table_name,
+          target_table_name,
+          column: column_name,
+          name: foreign_key_name,
+          validate: false
+        )
+      end
+
+      it 'validates FK for each partition' do
+        expect(migration).to receive(:execute).with(/SET statement_timeout TO 0/).twice
+        expect(migration).to receive(:execute).with(/RESET statement_timeout/).twice
+        expect(migration).to receive(:execute)
+          .with(/ALTER TABLE #{partition1_name} VALIDATE CONSTRAINT #{foreign_key_name}/).ordered
+        expect(migration).to receive(:execute)
+          .with(/ALTER TABLE #{partition2_name} VALIDATE CONSTRAINT #{foreign_key_name}/).ordered
+
+        migration.validate_partitioned_foreign_key(source_table_name, column_name, name: foreign_key_name)
+      end
+    end
+  end
 end