diff --git a/pkg/controller/podautoscaler/horizontal.go b/pkg/controller/podautoscaler/horizontal.go
index 703161c2870..38b171f672c 100644
--- a/pkg/controller/podautoscaler/horizontal.go
+++ b/pkg/controller/podautoscaler/horizontal.go
@@ -50,6 +50,7 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/controller"
 	metricsclient "k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/controller/util/selectors"
 )
 
 var (
@@ -103,6 +104,10 @@ type HorizontalController struct {
 	scaleUpEventsLock   sync.RWMutex
 	scaleDownEvents     map[string][]timestampedScaleEvent
 	scaleDownEventsLock sync.RWMutex
+
+	// Storage of HPAs and their selectors.
+	hpaSelectors    *selectors.BiMultimap
+	hpaSelectorsMux sync.Mutex
 }
 
 // NewHorizontalController creates a new HorizontalController.
@@ -139,6 +144,7 @@ func NewHorizontalController(
 		scaleUpEventsLock:            sync.RWMutex{},
 		scaleDownEvents:              map[string][]timestampedScaleEvent{},
 		scaleDownEventsLock:          sync.RWMutex{},
+		hpaSelectors:                 selectors.NewBiMultimap(),
 	}
 
 	hpaInformer.Informer().AddEventHandlerWithResyncPeriod(
@@ -203,6 +209,15 @@ func (a *HorizontalController) enqueueHPA(obj interface{}) {
 	// request for the HPA in the queue then a new request is always dropped. Requests spend resync
 	// interval in queue so HPAs are processed every resync interval.
 	a.queue.AddRateLimited(key)
+
+	// Register HPA in the hpaSelectors map if it's not present yet. Attaching the Nothing selector
+	// that does not select objects. The actual selector is going to be updated
+	// when it's available during the autoscaler reconciliation.
+	a.hpaSelectorsMux.Lock()
+	defer a.hpaSelectorsMux.Unlock()
+	if hpaKey := selectors.Parse(key); !a.hpaSelectors.SelectorExists(hpaKey) {
+		a.hpaSelectors.PutSelector(hpaKey, labels.Nothing())
+	}
 }
 
 func (a *HorizontalController) deleteHPA(obj interface{}) {
@@ -214,6 +229,11 @@ func (a *HorizontalController) deleteHPA(obj interface{}) {
 
 	// TODO: could we leak if we fail to get the key?
 	a.queue.Forget(key)
+
+	// Remove HPA and attached selector.
+	a.hpaSelectorsMux.Lock()
+	defer a.hpaSelectorsMux.Unlock()
+	a.hpaSelectors.DeleteSelector(selectors.Parse(key))
 }
 
 func (a *HorizontalController) worker(ctx context.Context) {
@@ -254,19 +274,10 @@ func (a *HorizontalController) processNextWorkItem(ctx context.Context) bool {
 // all metrics computed.
 func (a *HorizontalController) computeReplicasForMetrics(ctx context.Context, hpa *autoscalingv2.HorizontalPodAutoscaler, scale *autoscalingv1.Scale,
 	metricSpecs []autoscalingv2.MetricSpec) (replicas int32, metric string, statuses []autoscalingv2.MetricStatus, timestamp time.Time, err error) {
-	if scale.Status.Selector == "" {
-		errMsg := "selector is required"
-		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "SelectorRequired", errMsg)
-		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", "the HPA target's scale is missing a selector")
-		return 0, "", nil, time.Time{}, fmt.Errorf(errMsg)
-	}
 
-	selector, err := labels.Parse(scale.Status.Selector)
+	selector, err := a.validateAndParseSelector(hpa, scale.Status.Selector)
 	if err != nil {
-		errMsg := fmt.Sprintf("couldn't convert selector into a corresponding internal selector object: %v", err)
-		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "InvalidSelector", errMsg)
-		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", errMsg)
-		return 0, "", nil, time.Time{}, fmt.Errorf(errMsg)
+		return 0, "", nil, time.Time{}, err
 	}
 
 	specReplicas := scale.Spec.Replicas
@@ -305,6 +316,80 @@ func (a *HorizontalController) computeReplicasForMetrics(ctx context.Context, hp
 	return replicas, metric, statuses, timestamp, nil
 }
 
+// hpasControllingPodsUnderSelector returns a list of keys of all HPAs that control a given list of pods.
+func (a *HorizontalController) hpasControllingPodsUnderSelector(pods []*v1.Pod) []selectors.Key {
+	a.hpaSelectorsMux.Lock()
+	defer a.hpaSelectorsMux.Unlock()
+
+	hpas := map[selectors.Key]struct{}{}
+	for _, p := range pods {
+		podKey := selectors.Key{Name: p.Name, Namespace: p.Namespace}
+		a.hpaSelectors.Put(podKey, p.Labels)
+
+		selectingHpas, ok := a.hpaSelectors.ReverseSelect(podKey)
+		if !ok {
+			continue
+		}
+		for _, hpa := range selectingHpas {
+			hpas[hpa] = struct{}{}
+		}
+	}
+	// Clean up all added pods.
+	a.hpaSelectors.KeepOnly([]selectors.Key{})
+
+	hpaList := []selectors.Key{}
+	for hpa := range hpas {
+		hpaList = append(hpaList, hpa)
+	}
+	return hpaList
+}
+
+// validateAndParseSelector verifies that:
+// - selector is not empty;
+// - selector format is valid;
+// - all pods by current selector are controlled by only one HPA.
+// Returns an error if the check has failed or the parsed selector if succeeded.
+// In case of an error the ScalingActive is set to false with the corresponding reason.
+func (a *HorizontalController) validateAndParseSelector(hpa *autoscalingv2.HorizontalPodAutoscaler, selector string) (labels.Selector, error) {
+	if selector == "" {
+		errMsg := "selector is required"
+		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "SelectorRequired", errMsg)
+		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", "the HPA target's scale is missing a selector")
+		return nil, fmt.Errorf(errMsg)
+	}
+
+	parsedSelector, err := labels.Parse(selector)
+	if err != nil {
+		errMsg := fmt.Sprintf("couldn't convert selector into a corresponding internal selector object: %v", err)
+		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "InvalidSelector", errMsg)
+		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "InvalidSelector", errMsg)
+		return nil, fmt.Errorf(errMsg)
+	}
+
+	hpaKey := selectors.Key{Name: hpa.Name, Namespace: hpa.Namespace}
+	a.hpaSelectorsMux.Lock()
+	if a.hpaSelectors.SelectorExists(hpaKey) {
+		// Update HPA selector only if the HPA was registered in enqueueHPA.
+		a.hpaSelectors.PutSelector(hpaKey, parsedSelector)
+	}
+	a.hpaSelectorsMux.Unlock()
+
+	pods, err := a.podLister.Pods(hpa.Namespace).List(parsedSelector)
+	if err != nil {
+		return nil, err
+	}
+
+	selectingHpas := a.hpasControllingPodsUnderSelector(pods)
+	if len(selectingHpas) > 1 {
+		errMsg := fmt.Sprintf("pods by selector %v are controlled by multiple HPAs: %v", selector, selectingHpas)
+		a.eventRecorder.Event(hpa, v1.EventTypeWarning, "AmbiguousSelector", errMsg)
+		setCondition(hpa, autoscalingv2.ScalingActive, v1.ConditionFalse, "AmbiguousSelector", errMsg)
+		return nil, fmt.Errorf(errMsg)
+	}
+
+	return parsedSelector, nil
+}
+
 // Computes the desired number of replicas for a specific hpa and metric specification,
 // returning the metric status and a proposed condition to be set on the HPA object.
 func (a *HorizontalController) computeReplicasForMetric(ctx context.Context, hpa *autoscalingv2.HorizontalPodAutoscaler, spec autoscalingv2.MetricSpec,
diff --git a/pkg/controller/podautoscaler/horizontal_test.go b/pkg/controller/podautoscaler/horizontal_test.go
index 7d7281e0adb..407e6fe0da0 100644
--- a/pkg/controller/podautoscaler/horizontal_test.go
+++ b/pkg/controller/podautoscaler/horizontal_test.go
@@ -43,6 +43,7 @@ import (
 	autoscalingapiv2 "k8s.io/kubernetes/pkg/apis/autoscaling/v2"
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/controller/util/selectors"
 	cmapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta2"
 	emapi "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
@@ -146,6 +147,7 @@ type testCase struct {
 	testScaleClient   *scalefake.FakeScaleClient
 
 	recommendations []timestampedRecommendation
+	hpaSelectors    *selectors.BiMultimap
 }
 
 // Needs to be called under a lock.
@@ -741,6 +743,9 @@ func (tc *testCase) setupController(t *testing.T) (*HorizontalController, inform
 	if tc.recommendations != nil {
 		hpaController.recommendations["test-namespace/test-hpa"] = tc.recommendations
 	}
+	if tc.hpaSelectors != nil {
+		hpaController.hpaSelectors = tc.hpaSelectors
+	}
 
 	return hpaController, informerFactory
 }
@@ -2387,6 +2392,112 @@ func TestConditionInvalidSelectorUnparsable(t *testing.T) {
 	tc.runTest(t)
 }
 
+func TestConditionNoAmbiguousSelectorWhenNoSelectorOverlapBetweenHPAs(t *testing.T) {
+	hpaSelectors := selectors.NewBiMultimap()
+	hpaSelectors.PutSelector(selectors.Key{Name: "test-hpa-2", Namespace: testNamespace}, labels.SelectorFromSet(labels.Set{"cheddar": "cheese"}))
+
+	tc := testCase{
+		minReplicas:             2,
+		maxReplicas:             6,
+		specReplicas:            3,
+		statusReplicas:          3,
+		expectedDesiredReplicas: 5,
+		CPUTarget:               30,
+		reportedLevels:          []uint64{300, 500, 700},
+		reportedCPURequests:     []resource.Quantity{resource.MustParse("1.0"), resource.MustParse("1.0"), resource.MustParse("1.0")},
+		useMetricsAPI:           true,
+		hpaSelectors:            hpaSelectors,
+	}
+	tc.runTest(t)
+}
+
+func TestConditionAmbiguousSelectorWhenFullSelectorOverlapBetweenHPAs(t *testing.T) {
+	hpaSelectors := selectors.NewBiMultimap()
+	hpaSelectors.PutSelector(selectors.Key{Name: "test-hpa-2", Namespace: testNamespace}, labels.SelectorFromSet(labels.Set{"name": podNamePrefix}))
+
+	tc := testCase{
+		minReplicas:             2,
+		maxReplicas:             6,
+		specReplicas:            3,
+		statusReplicas:          3,
+		expectedDesiredReplicas: 3,
+		CPUTarget:               30,
+		reportedLevels:          []uint64{300, 500, 700},
+		reportedCPURequests:     []resource.Quantity{resource.MustParse("1.0"), resource.MustParse("1.0"), resource.MustParse("1.0")},
+		useMetricsAPI:           true,
+		expectedConditions: []autoscalingv2.HorizontalPodAutoscalerCondition{
+			{
+				Type:   autoscalingv2.AbleToScale,
+				Status: v1.ConditionTrue,
+				Reason: "SucceededGetScale",
+			},
+			{
+				Type:   autoscalingv2.ScalingActive,
+				Status: v1.ConditionFalse,
+				Reason: "AmbiguousSelector",
+			},
+		},
+		hpaSelectors: hpaSelectors,
+	}
+	tc.runTest(t)
+}
+
+func TestConditionAmbiguousSelectorWhenPartialSelectorOverlapBetweenHPAs(t *testing.T) {
+	hpaSelectors := selectors.NewBiMultimap()
+	hpaSelectors.PutSelector(selectors.Key{Name: "test-hpa-2", Namespace: testNamespace}, labels.SelectorFromSet(labels.Set{"cheddar": "cheese"}))
+
+	tc := testCase{
+		minReplicas:             2,
+		maxReplicas:             6,
+		specReplicas:            3,
+		statusReplicas:          3,
+		expectedDesiredReplicas: 3,
+		CPUTarget:               30,
+		reportedLevels:          []uint64{300, 500, 700},
+		reportedCPURequests:     []resource.Quantity{resource.MustParse("1.0"), resource.MustParse("1.0"), resource.MustParse("1.0")},
+		useMetricsAPI:           true,
+		expectedConditions: []autoscalingv2.HorizontalPodAutoscalerCondition{
+			{
+				Type:   autoscalingv2.AbleToScale,
+				Status: v1.ConditionTrue,
+				Reason: "SucceededGetScale",
+			},
+			{
+				Type:   autoscalingv2.ScalingActive,
+				Status: v1.ConditionFalse,
+				Reason: "AmbiguousSelector",
+			},
+		},
+		hpaSelectors: hpaSelectors,
+	}
+
+	testClient, _, _, _, _ := tc.prepareTestClient(t)
+	tc.testClient = testClient
+
+	testClient.PrependReactor("list", "pods", func(action core.Action) (handled bool, ret runtime.Object, err error) {
+		tc.Lock()
+		defer tc.Unlock()
+
+		obj := &v1.PodList{}
+		for i := range tc.reportedCPURequests {
+			pod := v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("%s-%d", podNamePrefix, i),
+					Namespace: testNamespace,
+					Labels: map[string]string{
+						"name":    podNamePrefix, // selected by the original HPA
+						"cheddar": "cheese",      // selected by test-hpa-2
+					},
+				},
+			}
+			obj.Items = append(obj.Items, pod)
+		}
+		return true, obj, nil
+	})
+
+	tc.runTest(t)
+}
+
 func TestConditionFailedGetMetrics(t *testing.T) {
 	targetValue := resource.MustParse("15.0")
 	averageValue := resource.MustParse("15.0")
diff --git a/pkg/controller/util/selectors/bimultimap.go b/pkg/controller/util/selectors/bimultimap.go
new file mode 100644
index 00000000000..f81b9adc7a8
--- /dev/null
+++ b/pkg/controller/util/selectors/bimultimap.go
@@ -0,0 +1,380 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package selectors
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+
+	pkglabels "k8s.io/apimachinery/pkg/labels"
+)
+
+// BiMultimap is an efficient, bi-directional mapping of object
+// keys. Associations are created by putting keys with a selector.
+type BiMultimap struct {
+	mux sync.RWMutex
+
+	// Objects.
+	labeledObjects   map[Key]*labeledObject
+	selectingObjects map[Key]*selectingObject
+
+	// Associations.
+	labeledBySelecting map[selectorKey]*labeledObjects
+	selectingByLabeled map[labelsKey]*selectingObjects
+}
+
+// NewBiMultimap creates a map.
+func NewBiMultimap() *BiMultimap {
+	return &BiMultimap{
+		labeledObjects:     make(map[Key]*labeledObject),
+		selectingObjects:   make(map[Key]*selectingObject),
+		labeledBySelecting: make(map[selectorKey]*labeledObjects),
+		selectingByLabeled: make(map[labelsKey]*selectingObjects),
+	}
+}
+
+// Key is a tuple of name and namespace.
+type Key struct {
+	Name      string
+	Namespace string
+}
+
+// Parse turns a string in the format namespace/name into a Key.
+func Parse(s string) (key Key) {
+	ns := strings.SplitN(s, "/", 2)
+	if len(ns) == 2 {
+		key.Namespace = ns[0]
+		key.Name = ns[1]
+	} else {
+		key.Name = ns[0]
+	}
+	return key
+}
+
+func (k Key) String() string {
+	return fmt.Sprintf("%v/%v", k.Namespace, k.Name)
+}
+
+type selectorKey struct {
+	key       string
+	namespace string
+}
+
+type selectingObject struct {
+	key      Key
+	selector pkglabels.Selector
+	// selectorKey is a stable serialization of selector for
+	// association caching.
+	selectorKey selectorKey
+}
+
+type selectingObjects struct {
+	objects  map[Key]*selectingObject
+	refCount int
+}
+
+type labelsKey struct {
+	key       string
+	namespace string
+}
+
+type labeledObject struct {
+	key    Key
+	labels map[string]string
+	// labelsKey is a stable serialization of labels for association
+	// caching.
+	labelsKey labelsKey
+}
+
+type labeledObjects struct {
+	objects  map[Key]*labeledObject
+	refCount int
+}
+
+// Put inserts or updates an object and the incoming associations
+// based on the object labels.
+func (m *BiMultimap) Put(key Key, labels map[string]string) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	labelsKey := labelsKey{
+		key:       pkglabels.Set(labels).String(),
+		namespace: key.Namespace,
+	}
+	if l, ok := m.labeledObjects[key]; ok {
+		// Update labeled object.
+		if labelsKey == l.labelsKey {
+			// No change to labels.
+			return
+		}
+		// Delete before readding.
+		m.delete(key)
+	}
+	// Add labeled object.
+	labels = copyLabels(labels)
+	labeledObject := &labeledObject{
+		key:       key,
+		labels:    labels,
+		labelsKey: labelsKey,
+	}
+	m.labeledObjects[key] = labeledObject
+	// Add associations.
+	if _, ok := m.selectingByLabeled[labelsKey]; !ok {
+		// Cache miss. Scan selecting objects.
+		selecting := &selectingObjects{
+			objects: make(map[Key]*selectingObject),
+		}
+		set := pkglabels.Set(labels)
+		for _, s := range m.selectingObjects {
+			if s.key.Namespace != key.Namespace {
+				continue
+			}
+			if s.selector.Matches(set) {
+				selecting.objects[s.key] = s
+			}
+		}
+		// Associate selecting with labeled.
+		m.selectingByLabeled[labelsKey] = selecting
+	}
+	selecting := m.selectingByLabeled[labelsKey]
+	selecting.refCount += 1
+	for _, sObject := range selecting.objects {
+		// Associate labeled with selecting.
+		labeled := m.labeledBySelecting[sObject.selectorKey]
+		labeled.objects[labeledObject.key] = labeledObject
+	}
+}
+
+// Delete removes a labeled object and incoming associations.
+func (m *BiMultimap) Delete(key Key) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	m.delete(key)
+}
+
+func (m *BiMultimap) delete(key Key) {
+	if _, ok := m.labeledObjects[key]; !ok {
+		// Does not exist.
+		return
+	}
+	labeledObject := m.labeledObjects[key]
+	labelsKey := labeledObject.labelsKey
+	defer delete(m.labeledObjects, key)
+	if _, ok := m.selectingByLabeled[labelsKey]; !ok {
+		// No associations.
+		return
+	}
+	// Remove associations.
+	for _, selectingObject := range m.selectingByLabeled[labelsKey].objects {
+		selectorKey := selectingObject.selectorKey
+		// Delete selectingObject to labeledObject association.
+		delete(m.labeledBySelecting[selectorKey].objects, key)
+	}
+	m.selectingByLabeled[labelsKey].refCount -= 1
+	// Garbage collect labeledObject to selectingObject associations.
+	if m.selectingByLabeled[labelsKey].refCount == 0 {
+		delete(m.selectingByLabeled, labelsKey)
+	}
+}
+
+// Exists returns true if the labeled object is present in the map.
+func (m *BiMultimap) Exists(key Key) bool {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	_, exists := m.labeledObjects[key]
+	return exists
+}
+
+// PutSelector inserts or updates an object with a selector. Associations
+// are created or updated based on the selector.
+func (m *BiMultimap) PutSelector(key Key, selector pkglabels.Selector) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	selectorKey := selectorKey{
+		key:       selector.String(),
+		namespace: key.Namespace,
+	}
+	if s, ok := m.selectingObjects[key]; ok {
+		// Update selecting object.
+		if selectorKey == s.selectorKey {
+			// No change to selector.
+			return
+		}
+		// Delete before readding.
+		m.deleteSelector(key)
+	}
+	// Add selecting object.
+	selectingObject := &selectingObject{
+		key:         key,
+		selector:    selector,
+		selectorKey: selectorKey,
+	}
+	m.selectingObjects[key] = selectingObject
+	// Add associations.
+	if _, ok := m.labeledBySelecting[selectorKey]; !ok {
+		// Cache miss. Scan labeled objects.
+		labeled := &labeledObjects{
+			objects: make(map[Key]*labeledObject),
+		}
+		for _, l := range m.labeledObjects {
+			if l.key.Namespace != key.Namespace {
+				continue
+			}
+			set := pkglabels.Set(l.labels)
+			if selector.Matches(set) {
+				labeled.objects[l.key] = l
+			}
+		}
+		// Associate labeled with selecting.
+		m.labeledBySelecting[selectorKey] = labeled
+	}
+	labeled := m.labeledBySelecting[selectorKey]
+	labeled.refCount += 1
+	for _, labeledObject := range labeled.objects {
+		// Associate selecting with labeled.
+		selecting := m.selectingByLabeled[labeledObject.labelsKey]
+		selecting.objects[selectingObject.key] = selectingObject
+	}
+}
+
+// DeleteSelector deletes a selecting object and associations created by its
+// selector.
+func (m *BiMultimap) DeleteSelector(key Key) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	m.deleteSelector(key)
+}
+
+func (m *BiMultimap) deleteSelector(key Key) {
+	if _, ok := m.selectingObjects[key]; !ok {
+		// Does not exist.
+		return
+	}
+	selectingObject := m.selectingObjects[key]
+	selectorKey := selectingObject.selectorKey
+	defer delete(m.selectingObjects, key)
+	if _, ok := m.labeledBySelecting[selectorKey]; !ok {
+		// No associations.
+		return
+	}
+	// Remove associations.
+	for _, labeledObject := range m.labeledBySelecting[selectorKey].objects {
+		labelsKey := labeledObject.labelsKey
+		// Delete labeledObject to selectingObject association.
+		delete(m.selectingByLabeled[labelsKey].objects, key)
+	}
+	m.labeledBySelecting[selectorKey].refCount -= 1
+	// Garbage collect selectingObjects to labeledObject associations.
+	if m.labeledBySelecting[selectorKey].refCount == 0 {
+		delete(m.labeledBySelecting, selectorKey)
+	}
+}
+
+// SelectorExists returns true if the selecting object is present in the map.
+func (m *BiMultimap) SelectorExists(key Key) bool {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	_, exists := m.selectingObjects[key]
+	return exists
+}
+
+// KeepOnly retains only the specified labeled objects and deletes the
+// rest. Like calling Delete for all keys not specified.
+func (m *BiMultimap) KeepOnly(keys []Key) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	keyMap := make(map[Key]bool)
+	for _, k := range keys {
+		keyMap[k] = true
+	}
+	for k := range m.labeledObjects {
+		if !keyMap[k] {
+			m.delete(k)
+		}
+	}
+}
+
+// KeepOnlySelectors retains only the specified selecting objects and
+// deletes the rest. Like calling DeleteSelector for all keys not
+// specified.
+func (m *BiMultimap) KeepOnlySelectors(keys []Key) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	keyMap := make(map[Key]bool)
+	for _, k := range keys {
+		keyMap[k] = true
+	}
+	for k := range m.selectingObjects {
+		if !keyMap[k] {
+			m.deleteSelector(k)
+		}
+	}
+}
+
+// Select finds objects associated with a selecting object. If the
+// given key was found in the map `ok` will be true. Otherwise false.
+func (m *BiMultimap) Select(key Key) (keys []Key, ok bool) {
+	m.mux.RLock()
+	defer m.mux.RUnlock()
+
+	selectingObject, ok := m.selectingObjects[key]
+	if !ok {
+		// Does not exist.
+		return nil, false
+	}
+	keys = make([]Key, 0)
+	if labeled, ok := m.labeledBySelecting[selectingObject.selectorKey]; ok {
+		for _, labeledObject := range labeled.objects {
+			keys = append(keys, labeledObject.key)
+		}
+	}
+	return keys, true
+}
+
+// ReverseSelect finds objects selecting the given object. If the
+// given key was found in the map `ok` will be true. Otherwise false.
+func (m *BiMultimap) ReverseSelect(key Key) (keys []Key, ok bool) {
+	m.mux.RLock()
+	defer m.mux.RUnlock()
+
+	labeledObject, ok := m.labeledObjects[key]
+	if !ok {
+		// Does not exist.
+		return []Key{}, false
+	}
+	keys = make([]Key, 0)
+	if selecting, ok := m.selectingByLabeled[labeledObject.labelsKey]; ok {
+		for _, selectingObject := range selecting.objects {
+			keys = append(keys, selectingObject.key)
+		}
+	}
+	return keys, true
+}
+
+func copyLabels(labels map[string]string) map[string]string {
+	l := make(map[string]string)
+	for k, v := range labels {
+		l[k] = v
+	}
+	return l
+}
diff --git a/pkg/controller/util/selectors/bimultimap_test.go b/pkg/controller/util/selectors/bimultimap_test.go
new file mode 100644
index 00000000000..5477d40cc75
--- /dev/null
+++ b/pkg/controller/util/selectors/bimultimap_test.go
@@ -0,0 +1,641 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package selectors
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	pkglabels "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+)
+
+func TestAssociations(t *testing.T) {
+	cases := []struct {
+		name                string
+		ops                 []operation
+		want                []expectation
+		testAllPermutations bool
+	}{{
+		name: "single association",
+		ops: []operation{
+			putSelectingObject(key("hpa"), selector("a", "1")),
+			putLabeledObject(key("pod"), labels("a", "1")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa"), key("pod")),
+			reverseSelect(key("pod"), key("hpa")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "multiple associations from a selecting object",
+		ops: []operation{
+			putSelectingObject(key("hpa"), selector("a", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			putLabeledObject(key("pod-2"), labels("a", "1")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa"), key("pod-1"), key("pod-2")),
+			reverseSelect(key("pod-1"), key("hpa")),
+			reverseSelect(key("pod-2"), key("hpa")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "multiple associations to a labeled object",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putSelectingObject(key("hpa-2"), selector("a", "1")),
+			putLabeledObject(key("pod"), labels("a", "1")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod")),
+			forwardSelect(key("hpa-2"), key("pod")),
+			reverseSelect(key("pod"), key("hpa-1"), key("hpa-2")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "disjoint association sets",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putSelectingObject(key("hpa-2"), selector("a", "2")),
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			putLabeledObject(key("pod-2"), labels("a", "2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod-1")),
+			forwardSelect(key("hpa-2"), key("pod-2")),
+			reverseSelect(key("pod-1"), key("hpa-1")),
+			reverseSelect(key("pod-2"), key("hpa-2")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "separate label cache paths",
+		ops: []operation{
+			putSelectingObject(key("hpa"), selector("a", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "1", "b", "2")),
+			putLabeledObject(key("pod-2"), labels("a", "1", "b", "3")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa"), key("pod-1"), key("pod-2")),
+			reverseSelect(key("pod-1"), key("hpa")),
+			reverseSelect(key("pod-2"), key("hpa")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "separate selector cache paths",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putSelectingObject(key("hpa-2"), selector("b", "2")),
+			putLabeledObject(key("pod"), labels("a", "1", "b", "2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod")),
+			forwardSelect(key("hpa-2"), key("pod")),
+			reverseSelect(key("pod"), key("hpa-1"), key("hpa-2")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "selection in different namespaces",
+		ops: []operation{
+			putLabeledObject(key("pod-1", "namespace-1"), labels("a", "1")),
+			putLabeledObject(key("pod-1", "namespace-2"), labels("a", "1")),
+			putSelectingObject(key("hpa-1", "namespace-2"), selector("a", "1")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1", "namespace-1")), // selects nothing
+			forwardSelect(key("hpa-1", "namespace-2"), key("pod-1", "namespace-2")),
+			reverseSelect(key("pod-1", "namespace-1")), // selects nothing
+			reverseSelect(key("pod-1", "namespace-2"), key("hpa-1", "namespace-2")),
+		},
+		testAllPermutations: true,
+	}, {
+		name: "update labeled objects",
+		ops: []operation{
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			putSelectingObject(key("hpa-1"), selector("a", "2")),
+			putLabeledObject(key("pod-1"), labels("a", "2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod-1")),
+			reverseSelect(key("pod-1"), key("hpa-1")),
+		},
+	}, {
+		name: "update selecting objects",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "2")),
+			putSelectingObject(key("hpa-1"), selector("a", "2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod-1")),
+			reverseSelect(key("pod-1"), key("hpa-1")),
+		},
+	}, {
+		name: "keep only labeled objects",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			putLabeledObject(key("pod-2"), labels("a", "1")),
+			putLabeledObject(key("pod-3"), labels("a", "1")),
+			keepOnly(key("pod-1"), key("pod-2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod-1"), key("pod-2")),
+			reverseSelect(key("pod-1"), key("hpa-1")),
+			reverseSelect(key("pod-2"), key("hpa-1")),
+		},
+	}, {
+		name: "keep only selecting objects",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putSelectingObject(key("hpa-2"), selector("a", "1")),
+			putSelectingObject(key("hpa-3"), selector("a", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			keepOnlySelectors(key("hpa-1"), key("hpa-2")),
+		},
+		want: []expectation{
+			forwardSelect(key("hpa-1"), key("pod-1")),
+			forwardSelect(key("hpa-2"), key("pod-1")),
+			reverseSelect(key("pod-1"), key("hpa-1"), key("hpa-2")),
+		},
+	}, {
+		name: "put multiple associations and delete all",
+		ops: []operation{
+			putSelectingObject(key("hpa-1"), selector("a", "1")),
+			putSelectingObject(key("hpa-2"), selector("a", "1")),
+			putSelectingObject(key("hpa-3"), selector("a", "2")),
+			putSelectingObject(key("hpa-4"), selector("b", "1")),
+			putLabeledObject(key("pod-1"), labels("a", "1")),
+			putLabeledObject(key("pod-2"), labels("a", "2")),
+			putLabeledObject(key("pod-3"), labels("a", "1", "b", "1")),
+			putLabeledObject(key("pod-4"), labels("a", "2", "b", "1")),
+			putLabeledObject(key("pod-5"), labels("b", "1")),
+			putLabeledObject(key("pod-6"), labels("b", "2")),
+			deleteSelecting(key("hpa-1")),
+			deleteSelecting(key("hpa-2")),
+			deleteSelecting(key("hpa-3")),
+			deleteSelecting(key("hpa-4")),
+			deleteLabeled(key("pod-1")),
+			deleteLabeled(key("pod-2")),
+			deleteLabeled(key("pod-3")),
+			deleteLabeled(key("pod-4")),
+			deleteLabeled(key("pod-5")),
+			deleteLabeled(key("pod-6")),
+		},
+		want: []expectation{
+			emptyMap,
+		},
+	}, {
+		name: "fuzz testing",
+		ops: []operation{
+			randomOperations(10000),
+			deleteAll,
+		},
+		want: []expectation{
+			emptyMap,
+		},
+	}}
+
+	for _, tc := range cases {
+		var permutations [][]int
+		if tc.testAllPermutations {
+			// Run test case with all permutations of operations.
+			permutations = indexPermutations(len(tc.ops))
+		} else {
+			// Unless test is order dependent (e.g. includes
+			// deletes) or just too big.
+			var p []int
+			for i := 0; i < len(tc.ops); i++ {
+				p = append(p, i)
+			}
+			permutations = [][]int{p}
+		}
+		for _, permutation := range permutations {
+			name := tc.name + fmt.Sprintf(" permutation %v", permutation)
+			t.Run(name, func(t *testing.T) {
+				multimap := NewBiMultimap()
+				for i := range permutation {
+					tc.ops[i](multimap)
+					// Run consistency check after every operation.
+					err := consistencyCheck(multimap)
+					if err != nil {
+						t.Fatalf(err.Error())
+					}
+				}
+				for _, expect := range tc.want {
+					err := expect(multimap)
+					if err != nil {
+						t.Errorf("%v %v", tc.name, err)
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestEfficientAssociation(t *testing.T) {
+	useOnceSelector := useOnce(selector("a", "1"))
+	m := NewBiMultimap()
+	m.PutSelector(key("hpa-1"), useOnceSelector)
+	m.Put(key("pod-1"), labels("a", "1"))
+
+	// Selector is used only during full scan. Second Put will use
+	// cached association or explode.
+	m.Put(key("pod-2"), labels("a", "1"))
+
+	err := forwardSelect(key("hpa-1"), key("pod-1"), key("pod-2"))(m)
+	if err != nil {
+		t.Errorf(err.Error())
+	}
+}
+
+func TestUseOnceSelector(t *testing.T) {
+	useOnceSelector := useOnce(selector("a", "1"))
+	labels := pkglabels.Set(labels("a", "1"))
+
+	// Use once.
+	useOnceSelector.Matches(labels)
+	// Use twice.
+	defer func() {
+		if r := recover(); r == nil {
+			t.Errorf("Expected panic when using selector twice.")
+		}
+	}()
+	useOnceSelector.Matches(labels)
+}
+
+func TestObjectsExist(t *testing.T) {
+	m := NewBiMultimap()
+
+	// Nothing exists in the empty map.
+	assert.False(t, m.Exists(key("pod-1")))
+	assert.False(t, m.SelectorExists(key("hpa-1")))
+
+	// Adding entries.
+	m.PutSelector(key("hpa-1"), useOnce(selector("a", "1")))
+	m.Put(key("pod-1"), labels("a", "1"))
+
+	// Entries exist.
+	assert.True(t, m.Exists(key("pod-1")))
+	assert.True(t, m.SelectorExists(key("hpa-1")))
+
+	// Removing the entries.
+	m.DeleteSelector(key("hpa-1"))
+	m.Delete(key("pod-1"))
+
+	// They don't exist anymore.
+	assert.False(t, m.Exists(key("pod-1")))
+	assert.False(t, m.SelectorExists(key("hpa-1")))
+}
+
+type useOnceSelector struct {
+	used     bool
+	selector pkglabels.Selector
+}
+
+func useOnce(s pkglabels.Selector) pkglabels.Selector {
+	return &useOnceSelector{
+		selector: s,
+	}
+}
+
+func (u *useOnceSelector) Matches(l pkglabels.Labels) bool {
+	if u.used {
+		panic("useOnceSelector used more than once")
+	}
+	u.used = true
+	return u.selector.Matches(l)
+}
+
+func (u *useOnceSelector) Empty() bool {
+	return u.selector.Empty()
+}
+
+func (u *useOnceSelector) String() string {
+	return u.selector.String()
+}
+
+func (u *useOnceSelector) Add(r ...pkglabels.Requirement) pkglabels.Selector {
+	u.selector = u.selector.Add(r...)
+	return u
+}
+
+func (u *useOnceSelector) Requirements() (pkglabels.Requirements, bool) {
+	return u.selector.Requirements()
+}
+
+func (u *useOnceSelector) DeepCopySelector() pkglabels.Selector {
+	u.selector = u.selector.DeepCopySelector()
+	return u
+}
+
+func (u *useOnceSelector) RequiresExactMatch(label string) (value string, found bool) {
+	v, f := u.selector.RequiresExactMatch(label)
+	return v, f
+}
+
+func indexPermutations(size int) [][]int {
+	var permute func([]int, []int) [][]int
+	permute = func(placed, remaining []int) (permutations [][]int) {
+		if len(remaining) == 0 {
+			return [][]int{placed}
+		}
+		for i, v := range remaining {
+			r := append([]int(nil), remaining...) // copy remaining
+			r = append(r[:i], r[i+1:]...)         // delete placed index
+			p := permute(append(placed, v), r)    // place index and permute
+			permutations = append(permutations, p...)
+		}
+		return
+	}
+	var remaining []int
+	for i := 0; i < size; i++ {
+		remaining = append(remaining, i)
+	}
+	return permute(nil, remaining)
+}
+
+type operation func(*BiMultimap)
+
+func putLabeledObject(key Key, labels map[string]string) operation {
+	return func(m *BiMultimap) {
+		m.Put(key, labels)
+	}
+}
+
+func putSelectingObject(key Key, selector pkglabels.Selector) operation {
+	return func(m *BiMultimap) {
+		m.PutSelector(key, selector)
+	}
+}
+
+func deleteLabeled(key Key) operation {
+	return func(m *BiMultimap) {
+		m.Delete(key)
+	}
+}
+
+func deleteSelecting(key Key) operation {
+	return func(m *BiMultimap) {
+		m.DeleteSelector(key)
+	}
+}
+
+func deleteAll(m *BiMultimap) {
+	for key := range m.labeledObjects {
+		m.Delete(key)
+	}
+	for key := range m.selectingObjects {
+		m.DeleteSelector(key)
+	}
+}
+
+func keepOnly(keys ...Key) operation {
+	return func(m *BiMultimap) {
+		m.KeepOnly(keys)
+	}
+}
+
+func keepOnlySelectors(keys ...Key) operation {
+	return func(m *BiMultimap) {
+		m.KeepOnlySelectors(keys)
+	}
+}
+
+func randomOperations(times int) operation {
+	pods := []Key{
+		key("pod-1"),
+		key("pod-2"),
+		key("pod-3"),
+		key("pod-4"),
+		key("pod-5"),
+		key("pod-6"),
+	}
+	randomPod := func() Key {
+		return pods[rand.Intn(len(pods))]
+	}
+	labels := []map[string]string{
+		labels("a", "1"),
+		labels("a", "2"),
+		labels("b", "1"),
+		labels("b", "2"),
+		labels("a", "1", "b", "1"),
+		labels("a", "2", "b", "2"),
+		labels("a", "3"),
+		labels("c", "1"),
+	}
+	randomLabels := func() map[string]string {
+		return labels[rand.Intn(len(labels))]
+	}
+	hpas := []Key{
+		key("hpa-1"),
+		key("hpa-2"),
+		key("hpa-3"),
+	}
+	randomHpa := func() Key {
+		return hpas[rand.Intn(len(hpas))]
+	}
+	selectors := []pkglabels.Selector{
+		selector("a", "1"),
+		selector("b", "1"),
+		selector("a", "1", "b", "1"),
+		selector("c", "2"),
+	}
+	randomSelector := func() pkglabels.Selector {
+		return selectors[rand.Intn(len(selectors))]
+	}
+	randomOp := func(m *BiMultimap) {
+		switch rand.Intn(4) {
+		case 0:
+			m.Put(randomPod(), randomLabels())
+		case 1:
+			m.PutSelector(randomHpa(), randomSelector())
+		case 2:
+			m.Delete(randomPod())
+		case 3:
+			m.DeleteSelector(randomHpa())
+		}
+	}
+	return func(m *BiMultimap) {
+		for i := 0; i < times; i++ {
+			randomOp(m)
+		}
+	}
+}
+
+type expectation func(*BiMultimap) error
+
+func forwardSelect(key Key, want ...Key) expectation {
+	return func(m *BiMultimap) error {
+		got, _ := m.Select(key)
+		if !unorderedEqual(want, got) {
+			return fmt.Errorf("forward select %v wanted %v. got %v.", key, want, got)
+		}
+		return nil
+	}
+}
+
+func reverseSelect(key Key, want ...Key) expectation {
+	return func(m *BiMultimap) error {
+		got, _ := m.ReverseSelect(key)
+		if !unorderedEqual(want, got) {
+			return fmt.Errorf("reverse select %v wanted %v. got %v.", key, want, got)
+		}
+		return nil
+	}
+}
+
+func emptyMap(m *BiMultimap) error {
+	if len(m.labeledObjects) != 0 {
+		return fmt.Errorf("Found %v labeledObjects. Wanted none.", len(m.labeledObjects))
+	}
+	if len(m.selectingObjects) != 0 {
+		return fmt.Errorf("Found %v selectingObjects. Wanted none.", len(m.selectingObjects))
+	}
+	if len(m.labeledBySelecting) != 0 {
+		return fmt.Errorf("Found %v cached labeledBySelecting associations. Wanted none.", len(m.labeledBySelecting))
+	}
+	if len(m.selectingByLabeled) != 0 {
+		return fmt.Errorf("Found %v cached selectingByLabeled associations. Wanted none.", len(m.selectingByLabeled))
+	}
+	return nil
+}
+
+func consistencyCheck(m *BiMultimap) error {
+	emptyKey := Key{}
+	emptyLabelsKey := labelsKey{}
+	emptySelectorKey := selectorKey{}
+	for k, v := range m.labeledObjects {
+		if v == nil {
+			return fmt.Errorf("Found nil labeled object for key %q", k)
+		}
+		if k == emptyKey {
+			return fmt.Errorf("Found empty key for labeled object %+v", v)
+		}
+	}
+	for k, v := range m.selectingObjects {
+		if v == nil {
+			return fmt.Errorf("Found nil selecting object for key %q", k)
+		}
+		if k == emptyKey {
+			return fmt.Errorf("Found empty key for selecting object %+v", v)
+		}
+	}
+	for k, v := range m.labeledBySelecting {
+		if v == nil {
+			return fmt.Errorf("Found nil labeledBySelecting entry for key %q", k)
+		}
+		if k == emptySelectorKey {
+			return fmt.Errorf("Found empty key for labeledBySelecting object %+v", v)
+		}
+		for k2, v2 := range v.objects {
+			if v2 == nil {
+				return fmt.Errorf("Found nil object in labeledBySelecting under keys %q and %q", k, k2)
+			}
+			if k2 == emptyKey {
+				return fmt.Errorf("Found empty key for object in labeledBySelecting under key %+v", k)
+			}
+		}
+		if v.refCount < 1 {
+			return fmt.Errorf("Found labeledBySelecting entry with no references (orphaned) under key %q", k)
+		}
+	}
+	for k, v := range m.selectingByLabeled {
+		if v == nil {
+			return fmt.Errorf("Found nil selectingByLabeled entry for key %q", k)
+		}
+		if k == emptyLabelsKey {
+			return fmt.Errorf("Found empty key for selectingByLabeled object %+v", v)
+		}
+		for k2, v2 := range v.objects {
+			if v2 == nil {
+				return fmt.Errorf("Found nil object in selectingByLabeled under keys %q and %q", k, k2)
+			}
+			if k2 == emptyKey {
+				return fmt.Errorf("Found empty key for object in selectingByLabeled under key %+v", k)
+			}
+		}
+		if v.refCount < 1 {
+			return fmt.Errorf("Found selectingByLabeled entry with no references (orphaned) under key %q", k)
+		}
+	}
+	return nil
+}
+
+func key(s string, ss ...string) Key {
+	if len(ss) > 1 {
+		panic("Key requires 1 or 2 parts.")
+	}
+	k := Key{
+		Name: s,
+	}
+	if len(ss) >= 1 {
+		k.Namespace = ss[0]
+	}
+	return k
+}
+
+func labels(ls ...string) map[string]string {
+	if len(ls)%2 != 0 {
+		panic("labels requires pairs of strings.")
+	}
+	ss := make(map[string]string)
+	for i := 0; i < len(ls); i += 2 {
+		ss[ls[i]] = ls[i+1]
+	}
+	return ss
+}
+
+func selector(ss ...string) pkglabels.Selector {
+	if len(ss)%2 != 0 {
+		panic("selector requires pairs of strings.")
+	}
+	s := pkglabels.NewSelector()
+	for i := 0; i < len(ss); i += 2 {
+		r, err := pkglabels.NewRequirement(ss[i], selection.Equals, []string{ss[i+1]})
+		if err != nil {
+			panic(err)
+		}
+		s = s.Add(*r)
+	}
+	return s
+}
+
+func unorderedEqual(as, bs []Key) bool {
+	if len(as) != len(bs) {
+		return false
+	}
+	aMap := make(map[Key]int)
+	for _, a := range as {
+		aMap[a] += 1
+	}
+	bMap := make(map[Key]int)
+	for _, b := range bs {
+		bMap[b] += 1
+	}
+	if len(aMap) != len(bMap) {
+		return false
+	}
+	for a, count := range aMap {
+		if bMap[a] != count {
+			return false
+		}
+	}
+	return true
+}