util/sort: Add MergePreservingRelativeOrder for topological sorting

Add a topological sort implementation that merges multiple ordered lists
while preserving their relative ordering. This is used by peer-aggregated
discovery to deterministically merge discovery documents from multiple
API servers.

Part of KEP-4020: Unknown Version Interoperability Proxy

staging/src/k8s.io/apimachinery/pkg/util/sort/sort.go

@@ -0,0 +1,191 @@
+/*
+Copyright 2025 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 sort
+
+import (
+	"container/heap"
+	"fmt"
+	"sort"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+// MergePreservingRelativeOrder performs a topological consensus sort of items from multiple sources.
+// It merges multiple lists of strings into a single list, preserving the relative order of
+// elements within each source list.
+//
+// For any two items, if one appears before the other in any of the input lists,
+// that relative order will be preserved in the output. If no relative ordering is
+// defined between two items, they are sorted lexicographically.
+//
+// The function uses Kahn's algorithm for topological sorting with a min-heap to ensure
+// deterministic output. Items with no dependencies are processed in lexicographic order,
+// guaranteeing consistent results across multiple invocations with the same input.
+//
+// This function contains a shortcut optimization that returns an input list directly
+// if it already contains all unique items. This provides O(n) performance in the best case.
+//
+// Example:
+//   - Input: {{"a", "b", "c"}, {"b", "c"}} returns {"a", "b", "c"}
+//   - Input: {{"a", "c"}, {"b", "c"}} returns {"a", "b", "c"} (lexicographic tie-breaking)
+//   - Input: {{"a", "b"}, {"b", "a"}} returns error (cycle detected)
+//
+// Complexity: O(L*n + V*log(V) + E) where L is the number of lists, n is the average
+// list size, V is the number of unique items, and E is the number of precedence edges.
+//
+// This is useful for creating a stable, consistent ordering when merging data from
+// multiple sources that may have partial but not conflicting orderings.
+func MergePreservingRelativeOrder(inputLists [][]string) []string {
+	if len(inputLists) == 0 {
+		return nil
+	}
+
+	// Build a directed graph of precedence relationships
+	graph := make(map[string]*graphNode)
+	for _, list := range inputLists {
+		for i, item := range list {
+			node := getOrCreateNode(graph, item)
+
+			// Add edge from current item to next item in list
+			if i < len(list)-1 {
+				nextItem := list[i+1]
+				nextNode := getOrCreateNode(graph, nextItem)
+
+				// Only add edge if not already present (avoid incrementing in-degree multiple times)
+				if !node.outEdges.Has(nextItem) {
+					node.outEdges.Insert(nextItem)
+					nextNode.inDegree++
+				}
+			}
+		}
+	}
+
+	// Shortcut: if any input list contains all items (no duplicates), use it
+	allItems := sets.New[string]()
+	for name := range graph {
+		allItems.Insert(name)
+	}
+	for _, list := range inputLists {
+		if len(list) == allItems.Len() && isUnique(list) {
+			return list
+		}
+	}
+
+	// Perform topological sort using Kahn's algorithm with min-heap for determinism
+	result, err := topologicalSort(graph)
+	if err != nil {
+		// This should not happen with valid input, but if it does,
+		// fall back to lexicographic sort to provide some result
+		items := make([]string, 0, len(graph))
+		for name := range graph {
+			items = append(items, name)
+		}
+		sort.Strings(items)
+		return items
+	}
+
+	return result
+}
+
+// getOrCreateNode retrieves or creates a graph node for the given name
+func getOrCreateNode(graph map[string]*graphNode, name string) *graphNode {
+	if graph[name] == nil {
+		graph[name] = &graphNode{
+			outEdges: sets.New[string](),
+			inDegree: 0,
+		}
+	}
+	return graph[name]
+}
+
+// isUnique checks if a list contains no duplicate items
+func isUnique(list []string) bool {
+	seen := make(map[string]bool, len(list))
+	for _, item := range list {
+		if seen[item] {
+			return false
+		}
+		seen[item] = true
+	}
+	return true
+}
+
+// topologicalSort performs Kahn's algorithm with a min-heap for deterministic ordering
+func topologicalSort(graph map[string]*graphNode) ([]string, error) {
+	// Initialize min-heap with all nodes that have no incoming edges
+	pq := &stringMinHeap{}
+	heap.Init(pq)
+
+	for name, node := range graph {
+		if node.inDegree == 0 {
+			heap.Push(pq, name)
+		}
+	}
+
+	result := make([]string, 0, len(graph))
+
+	for pq.Len() > 0 {
+		// Pop item with lowest lexicographic value
+		current := heap.Pop(pq).(string)
+		result = append(result, current)
+
+		currentNode := graph[current]
+
+		// Reduce in-degree for all neighbors
+		for neighbor := range currentNode.outEdges {
+			neighborNode := graph[neighbor]
+			neighborNode.inDegree--
+
+			// If in-degree becomes 0, add to heap
+			if neighborNode.inDegree == 0 {
+				heap.Push(pq, neighbor)
+			}
+		}
+	}
+
+	// Check for cycles
+	if len(result) != len(graph) {
+		return nil, fmt.Errorf("cycle detected in precedence graph: sorted %d items but graph has %d items", len(result), len(graph))
+	}
+
+	return result, nil
+}
+
+// graphNode represents a node in the precedence graph
+type graphNode struct {
+	// Items that should come after this item
+	outEdges sets.Set[string]
+	// Number of items that should come before this item
+	inDegree int
+}
+
+// stringMinHeap implements heap.Interface for strings (min-heap with lexicographic ordering)
+type stringMinHeap []string
+
+func (h stringMinHeap) Len() int           { return len(h) }
+func (h stringMinHeap) Less(i, j int) bool { return h[i] < h[j] }
+func (h stringMinHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+func (h *stringMinHeap) Push(x interface{}) {
+	*h = append(*h, x.(string))
+}
+func (h *stringMinHeap) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}

staging/src/k8s.io/apimachinery/pkg/util/sort/sort_test.go

@@ -0,0 +1,148 @@
+/*
+Copyright 2025 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 sort
+
+import (
+	"testing"
+)
+
+func TestSortDiscoveryGroupsTopo(t *testing.T) {
+	cases := []struct {
+		name  string
+		input [][]string
+		want  []string
+	}{
+		{
+			name: "consensus ordering",
+			input: [][]string{
+				{"A", "B", "C", "D"},
+				{"A", "B", "C", "D"},
+				{"A", "X", "Z", "D"},
+				{"Z", "Y"},
+				{"Q"},
+			},
+			want: []string{"A", "B", "C", "Q", "X", "Z", "D", "Y"},
+		},
+		{
+			name:  "empty input",
+			input: [][]string{},
+			want:  []string{},
+		},
+		{
+			name:  "single peer",
+			input: [][]string{{"foo", "bar", "baz"}},
+			want:  []string{"foo", "bar", "baz"},
+		},
+		{
+			name:  "conflicting orderings",
+			input: [][]string{{"A", "B"}, {"B", "A"}},
+			want:  []string{"A", "B"},
+		},
+		{
+			name:  "empty list merged with non-empty list",
+			input: [][]string{{}, {"A", "B", "C"}},
+			want:  []string{"A", "B", "C"},
+		},
+		{
+			name:  "multiple empty lists merged",
+			input: [][]string{{}, {}, {}},
+			want:  []string{},
+		},
+		{
+			name: "lexical tiebreak at beginning",
+			input: [][]string{
+				{"C", "D", "E"},
+				{"B", "D", "E"},
+				{"A", "D", "E"},
+			},
+			// A, B, C have no precedence constraints, so lexical order
+			want: []string{"A", "B", "C", "D", "E"},
+		},
+		{
+			name: "lexical tiebreak in middle",
+			input: [][]string{
+				{"A", "D", "E"},
+				{"A", "C", "E"},
+				{"A", "B", "E"},
+			},
+			// A comes first (consensus), then B, C, D (lexical), then E (consensus)
+			want: []string{"A", "B", "C", "D", "E"},
+		},
+		{
+			name: "conflicting orderings of 3 lists",
+			input: [][]string{
+				{"A", "B", "C"},
+				{"B", "C", "A"},
+				{"C", "A", "B"},
+			},
+			// Creates cycle: A->B, B->C, C->A
+			// Fallback to lexicographic sort
+			want: []string{"A", "B", "C"},
+		},
+		{
+			name: "conflicting ordering with different list lengths",
+			input: [][]string{
+				{"A", "B", "C", "D"},
+				{"B", "A"},
+				{"C", "D"},
+			},
+			// A->B->C->D from first list, but B->A from second
+			// Creates cycle between A and B
+			// Fallback to lexicographic sort
+			want: []string{"A", "B", "C", "D"},
+		},
+		{
+			name: "conflicting partial lists",
+			input: [][]string{
+				{"A", "B"},
+				{"C", "D"},
+				{"B", "A"},
+			},
+			// A->B from first, B->A from third (cycle)
+			// C->D is independent
+			// Fallback to lexicographic sort
+			want: []string{"A", "B", "C", "D"},
+		},
+		{
+			name: "cycle",
+			input: [][]string{
+				{"A", "B"},
+				{"B", "C"},
+				{"C", "A"},
+			},
+			// Creates cycle: A->B->C->A
+			// Fallback to lexicographic sort
+			want: []string{"A", "B", "C"},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := MergePreservingRelativeOrder(tc.input)
+			if len(got) != len(tc.want) {
+				t.Errorf("length mismatch:\n  got: %d\n  want: %d", len(got), len(tc.want))
+				return
+			}
+			for i := range got {
+				if got[i] != tc.want[i] {
+					t.Errorf("mismatch got: %v\n  want: %v", got, tc.want)
+					return
+				}
+			}
+		})
+	}
+}