Repository: kubernetes-sigs/kueue

Issue:
Flavor Fungibility Prefers Fit over NoBorrow
**What would you like to be added**:
Flavor assigner implicitly prefers (`Fit`, `Borrow`) over (`Preempt`, `NoBorrow`)

Customers may want to express that it is more important for their workload to fit within nominal quota (and therefore be un-preemptable) than upgrading from `Preempt` to `Fit`

**Why is this needed**:
- Suppose two flavors, with capacity for 1 workload each
  - Flavor1: capacity provided by cq-2
  - Flavor2: capacity provided by cq-1
- Suppose `FlavorFungibility.whenCanBorrow=TryNextFlavor`
- Suppose two CQs, both with `ReclaimWithinCohort=Any`, `WithinClusterQueue=LowerPriority`

Then we create 3 workloads in order:
1) cq-1: a low-priority workload schedules into into `Flavor2`, since (`Fit`, `NoBorrow`)
2) cq-1: a high-priority workload schedules into Flavor1, as (`Fit`, `Borrow`) > (`Preempt`, `NoBorrow`)
3) cq-2: a low-priority workload preempts the high-priority workload from cq-1 in Flavor1, as it is reclaiming its quota

The user might expect that, during step 2, the high-priority workload preempts the low-priority workload, so that it can run within nominal quota and be safe from preemptions in the future.

The scope of this task is to propose an API which would allow user to express this preference, and to implement this change in the code

**Completion requirements**:

This enhancement requires the following artifacts:

- [x] Design doc
- [x] API change
- [x] Docs update

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@tenzen-y: @gabesaba Do you assume the flavor assigner prioritizes the `whenCanPreempt` rather than `whenCanBorrow`?
 In that case, do you expect adding a new API to specify which mode (`whenCanPreempt`) and (`whenCanborrow`) should be evaluated first?
@mimowo: > The user might expect that, during step 2, the high-priority workload preempts the low-priority workload, so that it can run within nominal quota and be safe from preemptions in the future.

Isn't this achievable already with the current API:
```yaml
flavorFungibility:
  whenCanBorrow: TryNextFlavor
  whenCanPreempt: Preempt
```
or it works, but there are other drawbacks?@gabesaba: > > The user might expect that, during step 2, the high-priority workload preempts the low-priority workload, so that it can run within nominal quota and be safe from preemptions in the future.
> 
> Isn't this achievable already with the current API:
> 
> flavorFungibility:
>   whenCanBorrow: TryNextFlavor
>   whenCanPreempt: Preempt
> or it works, but there are other drawbacks?

With the current API, no. It tries the 2nd flavor, and even if the 2nd flavor doesn't require borrowing, it decides 1st flavor is better since it `Fit`, while the 2nd flavor is `Preempt`. Hence the issue description: Flavor assigner implicitly prefers (Fit, Borrow) over (Preempt, NoBorrow)@gabesaba: > [@gabesaba](https://github.com/gabesaba) Do you assume the flavor assigner prioritizes the `whenCanPreempt` rather than `whenCanBorrow`? In that case, do you expect adding a new API to specify which mode (`whenCanPreempt`) and (`whenCanborrow`) should be evaluated first?

As stated in my response to @mimowo's comment: even if only whenCanBorrow=TryNextFlavor, we implicitly prefer `Fit` over `Preempt`, regardless of borrowing status. I think that the issue is a little more general than just the order of evaluation@tenzen-y: > > [@gabesaba](https://github.com/gabesaba) Do you assume the flavor assigner prioritizes the `whenCanPreempt` rather than `whenCanBorrow`? In that case, do you expect adding a new API to specify which mode (`whenCanPreempt`) and (`whenCanborrow`) should be evaluated first?
> 
> As stated in my response to [@mimowo](https://github.com/mimowo)'s comment: even if only whenCanBorrow=TryNextFlavor, we implicitly prefer `Fit` over `Preempt`, regardless of borrowing status. I think that the issue is a little more general than just the order of evaluation

Thank you for describing that. So, I guess that we might need to support new flavor fungibility (preference selection ways).
@pajakd: /assign

Solution patch:
diff --git a/keps/582-preempt-based-on-flavor-order/README.md b/keps/582-preempt-based-on-flavor-order/README.md
index 6dda5e4bf..153cca91d 100644
--- a/keps/582-preempt-based-on-flavor-order/README.md
+++ b/keps/582-preempt-based-on-flavor-order/README.md
@@ -24,6 +24,7 @@ tags, and then generate with `hack/update-toc.sh`.
 - [Proposal](#proposal)
   - [User Stories (Optional)](#user-stories-optional)
     - [Story 1](#story-1)
+    - [Story 2](#story-2)
   - [Notes/Constraints/Caveats (Optional)](#notesconstraintscaveats-optional)
   - [Risks and Mitigations](#risks-and-mitigations)
 - [Design Details](#design-details)
@@ -122,7 +123,13 @@ in my cluster. In this case I prefer my high priority jobs not running on spot
 instances. If high priority jobs can preempt jobs in standard instances before trying spot instances,
 stability can be achieved.
 
-My use case can be supported by setting `.Spec.FlavorFungibility.WhenCanPreempt` to `Preempt`  in the ClusterQueue's spec.
+My use case can be supported by setting `.Spec.FlavorFungibility.WhenCanPreempt` to `Preempt` in the ClusterQueue's spec.
+
+#### Story 2
+
+As an admin of system managed by Kueue I would like to minimize the risk that admitted workloads get preempted soon after admission. Since every borrowing workload is a preemption (reclaim) candidate, to minimize the risk, I would like to prioritize selecting flavors which are preempting rather than borrowing.
+
+My use case can be supported by setting `Spec.FlavorFungibility.WhenCanPreempt: Preempt`.
 
 ### Notes/Constraints/Caveats (Optional)
 
@@ -164,15 +171,33 @@ For each type of resource in each podSet, Kueue will traverse all resource group
 
 ```
 const (
-	Borrow FlavorFungibilityPolicy = "Borrow"
-	Preempt  FlavorFungibilityPolicy = "Preempt"
+  Borrow FlavorFungibilityPolicy = "Borrow"
+  Preempt  FlavorFungibilityPolicy = "Preempt"
   TryNextFlavor FlavorFungibilityPolicy = "TryNextFlavor"
 )
 
 type FlavorFungibility struct {
-  // +kubebuilder:validation:Enum="Borrow,TryNextFlavor"
+  // whenCanBorrow determines whether a workload should try the next flavor
+  // or stop the search. The possible values are:
+  //
+  // - `Borrow` (default): stop searching and use the best flavor found so far
+  //   according to the selection strategy
+  // - `TryNextFlavor`: try next flavor even if the current
+  //   flavor has enough resources to borrow.
+  //
+  // +kubebuilder:validation:Enum={Borrow,TryNextFlavor}
+  // +kubebuilder:default="Borrow"
   WhenCanBorrow FlavorFungibilityPolicy  `json:"whenCanBorrow"`
-  // +kubebuilder:validation:Enum="Preempt,TryNextFlavor"
+  // whenCanPreempt determines whether a workload should try the next flavor
+  // or stop the search. The possible values are:
+  //
+  // - `Preempt`: stop searching and use the best flavor found so far 
+  //   according to the selection strategy
+  // - `TryNextFlavor` (default): try next flavor even if there are enough
+  //   candidates for preemption in the current flavor.
+  //
+  // +kubebuilder:validation:Enum={Preempt,TryNextFlavor}
+  // +kubebuilder:default="TryNextFlavor"
   WhenCanPreempt FlavorFungibilityPolicy `json:"whenCanPreempt"`
 }
 
@@ -187,11 +212,28 @@ If flavorFungibility is nil in configuration, we will set the `WhenCanBorrow` to
 
 ### Behavior Changes
 
-We will not change the behavior to judge whether a podset can get enough resource in certain resource flavor. Preemption and admission will not be influenced also. We only change the order these flavors were considered.
+We will not change the behavior to judge whether a podset can get enough resource in certain resource flavor. Preemption and admission will not be influenced also. We don't change the order these flavors are considered, but `WhenCanBorrow` and `WhenCanPreempt` control how many flavors are considered.
+
+We try to schedule a podset in succesive resource flavors in a loop and we decide whether to traverse to the next flavor or break from the loop based on the `flavorFungibility`. The behavior depending on the simulated scheduling result and the configuration is as follows:
 
-After we try to schedule a podset in a resource flavor, we decide whether to traverse to the next flavor base on the `flavorFungibility`. If the assignment mode is `NoFit`, we will always try the next flavor until the last one. When the assignment mode is `Preempt`, we can return the current assignment if `WhenCanPreempt` is `Preempt`. Otherwise if the assignment mode is `Fit`, we try the next flavor only when we need borrowing in the current flavor and `WhenCanBorrow` is `TryNextFlavor`.
+| Simulation result  |Configuration  | Behavior |
+|------ | -------- | ------- |
+| `NoFit` | any | continue |
+| `Preempt`, `NoBorrow` | `WhenCanPreempt = TryNextFlavor` | continue |
+| `Preempt`, `NoBorrow` | `WhenCanPreempt = Preempt` | break |
+| `Fit`, `Borrow` | `WhenCanBorrow = TryNextFlavor` | continue |
+| `Fit`, `Borrow` | `WhenCanBorrow = Borrow` | break |
+| `Fit`, `NoBorrow`| any| break| 
 
-We will store the scheduling context in workload info so that we can start from where we stop in previous scheduling attempts. This will be useful to avoid to waste time in one flavor all the time if we try to preempt in a flavor and failed. Scheduling context will contain the `LastScheduledFlavorIdx`, `ClusterQueueGeneration` attached to the CQ and `CohortGeneration`. Any changes to these properties will lead to a scheduling from the first flavor.
+By `Borrow`/`NoBorrow` we mean whether borrowing is required or not to fit the considered podset in the sumulation result. After we complete the loop, either by trying all the flavors or by breaking out of it, we end up with a list of possible flavors containing all the considered flavors that yielded a result different than `NoFit`. We choose the flavor to assign based on the following default preference order: (`Fit`, `NoBorrow`), (`Fit`, `Borrow`), (`Preempt`, `NoBorrow`), (`Preempt`, `Borrow`).
+
+An alternative order of preference can be set by enabling the feature gate `FlavorFungibilityImplicitPreferenceDefault`.
+The alternative order prioritizes assignments that don't borrow: (`Fit`, `NoBorrow`),(`Preempt`, `NoBorrow`), (`Fit`, `Borrow`), (`Preempt`, `Borrow`). It is used in two cases:
+
+1. If `WhenCanPreempt = Preempt` and `WhenCanBorrow = TryNextFlavor`
+2. If `WhenCanPreempt = TryNextFlavor` and `WhenCanBorrow = TryNextFlavor`
+
+We will store the scheduling context in workload info so that we can start from where we stop in previous scheduling attempts. This will be useful to avoid to waste time in one flavor all the time if we try to preempt in a flavor and failed. Scheduling context will contain the `LastTriedFlavorIdx`, `ClusterQueueGeneration` attached to the CQ and `CohortGeneration`. Any changes to these properties will lead to a scheduling from the first flavor.
 
 `ClusterQueueGeneration` and `CohortGeneration` mark record the resource consumption of the CQs and Cohort. Any time the available resources of the CQs or Cohort increase, we will increase the generation. So that if the Generation in scheduling context is lower, we should retry from the first flavor. Note that increasing after decreasing of the available resource will also make the generation increased, but I think this is acceptable since we can save the memory by just storing the generation instead of the usage state for each scheduling attempt.
 
@@ -366,6 +408,11 @@ milestones with these graduation criteria:
 [deprecation-policy]: https://kubernetes.io/docs/reference/using-api/deprecation-policy/
 -->
 
+The feature gate `FlavorFungibilityImplicitPreferenceDefault` is a temporary measure
+until a new shape of the API, that allows to explicitely define the preference
+is implemented. The feature gate will be removed once the API is updated
+(planned for kueue v0.14).
+
 ## Implementation History
 
 <!--
diff --git a/keps/582-preempt-based-on-flavor-order/kep.yaml b/keps/582-preempt-based-on-flavor-order/kep.yaml
index 74967e077..4a9dd517f 100644
--- a/keps/582-preempt-based-on-flavor-order/kep.yaml
+++ b/keps/582-preempt-based-on-flavor-order/kep.yaml
@@ -24,7 +24,9 @@ milestone:
 
 # The following PRR answers are required at alpha release
 # List the feature gate name and the components for which it must be enabled
-feature-gates: FlavorFungibility
+feature-gates:
+  - FlavorFungibility
+  - FlavorFungibilityImplicitPreferenceDefault
 disable-supported: true
 
 # The following PRR answers are required at beta release
diff --git a/pkg/controller/core/indexer/indexer.go b/pkg/controller/core/indexer/indexer.go
index df9aa3ed0..883fad67b 100644
--- a/pkg/controller/core/indexer/indexer.go
+++ b/pkg/controller/core/indexer/indexer.go
@@ -23,6 +23,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -38,8 +39,56 @@ const (
 	WorkloadQuotaReservedKey   = "status.quotaReserved"
 	WorkloadRuntimeClassKey    = "spec.runtimeClass"
 	OwnerReferenceUID          = "metadata.ownerReferences.uid"
+
+	// OwnerReferenceGroupKindFmt defines the format string used to construct a field path
+	// for indexing or matching against a specific owner Group and Kind in a Kubernetes object's metadata.
+	//
+	// The format expects two placeholders: the owner's Group and Kind, and produces a path like:
+	// ".metadata.ownerReferences[<group>.<kind>]"
+	// which can be used as an index key in field selectors.
+	//
+	// Example:
+	//
+	//	fmt.Sprintf(OwnerReferenceGroupKindFmt, "batch", "Job")
+	//	=> ".metadata.ownerReferences[batch.Job]"
+	ownerReferenceGroupKindFmt = ".metadata.ownerReferences[%s.%s]"
 )
 
+// OwnerReferenceIndexKey returns an index key based on the workload owner's GroupVersionKind and Name.
+func OwnerReferenceIndexKey(ownerGVK schema.GroupVersionKind) string {
+	return fmt.Sprintf(ownerReferenceGroupKindFmt, ownerGVK.Group, ownerGVK.Kind)
+}
+
+// OwnerReferenceIndexFieldMatcher returns a field matcher used to filter objects based on a specific OwnerReference.
+func OwnerReferenceIndexFieldMatcher(gvk schema.GroupVersionKind, name string) client.MatchingFields {
+	return client.MatchingFields{OwnerReferenceIndexKey(gvk): name}
+}
+
+// SetupWorkloadOwnerIndex registers a field index on kueue.Workload objects based on their OwnerReferences
+// that match the specified GroupVersionKind.
+func SetupWorkloadOwnerIndex(ctx context.Context, indexer client.FieldIndexer, gvk schema.GroupVersionKind) error {
+	return indexer.IndexField(ctx, &kueue.Workload{}, OwnerReferenceIndexKey(gvk), WorkloadOwnerIndexFunc(gvk))
+}
+
+// WorkloadOwnerIndexFunc returns the field indexing function for kueue.Workload objects based on their OwnerReferences
+// that match the specified GroupVersionKind.
+func WorkloadOwnerIndexFunc(gvk schema.GroupVersionKind) client.IndexerFunc {
+	return func(object client.Object) []string {
+		wl, ok := object.(*kueue.Workload)
+		if !ok || len(wl.OwnerReferences) == 0 {
+			return nil
+		}
+		owners := make([]string, 0, len(wl.OwnerReferences))
+		for i := range wl.OwnerReferences {
+			owner := &wl.OwnerReferences[i]
+			if owner.Kind == gvk.Kind && owner.APIVersion == gvk.GroupVersion().String() {
+				owners = append(owners, owner.Name)
+			}
+		}
+		return owners
+	}
+}
+
 func IndexQueueClusterQueue(obj client.Object) []string {
 	q, ok := obj.(*kueue.LocalQueue)
 	if !ok {
diff --git a/pkg/controller/jobframework/indexer.go b/pkg/controller/jobframework/indexer.go
index 41094feb0..f8929ad57 100644
--- a/pkg/controller/jobframework/indexer.go
+++ b/pkg/controller/jobframework/indexer.go
@@ -18,30 +18,16 @@ package jobframework
 
 import (
 	"context"
-	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	coreindexer "sigs.k8s.io/kueue/pkg/controller/core/indexer"
 )
 
-// OwnerReferenceGroupKindFmt defines the format string used to construct a field path
-// for indexing or matching against a specific owner Group and Kind in a Kubernetes object's metadata.
-//
-// The format expects two placeholders: the owner's Group and Kind, and produces a path like:
-// ".metadata.ownerReferences[<group>.<kind>]"
-// which can be used as an index key in field selectors.
-//
-// Example:
-//
-//	fmt.Sprintf(OwnerReferenceGroupKindFmt, "batch", "Job")
-//	=> ".metadata.ownerReferences[batch.Job]"
-const OwnerReferenceGroupKindFmt = ".metadata.ownerReferences[%s.%s]"
-
 // OwnerReferenceIndexKey returns an index key based on the workload owner's GroupVersionKind and Name.
 func OwnerReferenceIndexKey(ownerGVK schema.GroupVersionKind) string {
-	return fmt.Sprintf(OwnerReferenceGroupKindFmt, ownerGVK.Group, ownerGVK.Kind)
+	return coreindexer.OwnerReferenceIndexKey(ownerGVK)
 }
 
 // OwnerReferenceIndexFieldMatcher returns a field matcher used to filter objects based on a specific OwnerReference.
@@ -57,7 +43,7 @@ func OwnerReferenceIndexKey(ownerGVK schema.GroupVersionKind) string {
 //
 // The index key is derived using OwnerReferenceIndexKey(gvk).
 func OwnerReferenceIndexFieldMatcher(gvk schema.GroupVersionKind, name string) client.MatchingFields {
-	return client.MatchingFields{OwnerReferenceIndexKey(gvk): name}
+	return coreindexer.OwnerReferenceIndexFieldMatcher(gvk, name)
 }
 
 // SetupWorkloadOwnerIndex registers a field index on kueue.Workload objects based on their OwnerReferences
@@ -87,18 +73,5 @@ func OwnerReferenceIndexFieldMatcher(gvk schema.GroupVersionKind, name string) c
 //	matcher := OwnerReferenceIndexFieldMatcher(gvk, "my-job-name")
 //	cl.List(ctx, &workloadList, matcher)
 func SetupWorkloadOwnerIndex(ctx context.Context, indexer client.FieldIndexer, gvk schema.GroupVersionKind) error {
-	return indexer.IndexField(ctx, &kueue.Workload{}, OwnerReferenceIndexKey(gvk), func(object client.Object) []string {
-		wl, ok := object.(*kueue.Workload)
-		if !ok || len(wl.OwnerReferences) == 0 {
-			return nil
-		}
-		owners := make([]string, 0, len(wl.OwnerReferences))
-		for i := range wl.OwnerReferences {
-			owner := &wl.OwnerReferences[i]
-			if owner.Kind == gvk.Kind && owner.APIVersion == gvk.GroupVersion().String() {
-				owners = append(owners, owner.Name)
-			}
-		}
-		return owners
-	})
+	return coreindexer.SetupWorkloadOwnerIndex(ctx, indexer, gvk)
 }
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index dc70666a9..b39bc49fb 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -180,6 +180,13 @@ const (
 	// Enforces that even Jobs with a queue-name label are only reconciled if their namespace
 	// matches managedJobsNamespaceSelector.
 	ManagedJobsNamespaceSelectorAlwaysRespected featuregate.Feature = "ManagedJobsNamespaceSelectorAlwaysRespected"
+
+	// owner: @pajakd
+	// kep: https://github.com/kubernetes-sigs/kueue/tree/main/keps/582-preempt-based-on-flavor-order
+	//
+	// In flavor fungiblity, the preference whether to preempt or borrow is inferred from flavor fungiblity policy
+	// This feature gate is going to be replaced by an API before graduation or deprecation.
+	FlavorFungibilityImplicitPreferenceDefault featuregate.Feature = "FlavorFungibilityImplicitPreferenceDefault"
 )
 
 func init() {
@@ -278,6 +285,9 @@ var defaultVersionedFeatureGates = map[featuregate.Feature]featuregate.Versioned
 	ManagedJobsNamespaceSelectorAlwaysRespected: {
 		{Version: version.MustParse("0.13"), Default: false, PreRelease: featuregate.Alpha},
 	},
+	FlavorFungibilityImplicitPreferenceDefault: {
+		{Version: version.MustParse("0.13"), Default: false, PreRelease: featuregate.Alpha},
+	},
 }
 
 func SetFeatureGateDuringTest(tb testing.TB, f featuregate.Feature, value bool) {
diff --git a/pkg/scheduler/flavorassigner/flavorassigner.go b/pkg/scheduler/flavorassigner/flavorassigner.go
index 2a0c9c0cb..59fca06a3 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner.go
@@ -309,11 +309,16 @@ func (m FlavorAssignmentMode) String() string {
 
 // granularMode is the FlavorAssignmentMode internal to
 // FlavorAssigner, which lets us distinguish priority based preemption,
-// and reclamation within Cohort.
-type granularMode int
+// reclamation within Cohort and borrowing.
+type granularMode struct {
+	preemptionMode preemptionMode
+	needsBorrowing bool
+}
+
+type preemptionMode int
 
 const (
-	noFit granularMode = iota
+	noFit preemptionMode = iota
 	// noPreemptionCandidates indicates that admission is possible with
 	// preemption, but simulation found no preemption targets.
 	noPreemptionCandidates
@@ -322,7 +327,42 @@ const (
 	fit
 )
 
-func fromPreemptionPossibility(preemptionPossibility preemptioncommon.PreemptionPossibility) granularMode {
+func preferToAvoidBorrowing(fungiblityConfig kueue.FlavorFungibility) bool {
+	return (fungiblityConfig.WhenCanBorrow == kueue.TryNextFlavor && fungiblityConfig.WhenCanPreempt == kueue.Preempt) ||
+		(fungiblityConfig.WhenCanBorrow == kueue.TryNextFlavor && fungiblityConfig.WhenCanPreempt == kueue.TryNextFlavor)
+}
+
+// isPreferred returns true if mode a is better than b according to the selected policy
+func isPreferred(a, b granularMode, fungiblityConfig kueue.FlavorFungibility) bool {
+	if a.preemptionMode == noFit {
+		return false
+	}
+	if b.preemptionMode == noFit {
+		return true
+	}
+
+	if !features.Enabled(features.FlavorFungibilityImplicitPreferenceDefault) {
+		if a.preemptionMode != b.preemptionMode {
+			return a.preemptionMode > b.preemptionMode
+		} else {
+			return !a.needsBorrowing && b.needsBorrowing
+		}
+	}
+
+	if preferToAvoidBorrowing(fungiblityConfig) {
+		if a.needsBorrowing != b.needsBorrowing {
+			return !a.needsBorrowing
+		}
+		return a.preemptionMode > b.preemptionMode
+	} else {
+		if a.preemptionMode != b.preemptionMode {
+			return a.preemptionMode > b.preemptionMode
+		}
+		return !a.needsBorrowing && b.needsBorrowing
+	}
+}
+
+func fromPreemptionPossibility(preemptionPossibility preemptioncommon.PreemptionPossibility) preemptionMode {
 	switch preemptionPossibility {
 	case preemptioncommon.NoCandidates:
 		return noPreemptionCandidates
@@ -334,7 +374,7 @@ func fromPreemptionPossibility(preemptionPossibility preemptioncommon.Preemption
 	panic(fmt.Sprintf("illegal PreemptionPossibility: %d", preemptionPossibility))
 }
 
-func (mode granularMode) flavorAssignmentMode() FlavorAssignmentMode {
+func (mode preemptionMode) flavorAssignmentMode() FlavorAssignmentMode {
 	switch mode {
 	case noFit:
 		return NoFit
@@ -353,7 +393,7 @@ func (mode granularMode) flavorAssignmentMode() FlavorAssignmentMode {
 
 // isPreemptMode indicates a mode where preemption targets were found.
 func (mode granularMode) isPreemptMode() bool {
-	return mode == preempt || mode == reclaim
+	return mode.preemptionMode == preempt || mode.preemptionMode == reclaim
 }
 
 type FlavorAssignment struct {
@@ -364,7 +404,7 @@ type FlavorAssignment struct {
 }
 
 type preemptionOracle interface {
-	SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) preemptioncommon.PreemptionPossibility
+	SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) (preemptioncommon.PreemptionPossibility, int)
 }
 
 type FlavorAssigner struct {
@@ -614,7 +654,7 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 	}
 
 	var bestAssignment ResourceAssignment
-	bestAssignmentMode := noFit
+	bestAssignmentMode := granularMode{preemptionMode: noFit, needsBorrowing: true}
 
 	// We will only check against the flavors' labels for the resource.
 	attemptedFlavorIdx := -1
@@ -661,10 +701,9 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 			status.reasons = append(status.reasons, *flavorUnmatchMessage)
 			continue
 		}
-		needsBorrowing := false
 		assignments := make(ResourceAssignment, len(requests))
 		// Calculate representativeMode for this assignment as the worst mode among all requests.
-		representativeMode := fit
+		representativeMode := granularMode{preemptionMode: fit, needsBorrowing: false}
 		for rName, val := range requests {
 			// Ensure the same resource flavor is used for the workload slice as in the original admitted slice.
 			if features.Enabled(features.ElasticJobsViaWorkloadSlices) && a.preemptWorkloadSlice != nil {
@@ -674,7 +713,7 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 					// Enforce consistent resource flavor assignment between slices.
 					if originalFlavor := preemptWorkloadRequests.Flavors[rName]; originalFlavor != fName {
 						// Flavor mismatch. Skip further checks for this resource.
-						representativeMode = noFit
+						representativeMode = granularMode{preemptionMode: noFit, needsBorrowing: true}
 						status.reasons = append(status.reasons, fmt.Sprintf("could not assign %s flavor since the original workload is assigned: %s", fName, originalFlavor))
 						break
 					}
@@ -687,40 +726,39 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 			resQuota := a.cq.QuotaFor(resources.FlavorResource{Flavor: fName, Resource: rName})
 			// Check considering the flavor usage by previous pod sets.
 			fr := resources.FlavorResource{Flavor: fName, Resource: rName}
-			mode, borrow, s := a.fitsResourceQuota(log, fr, val+assignmentUsage[fr], resQuota)
+			preemptionMode, borrow, s := a.fitsResourceQuota(log, fr, val+assignmentUsage[fr], resQuota)
 			if s != nil {
 				status.reasons = append(status.reasons, s.reasons...)
 			}
-			if mode < representativeMode {
+			mode := granularMode{preemptionMode, borrow > 0}
+			if isPreferred(representativeMode, mode, a.cq.FlavorFungibility) {
 				representativeMode = mode
 			}
-			needsBorrowing = needsBorrowing || (borrow > 0)
-			if representativeMode == noFit {
+			if representativeMode.preemptionMode == noFit {
 				// The flavor doesn't fit, no need to check other resources.
 				break
 			}
 
 			assignments[rName] = &FlavorAssignment{
 				Name:   fName,
-				Mode:   mode.flavorAssignmentMode(),
+				Mode:   preemptionMode.flavorAssignmentMode(),
 				borrow: borrow,
 			}
 		}
-
 		if features.Enabled(features.FlavorFungibility) {
-			if !shouldTryNextFlavor(representativeMode, a.cq.FlavorFungibility, needsBorrowing) {
+			if !shouldTryNextFlavor(representativeMode, a.cq.FlavorFungibility) {
 				bestAssignment = assignments
 				bestAssignmentMode = representativeMode
 				break
 			}
-			if representativeMode > bestAssignmentMode {
+			if isPreferred(representativeMode, bestAssignmentMode, a.cq.FlavorFungibility) {
 				bestAssignment = assignments
 				bestAssignmentMode = representativeMode
 			}
-		} else if representativeMode > bestAssignmentMode {
+		} else if representativeMode.preemptionMode > bestAssignmentMode.preemptionMode {
 			bestAssignment = assignments
 			bestAssignmentMode = representativeMode
-			if bestAssignmentMode == fit {
+			if bestAssignmentMode.preemptionMode == fit {
 				// All the resources fit in the cohort, no need to check more flavors.
 				return bestAssignment, nil
 			}
@@ -736,27 +774,27 @@ func (a *FlavorAssigner) findFlavorForPodSetResource(
 				assignment.TriedFlavorIdx = attemptedFlavorIdx
 			}
 		}
-		if bestAssignmentMode == fit {
+		if bestAssignmentMode.preemptionMode == fit {
 			return bestAssignment, nil
 		}
 	}
 	return bestAssignment, status
 }
 
-func shouldTryNextFlavor(representativeMode granularMode, flavorFungibility kueue.FlavorFungibility, needsBorrowing bool) bool {
+func shouldTryNextFlavor(representativeMode granularMode, flavorFungibility kueue.FlavorFungibility) bool {
 	policyPreempt := flavorFungibility.WhenCanPreempt
 	policyBorrow := flavorFungibility.WhenCanBorrow
 	if representativeMode.isPreemptMode() && policyPreempt == kueue.Preempt {
-		if !needsBorrowing || policyBorrow == kueue.Borrow {
+		if !representativeMode.needsBorrowing || policyBorrow == kueue.Borrow {
 			return false
 		}
 	}
 
-	if representativeMode == fit && needsBorrowing && policyBorrow == kueue.Borrow {
+	if representativeMode.preemptionMode == fit && representativeMode.needsBorrowing && policyBorrow == kueue.Borrow {
 		return false
 	}
 
-	if representativeMode == fit && !needsBorrowing {
+	if representativeMode.preemptionMode == fit && !representativeMode.needsBorrowing {
 		return false
 	}
 
@@ -815,7 +853,7 @@ func flavorSelector(spec *corev1.PodSpec, allowedKeys sets.Set[string]) nodeaffi
 // if borrowing is required when preempting.
 // If the flavor doesn't satisfy limits immediately (when waiting or preemption
 // could help), it returns a Status with reasons.
-func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorResource, val int64, rQuota cache.ResourceQuota) (granularMode, int, *Status) {
+func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorResource, val int64, rQuota cache.ResourceQuota) (preemptionMode, int, *Status) {
 	var status Status
 
 	available := a.cq.Available(fr)
@@ -839,8 +877,9 @@ func (a *FlavorAssigner) fitsResourceQuota(log logr.Logger, fr resources.FlavorR
 		fr.Resource, fr.Flavor, resources.ResourceQuantityString(fr.Resource, val-available))
 
 	if val <= rQuota.Nominal || mayReclaimInHierarchy || a.canPreemptWhileBorrowing() {
-		mode := fromPreemptionPossibility(a.oracle.SimulatePreemption(log, a.cq, *a.wl, fr, val))
-		return mode, borrow, &status
+		preemptionPossiblity, borrowAfterPreemptions := a.oracle.SimulatePreemption(log, a.cq, *a.wl, fr, val)
+		mode := fromPreemptionPossibility(preemptionPossiblity)
+		return mode, borrowAfterPreemptions, &status
 	}
 	return noFit, borrow, &status
 }
diff --git a/pkg/scheduler/preemption/preemption_oracle.go b/pkg/scheduler/preemption/preemption_oracle.go
index b4fca0d02..dd91128bb 100644
--- a/pkg/scheduler/preemption/preemption_oracle.go
+++ b/pkg/scheduler/preemption/preemption_oracle.go
@@ -22,6 +22,7 @@ import (
 
 	"sigs.k8s.io/kueue/pkg/cache"
 	"sigs.k8s.io/kueue/pkg/resources"
+	"sigs.k8s.io/kueue/pkg/scheduler/preemption/classical"
 	preemptioncommon "sigs.k8s.io/kueue/pkg/scheduler/preemption/common"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
@@ -37,7 +38,7 @@ type PreemptionOracle struct {
 
 // SimulatePreemption runs the preemption algorithm for a given flavor resource to check if
 // preemption and reclaim are possible in this flavor resource.
-func (p *PreemptionOracle) SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) preemptioncommon.PreemptionPossibility {
+func (p *PreemptionOracle) SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) (preemptioncommon.PreemptionPossibility, int) {
 	candidates := p.preemptor.getTargets(&preemptionCtx{
 		log:               log,
 		preemptor:         wl,
@@ -46,13 +47,23 @@ func (p *PreemptionOracle) SimulatePreemption(log logr.Logger, cq *cache.Cluster
 		frsNeedPreemption: sets.New(fr),
 		workloadUsage:     workload.Usage{Quota: resources.FlavorResourceQuantities{fr: quantity}},
 	})
+
+	if len(candidates) > 0 {
+		workloadsToPreempt := make([]*workload.Info, len(candidates))
+		for i, c := range candidates {
+			workloadsToPreempt[i] = c.WorkloadInfo
+		}
+		revertRemoval := cq.SimulateWorkloadRemoval(workloadsToPreempt)
+		defer revertRemoval()
+	}
+	borrowAfterPreemptions, _ := classical.FindHeightOfLowestSubtreeThatFits(cq, fr, quantity)
 	if len(candidates) == 0 {
-		return preemptioncommon.NoCandidates
+		return preemptioncommon.NoCandidates, borrowAfterPreemptions
 	}
 	for _, candidate := range candidates {
 		if candidate.WorkloadInfo.ClusterQueue == cq.Name {
-			return preemptioncommon.Preempt
+			return preemptioncommon.Preempt, borrowAfterPreemptions
 		}
 	}
-	return preemptioncommon.Reclaim
+	return preemptioncommon.Reclaim, borrowAfterPreemptions
 }
diff --git a/pkg/util/client/client.go b/pkg/util/client/client.go
index 3b494af08..f24235f00 100644
--- a/pkg/util/client/client.go
+++ b/pkg/util/client/client.go
@@ -22,7 +22,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func CreatePatch(before, after client.Object) (client.Patch, error) {
+func createPatch(before, after client.Object) (client.Patch, error) {
 	patchBase := client.MergeFrom(before)
 	patchBytes, err := patchBase.Data(after)
 	if err != nil {
@@ -35,42 +35,38 @@ func CreatePatch(before, after client.Object) (client.Patch, error) {
 // If strict is true, the resourceVersion will be part of the patch, make this call fail if
 // client.Object was changed.
 func Patch(ctx context.Context, c client.Client, obj client.Object, strict bool, update func() (bool, error)) error {
-	objOriginal := obj.DeepCopyObject().(client.Object)
-	if strict {
-		// Clearing ResourceVersion from the original object to make sure it is included in the generated patch.
-		objOriginal.SetResourceVersion("")
-	}
-	updated, err := update()
-	if err != nil || !updated {
-		return err
-	}
-	patch, err := CreatePatch(objOriginal, obj)
-	if err != nil {
-		return err
-	}
-	if err = c.Patch(ctx, obj, patch); err != nil {
-		return err
-	}
-	return nil
+	return updateAndPatch(obj, strict, update, func(patch client.Patch) error {
+		return c.Patch(ctx, obj, patch)
+	})
 }
 
 // PatchStatus applies the merge patch of client.Object status.
 // The resourceVersion will be part of the patch, make this call fail if
 // client.Object was changed.
 func PatchStatus(ctx context.Context, c client.Client, obj client.Object, update func() (bool, error)) error {
+	return updateAndPatch(obj, true, update, func(patch client.Patch) error {
+		return c.Status().Patch(ctx, obj, patch)
+	})
+}
+
+func getOriginalObject(obj client.Object, strict bool) client.Object {
 	objOriginal := obj.DeepCopyObject().(client.Object)
-	// Clearing ResourceVersion from the original object to make sure it is included in the generated patch.
-	objOriginal.SetResourceVersion("")
+	if strict {
+		// Clearing ResourceVersion from the original object to make sure it is included in the generated patch.
+		objOriginal.SetResourceVersion("")
+	}
+	return objOriginal
+}
+
+func updateAndPatch(obj client.Object, strict bool, update func() (bool, error), patchFn func(client.Patch) error) error {
+	objOriginal := getOriginalObject(obj, strict)
 	updated, err := update()
 	if err != nil || !updated {
 		return err
 	}
-	patch, err := CreatePatch(objOriginal, obj)
+	patch, err := createPatch(objOriginal, obj)
 	if err != nil {
 		return err
 	}
-	if err = c.Status().Patch(ctx, obj, patch); err != nil {
-		return err
-	}
-	return nil
+	return patchFn(patch)
 }
diff --git a/pkg/workloadslicing/workloadslicing.go b/pkg/workloadslicing/workloadslicing.go
index 3fd30dc18..03b526749 100644
--- a/pkg/workloadslicing/workloadslicing.go
+++ b/pkg/workloadslicing/workloadslicing.go
@@ -115,7 +115,7 @@ func Finish(ctx context.Context, clnt client.Client, workloadSlice *kueue.Worklo
 // without "Finished" condition with status = "True".
 func FindNotFinishedWorkloads(ctx context.Context, clnt client.Client, jobObject client.Object, jobObjectGVK schema.GroupVersionKind) ([]kueue.Workload, error) {
 	list := &kueue.WorkloadList{}
-	if err := clnt.List(ctx, list, client.InNamespace(jobObject.GetNamespace()), client.MatchingFields{fmt.Sprintf(".metadata.ownerReferences[%s.%s]", jobObjectGVK.Group, jobObjectGVK.Kind): jobObject.GetName()}); err != nil {
+	if err := clnt.List(ctx, list, client.InNamespace(jobObject.GetNamespace()), indexer.OwnerReferenceIndexFieldMatcher(jobObjectGVK, jobObject.GetName())); err != nil {
 		return nil, err
 	}
 



Test patch:
diff --git a/pkg/scheduler/flavorassigner/flavorassigner_test.go b/pkg/scheduler/flavorassigner/flavorassigner_test.go
index 34f77b3bc..67768df63 100644
--- a/pkg/scheduler/flavorassigner/flavorassigner_test.go
+++ b/pkg/scheduler/flavorassigner/flavorassigner_test.go
@@ -47,17 +47,22 @@ var (
 	})
 )
 
+type simulationResultForFlavor struct {
+	preemptionPossiblity     preemptioncommon.PreemptionPossibility
+	borrowingAfterSimulation int
+}
+
 type testOracle struct {
-	simulationResult map[resources.FlavorResource]preemptioncommon.PreemptionPossibility
+	simulationResult map[resources.FlavorResource]simulationResultForFlavor
 }
 
-func (f *testOracle) SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) preemptioncommon.PreemptionPossibility {
+func (f *testOracle) SimulatePreemption(log logr.Logger, cq *cache.ClusterQueueSnapshot, wl workload.Info, fr resources.FlavorResource, quantity int64) (preemptioncommon.PreemptionPossibility, int) {
 	if f.simulationResult != nil {
 		if result, ok := f.simulationResult[fr]; ok {
-			return result
+			return result.preemptionPossiblity, result.borrowingAfterSimulation
 		}
 	}
-	return preemptioncommon.Preempt
+	return preemptioncommon.Preempt, 0
 }
 
 func TestAssignFlavors(t *testing.T) {
@@ -99,9 +104,10 @@ func TestAssignFlavors(t *testing.T) {
 		wantAssignment                      Assignment
 		disableLendingLimit                 bool
 		enableFairSharing                   bool
-		simulationResult                    map[resources.FlavorResource]preemptioncommon.PreemptionPossibility
+		simulationResult                    map[resources.FlavorResource]simulationResultForFlavor
 		elasticJobsViaWorkloadSlicesEnabled bool
 		preemptWorkloadSlice                *workload.Info
+		enableImplicitPreferenceDefault     bool
 	}{
 		"single flavor, fits": {
 			wlPods: []kueue.PodSet{
@@ -573,6 +579,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "b_one", Resource: "example.com/gpu"}: 2,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "two", Resource: corev1.ResourceMemory}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				PodSets: []PodSetAssignment{{
@@ -1054,6 +1063,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 9_000,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				PodSets: []PodSetAssignment{{
@@ -1132,6 +1144,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 8_000,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				PodSets: []PodSetAssignment{{
@@ -1655,6 +1670,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				Borrowing: 1,
@@ -1710,6 +1728,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				Borrowing: 1,
@@ -1965,8 +1986,8 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 2_000,
 			},
-			simulationResult: map[resources.FlavorResource]preemptioncommon.PreemptionPossibility{
-				{Flavor: "one", Resource: corev1.ResourceCPU}: preemptioncommon.NoCandidates,
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.NoCandidates, 0},
 			},
 			wantRepMode: Fit,
 			wantAssignment: Assignment{
@@ -2011,7 +2032,9 @@ func TestAssignFlavors(t *testing.T) {
 				).
 				Cohort("test-cohort").
 				Obj(),
-
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				PodSets: []PodSetAssignment{{
@@ -2063,6 +2086,9 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
 			},
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.Preempt, 1},
+			},
 			wantRepMode: Preempt,
 			wantAssignment: Assignment{
 				Borrowing: 1,
@@ -2111,8 +2137,8 @@ func TestAssignFlavors(t *testing.T) {
 			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
 				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
 			},
-			simulationResult: map[resources.FlavorResource]preemptioncommon.PreemptionPossibility{
-				{Flavor: "one", Resource: corev1.ResourceCPU}: preemptioncommon.NoCandidates,
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: {preemptioncommon.NoCandidates, 0},
 			},
 			wantRepMode: Fit,
 			wantAssignment: Assignment{
@@ -2238,6 +2264,164 @@ func TestAssignFlavors(t *testing.T) {
 				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{}},
 			},
 		},
+		"preferPreemption; preempt without borrowing is preferred over fit with borrowing; preempt within cohort": {
+			enableImplicitPreferenceDefault: true,
+			wlPods: []kueue.PodSet{
+				*utiltesting.MakePodSet("main", 1).
+					Request(corev1.ResourceCPU, "10").
+					Obj(),
+			},
+			clusterQueue: utiltesting.MakeClusterQueue("cq-test").
+				Cohort("cohort").
+				FlavorFungibility(kueue.FlavorFungibility{
+					WhenCanPreempt: kueue.TryNextFlavor,
+					WhenCanBorrow:  kueue.TryNextFlavor,
+				}).
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "10").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+				).ClusterQueue,
+			secondaryClusterQueue: utiltesting.MakeClusterQueue("cq-secondary").
+				Cohort("cohort").
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+				).
+				Obj(),
+			secondaryClusterQueueUsage: resources.FlavorResourceQuantities{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
+			},
+			wantRepMode: Preempt,
+			wantAssignment: Assignment{
+				PodSets: []PodSetAssignment{{
+					Name: "main",
+					Flavors: ResourceAssignment{
+						corev1.ResourceCPU: {Name: "one", Mode: Preempt, TriedFlavorIdx: -1},
+					},
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU: resource.MustParse("10"),
+					},
+					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 5 more needed"),
+					Count:  1,
+				}},
+				Borrowing: 0,
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
+					{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
+				}},
+			},
+		},
+		"preferPreemption; preempt without borrowing is preferred over fit with borrowing; preempt in own CQ": {
+			enableImplicitPreferenceDefault: true,
+			wlPods: []kueue.PodSet{
+				*utiltesting.MakePodSet("main", 1).
+					Request(corev1.ResourceCPU, "10").
+					Obj(),
+			},
+			clusterQueue: utiltesting.MakeClusterQueue("cq-test").
+				Cohort("cohort").
+				FlavorFungibility(kueue.FlavorFungibility{
+					WhenCanPreempt: kueue.TryNextFlavor,
+					WhenCanBorrow:  kueue.TryNextFlavor,
+				}).
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "10").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+				).ClusterQueue,
+			secondaryClusterQueue: utiltesting.MakeClusterQueue("cq-secondary").
+				Cohort("cohort").
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+				).
+				Obj(),
+			clusterQueueUsage: resources.FlavorResourceQuantities{
+				{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
+			},
+			wantRepMode: Preempt,
+			wantAssignment: Assignment{
+				PodSets: []PodSetAssignment{{
+					Name: "main",
+					Flavors: ResourceAssignment{
+						corev1.ResourceCPU: {Name: "one", Mode: Preempt, TriedFlavorIdx: -1},
+					},
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU: resource.MustParse("10"),
+					},
+					Status: *NewStatus("insufficient unused quota for cpu in flavor one, 5 more needed"),
+					Count:  1,
+				}},
+				Borrowing: 0,
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
+					{Flavor: "one", Resource: corev1.ResourceCPU}: 10_000,
+				}},
+			},
+		},
+		"preferPreemption; fit without borrowing is preferred over fit with borrowing": {
+			enableImplicitPreferenceDefault: true,
+			wlPods: []kueue.PodSet{
+				*utiltesting.MakePodSet("main", 1).
+					Request(corev1.ResourceCPU, "10").
+					Obj(),
+			},
+			clusterQueue: utiltesting.MakeClusterQueue("cq-test").
+				Cohort("cohort").
+				FlavorFungibility(kueue.FlavorFungibility{
+					WhenCanPreempt: kueue.TryNextFlavor,
+					WhenCanBorrow:  kueue.TryNextFlavor,
+				}).
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "10").
+						FlavorQuotas,
+				).ClusterQueue,
+			secondaryClusterQueue: utiltesting.MakeClusterQueue("cq-secondary").
+				Cohort("cohort").
+				ResourceGroup(
+					utiltesting.MakeFlavorQuotas("one").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+					utiltesting.MakeFlavorQuotas("two").
+						Resource(corev1.ResourceCPU, "5").
+						FlavorQuotas,
+				).
+				Obj(),
+			wantRepMode: Fit,
+			wantAssignment: Assignment{
+				PodSets: []PodSetAssignment{{
+					Name: "main",
+					Flavors: ResourceAssignment{
+						corev1.ResourceCPU: {Name: "two", Mode: Fit, TriedFlavorIdx: -1},
+					},
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU: resource.MustParse("10"),
+					},
+					Count: 1,
+				}},
+				Borrowing: 0,
+				Usage: workload.Usage{Quota: resources.FlavorResourceQuantities{
+					{Flavor: "two", Resource: corev1.ResourceCPU}: 10_000,
+				}},
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
@@ -2245,6 +2429,9 @@ func TestAssignFlavors(t *testing.T) {
 			if tc.disableLendingLimit {
 				features.SetFeatureGateDuringTest(t, features.LendingLimit, false)
 			}
+			if tc.enableImplicitPreferenceDefault {
+				features.SetFeatureGateDuringTest(t, features.FlavorFungibilityImplicitPreferenceDefault, true)
+			}
 			log := testr.NewWithOptions(t, testr.Options{
 				Verbosity: 2,
 			})
@@ -2324,7 +2511,7 @@ func TestReclaimBeforePriorityPreemption(t *testing.T) {
 		flavorFungibility      *kueue.FlavorFungibility
 		wantMode               FlavorAssignmentMode
 		wantAssigment          rfMap
-		simulationResult       map[resources.FlavorResource]preemptioncommon.PreemptionPossibility
+		simulationResult       map[resources.FlavorResource]simulationResultForFlavor
 	}{
 		"Select first flavor which fits": {
 			workloadRequests: utiltesting.MakePodSet(kueue.DefaultPodSetName, 1).Request("gpu", "10"),
@@ -2346,9 +2533,9 @@ func TestReclaimBeforePriorityPreemption(t *testing.T) {
 				{Flavor: "due", Resource: "gpu"}: 1,
 				{Flavor: "tre", Resource: "gpu"}: 1,
 			},
-			simulationResult: map[resources.FlavorResource]preemptioncommon.PreemptionPossibility{
-				{Flavor: "uno", Resource: "gpu"}: preemptioncommon.Preempt,
-				{Flavor: "due", Resource: "gpu"}: preemptioncommon.Reclaim,
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "uno", Resource: "gpu"}: {preemptioncommon.Preempt, 0},
+				{Flavor: "due", Resource: "gpu"}: {preemptioncommon.Reclaim, 0},
 			},
 			wantMode:      Preempt,
 			wantAssigment: rfMap{"gpu": "due"},
@@ -2389,10 +2576,10 @@ func TestReclaimBeforePriorityPreemption(t *testing.T) {
 				{Flavor: "due", Resource: "gpu"}: 1,
 				{Flavor: "tre", Resource: "gpu"}: 1,
 			},
-			simulationResult: map[resources.FlavorResource]preemptioncommon.PreemptionPossibility{
-				{Flavor: "uno", Resource: "gpu"}: preemptioncommon.Preempt,
-				{Flavor: "due", Resource: "gpu"}: preemptioncommon.Reclaim,
-				{Flavor: "tre", Resource: "gpu"}: preemptioncommon.Reclaim,
+			simulationResult: map[resources.FlavorResource]simulationResultForFlavor{
+				{Flavor: "uno", Resource: "gpu"}: {preemptioncommon.Preempt, 0},
+				{Flavor: "due", Resource: "gpu"}: {preemptioncommon.Reclaim, 0},
+				{Flavor: "tre", Resource: "gpu"}: {preemptioncommon.Reclaim, 0},
 			},
 			wantMode:      Preempt,
 			wantAssigment: rfMap{"gpu": "tre", "compute": "tre"},
diff --git a/pkg/workloadslicing/workloadslicing_test.go b/pkg/workloadslicing/workloadslicing_test.go
index 45f276053..4d4397a02 100644
--- a/pkg/workloadslicing/workloadslicing_test.go
+++ b/pkg/workloadslicing/workloadslicing_test.go
@@ -19,7 +19,6 @@ package workloadslicing
 import (
 	"context"
 	"errors"
-	"fmt"
 	"strconv"
 	"testing"
 	"time"
@@ -155,20 +154,7 @@ func testWorkloadClientBuilder() *fake.ClientBuilder {
 	_ = kueue.AddToScheme(testSchema)
 	return fake.NewClientBuilder().
 		WithScheme(testSchema).
-		WithIndex(&kueue.Workload{}, fmt.Sprintf(".metadata.ownerReferences[%s.%s]", testJobGVK.Group, testJobGVK.Kind), func(object client.Object) []string {
-			wl, ok := object.(*kueue.Workload)
-			if !ok || len(wl.OwnerReferences) == 0 {
-				return nil
-			}
-			owners := make([]string, 0, len(wl.OwnerReferences))
-			for i := range wl.OwnerReferences {
-				owner := &wl.OwnerReferences[i]
-				if owner.Kind == testJobGVK.Kind && owner.APIVersion == testJobGVK.GroupVersion().String() {
-					owners = append(owners, owner.Name)
-				}
-			}
-			return owners
-		})
+		WithIndex(&kueue.Workload{}, indexer.OwnerReferenceIndexKey(testJobGVK), indexer.WorkloadOwnerIndexFunc(testJobGVK))
 }
 
 func testWorkload(name, jobName string, jobUID types.UID, created time.Time) *utiltesting.WorkloadWrapper {
diff --git a/test/integration/singlecluster/scheduler/scheduler_test.go b/test/integration/singlecluster/scheduler/scheduler_test.go
index d34947798..71eee5088 100644
--- a/test/integration/singlecluster/scheduler/scheduler_test.go
+++ b/test/integration/singlecluster/scheduler/scheduler_test.go
@@ -2580,6 +2580,107 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			}
 		})
 	})
+	ginkgo.When("FlavorFungibilityImplicitPreferenceDefault is enabled", func() {
+		ginkgo.BeforeEach(func() {
+			f1 := testing.MakeResourceFlavor("f1").Obj()
+			util.MustCreate(ctx, k8sClient, f1)
+
+			f2 := testing.MakeResourceFlavor("f2").Obj()
+			util.MustCreate(ctx, k8sClient, f2)
+			_ = features.SetEnable(features.FlavorFungibilityImplicitPreferenceDefault, true)
+		})
+		ginkgo.AfterEach(func() {
+			_ = features.SetEnable(features.FlavorFungibilityImplicitPreferenceDefault, false)
+		})
+		ginkgo.It("chooses a correct flavor when preemption is preferred", func() {
+			fungibility := kueue.FlavorFungibility{
+				WhenCanBorrow:  kueue.TryNextFlavor,
+				WhenCanPreempt: kueue.Preempt}
+			preemption := kueue.ClusterQueuePreemption{WithinClusterQueue: kueue.PreemptionPolicyLowerPriority, ReclaimWithinCohort: kueue.PreemptionPolicyAny, BorrowWithinCohort: &kueue.BorrowWithinCohort{Policy: kueue.BorrowWithinCohortPolicyLowerPriority}}
+
+			createQueue(testing.MakeClusterQueue("cq1").
+				FlavorFungibility(fungibility).Cohort("cohort").
+				Preemption(preemption).
+				ResourceGroup(testing.MakeFlavorQuotas("f1").Resource(corev1.ResourceCPU, "0").FlavorQuotas, testing.MakeFlavorQuotas("f2").Resource(corev1.ResourceCPU, "1").FlavorQuotas).Obj())
+
+			createQueue(testing.MakeClusterQueue("cq2").Cohort("cohort").
+				FlavorFungibility(fungibility).
+				Preemption(preemption).
+				ResourceGroup(testing.MakeFlavorQuotas("f1").Resource(corev1.ResourceCPU, "1").FlavorQuotas, testing.MakeFlavorQuotas("f2").Resource(corev1.ResourceCPU, "0").FlavorQuotas).Obj())
+
+			cq1LowPriority := createWorkloadWithPriority("cq1", "1", 0)
+			{
+				admission := testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f2", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq1LowPriority, admission)
+			}
+
+			cq2HighPriority := createWorkloadWithPriority("cq1", "1", 9999)
+			{
+				util.ExpectWorkloadsToBePreempted(ctx, k8sClient, cq1LowPriority)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, cq1LowPriority)
+
+				admission := testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f2", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq2HighPriority, admission)
+
+				admission = testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f1", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq1LowPriority, admission)
+			}
+
+			cq2MiddlePriority := createWorkloadWithPriority("cq2", "1", 105)
+			{
+				util.ExpectWorkloadsToBePreempted(ctx, k8sClient, cq1LowPriority)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, cq1LowPriority)
+
+				admission := testing.MakeAdmission("cq2").Assignment(corev1.ResourceCPU, "f1", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq2MiddlePriority, admission)
+			}
+		})
+		ginkgo.It("chooses a correct flavor when borrowing is preferred", func() {
+			fungibility := kueue.FlavorFungibility{
+				WhenCanBorrow:  kueue.Borrow,
+				WhenCanPreempt: kueue.TryNextFlavor}
+			preemption := kueue.ClusterQueuePreemption{WithinClusterQueue: kueue.PreemptionPolicyLowerPriority, ReclaimWithinCohort: kueue.PreemptionPolicyAny, BorrowWithinCohort: &kueue.BorrowWithinCohort{Policy: kueue.BorrowWithinCohortPolicyLowerPriority}}
+
+			createQueue(testing.MakeClusterQueue("cq1").
+				FlavorFungibility(fungibility).Cohort("cohort").
+				Preemption(preemption).
+				ResourceGroup(testing.MakeFlavorQuotas("f1").Resource(corev1.ResourceCPU, "1").FlavorQuotas, testing.MakeFlavorQuotas("f2").Resource(corev1.ResourceCPU, "0").FlavorQuotas).Obj())
+
+			createQueue(testing.MakeClusterQueue("cq2").Cohort("cohort").
+				FlavorFungibility(fungibility).
+				Preemption(preemption).
+				ResourceGroup(testing.MakeFlavorQuotas("f1").Resource(corev1.ResourceCPU, "0").FlavorQuotas, testing.MakeFlavorQuotas("f2").Resource(corev1.ResourceCPU, "1").FlavorQuotas).Obj())
+
+			cq1LowPriority := createWorkloadWithPriority("cq1", "1", 0)
+			{
+				admission := testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f1", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq1LowPriority, admission)
+			}
+
+			cq2HighPriority := createWorkloadWithPriority("cq1", "1", 9999)
+			{
+				admission := testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f2", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq2HighPriority, admission)
+			}
+
+			cq2MiddlePriority := createWorkloadWithPriority("cq2", "1", 105)
+			{
+				util.ExpectWorkloadsToBePreempted(ctx, k8sClient, cq2HighPriority)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, cq2HighPriority)
+
+				admission := testing.MakeAdmission("cq2").Assignment(corev1.ResourceCPU, "f2", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq2MiddlePriority, admission)
+			}
+
+			{
+				util.ExpectWorkloadsToBePreempted(ctx, k8sClient, cq1LowPriority)
+				util.FinishEvictionForWorkloads(ctx, k8sClient, cq1LowPriority)
+
+				admission := testing.MakeAdmission("cq1").Assignment(corev1.ResourceCPU, "f1", "1").Obj()
+				util.ExpectWorkloadToBeAdmittedAs(ctx, k8sClient, cq2HighPriority, admission)
+			}
+		})
+	})
 	ginkgo.When("Deleting ClusterQueue should update cohort borrowable resources", func() {
 		ginkgo.It("Should prevent incorrect admission through borrowing after ClusterQueue deletion", func() {
 			ginkgo.By("Creating two ClusterQueues in the same cohort")