Fix DataFrame.OrderBy to perform stable sorting

Replace the unstable IntrospectiveSort (introsort) with a stable merge sort
implementation for per-buffer sorting in both PrimitiveDataFrameColumn and
StringDataFrameColumn. Also fix the multi-buffer merge phase to always write
left-to-right using a direction-aware comparer, ensuring stability for both
ascending and descending sorts.

Fixes #6443

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: kubaflo

src/Microsoft.Data.Analysis/DataFrameColumn.cs

@@ -469,16 +469,15 @@ public string ToString(long rowsToShow)
         protected static void PopulateColumnSortIndicesWithHeap<T>(SortedDictionary<T, List<ValueTuple<int, int>>> heapOfValueAndListOfTupleOfSortAndBufferIndex,
                                                             PrimitiveDataFrameColumn<long> columnSortIndices,
                                                             PrimitiveDataFrameColumn<long> columnNullIndices,
-                                                            bool ascending,
                                                             bool putNullValuesLast,
                                                             GetBufferSortIndex getBufferSortIndex,
                                                             GetValueAndBufferSortIndexAtBuffer<T> getValueAndBufferSortIndexAtBuffer,
                                                             GetBufferLengthAtIndex getBufferLengthAtIndex)
         {
-            long i = ascending ? columnNullIndices.Length : columnSortIndices.Length - 1;
-
-            if (putNullValuesLast)
-                i -= columnNullIndices.Length;
+            // Always write left-to-right. For descending order, callers pass a SortedDictionary
+            // with a reversed comparer so that ElementAt(0) yields the largest value first.
+            // This ensures stable sorting: equal elements preserve their original relative order.
+            long i = putNullValuesLast ? 0 : columnNullIndices.Length;
 
             while (heapOfValueAndListOfTupleOfSortAndBufferIndex.Count > 0)
             {
@@ -499,7 +498,7 @@ protected static void PopulateColumnSortIndicesWithHeap<T>(SortedDictionary<T, L
                 int bufferIndex = sortAndBufferIndex.bufferIndex;
                 long bufferSortIndex = getBufferSortIndex(bufferIndex, sortIndex);
 
-                columnSortIndices[ascending ? i++ : i--] = bufferSortIndex;
+                columnSortIndices[i++] = bufferSortIndex;
 
                 if (sortIndex + 1 < getBufferLengthAtIndex(bufferIndex))
                 {
@@ -758,5 +757,163 @@ private static void Sort2<TKey>(
                 sortIndices[j] = temp;
             }
         }
+
+        /// <summary>
+        /// Stable merge sort that reorders <paramref name="sortIndices"/> so that
+        /// the values in <paramref name="span"/> are visited in sorted order.
+        /// </summary>
+        protected static void MergeSortIndices<TKey>(
+            ReadOnlySpan<TKey> span,
+            int length,
+            Span<int> sortIndices,
+            IComparer<TKey> comparer)
+        {
+            if (length <= 1)
+                return;
+
+            int[] auxiliary = System.Buffers.ArrayPool<int>.Shared.Rent(length);
+            try
+            {
+                MergeSortIndicesRecursive(span, sortIndices, auxiliary, 0, length - 1, comparer);
+            }
+            finally
+            {
+                System.Buffers.ArrayPool<int>.Shared.Return(auxiliary);
+            }
+        }
+
+        /// <summary>
+        /// Stable merge sort overload that accepts an <see cref="IList{TKey}"/> instead of a span,
+        /// avoiding the need to copy to an array first.
+        /// </summary>
+        protected static void MergeSortIndices<TKey>(
+            IList<TKey> list,
+            int length,
+            Span<int> sortIndices,
+            IComparer<TKey> comparer)
+        {
+            if (length <= 1)
+                return;
+
+            int[] auxiliary = System.Buffers.ArrayPool<int>.Shared.Rent(length);
+            try
+            {
+                MergeSortIndicesRecursive(list, sortIndices, auxiliary, 0, length - 1, comparer);
+            }
+            finally
+            {
+                System.Buffers.ArrayPool<int>.Shared.Return(auxiliary);
+            }
+        }
+
+        private static void MergeSortIndicesRecursive<TKey>(
+            ReadOnlySpan<TKey> span,
+            Span<int> sortIndices,
+            int[] auxiliary,
+            int lo,
+            int hi,
+            IComparer<TKey> comparer)
+        {
+            if (lo >= hi)
+                return;
+
+            int mid = lo + (hi - lo) / 2;
+            MergeSortIndicesRecursive(span, sortIndices, auxiliary, lo, mid, comparer);
+            MergeSortIndicesRecursive(span, sortIndices, auxiliary, mid + 1, hi, comparer);
+            Merge(span, sortIndices, auxiliary, lo, mid, hi, comparer);
+        }
+
+        private static void MergeSortIndicesRecursive<TKey>(
+            IList<TKey> list,
+            Span<int> sortIndices,
+            int[] auxiliary,
+            int lo,
+            int hi,
+            IComparer<TKey> comparer)
+        {
+            if (lo >= hi)
+                return;
+
+            int mid = lo + (hi - lo) / 2;
+            MergeSortIndicesRecursive(list, sortIndices, auxiliary, lo, mid, comparer);
+            MergeSortIndicesRecursive(list, sortIndices, auxiliary, mid + 1, hi, comparer);
+            MergeList(list, sortIndices, auxiliary, lo, mid, hi, comparer);
+        }
+
+        private static void Merge<TKey>(
+            ReadOnlySpan<TKey> span,
+            Span<int> sortIndices,
+            int[] auxiliary,
+            int lo,
+            int mid,
+            int hi,
+            IComparer<TKey> comparer)
+        {
+            for (int k = lo; k <= hi; k++)
+            {
+                auxiliary[k] = sortIndices[k];
+            }
+
+            int i = lo;
+            int j = mid + 1;
+
+            for (int k = lo; k <= hi; k++)
+            {
+                if (i > mid)
+                {
+                    sortIndices[k] = auxiliary[j++];
+                }
+                else if (j > hi)
+                {
+                    sortIndices[k] = auxiliary[i++];
+                }
+                else if (comparer.Compare(span[auxiliary[i]], span[auxiliary[j]]) <= 0)
+                {
+                    sortIndices[k] = auxiliary[i++];
+                }
+                else
+                {
+                    sortIndices[k] = auxiliary[j++];
+                }
+            }
+        }
+
+        private static void MergeList<TKey>(
+            IList<TKey> list,
+            Span<int> sortIndices,
+            int[] auxiliary,
+            int lo,
+            int mid,
+            int hi,
+            IComparer<TKey> comparer)
+        {
+            for (int k = lo; k <= hi; k++)
+            {
+                auxiliary[k] = sortIndices[k];
+            }
+
+            int i = lo;
+            int j = mid + 1;
+
+            for (int k = lo; k <= hi; k++)
+            {
+                if (i > mid)
+                {
+                    sortIndices[k] = auxiliary[j++];
+                }
+                else if (j > hi)
+                {
+                    sortIndices[k] = auxiliary[i++];
+                }
+                else if (comparer.Compare(list[auxiliary[i]], list[auxiliary[j]]) <= 0)
+                {
+                    sortIndices[k] = auxiliary[i++];
+                }
+                else
+                {
+                    sortIndices[k] = auxiliary[j++];
+                }
+            }
+        }
     }
 }

src/Microsoft.Data.Analysis/DataFrameColumns/StringDataFrameColumn.cs

@@ -205,6 +205,7 @@ public IEnumerator<string> GetEnumerator()
         protected internal override PrimitiveDataFrameColumn<long> GetSortIndices(bool ascending, bool putNullValuesLast)
         {
             var comparer = Comparer<string>.Default;
+            IComparer<string> sortComparer = ascending ? comparer : Comparer<string>.Create((a, b) => comparer.Compare(b, a));
 
             List<int[]> bufferSortIndices = new List<int[]>(_stringBuffers.Count);
             var columnNullIndices = new Int64DataFrameColumn("NullIndices", NullCount);
@@ -221,9 +222,7 @@ protected internal override PrimitiveDataFrameColumn<long> GetSortIndices(bool a
                         nullIndicesSlot++;
                     }
                 }
-                // TODO: Refactor the sort routine to also work with IList?
-                string[] array = buffer.ToArray();
-                IntrospectiveSort(array, array.Length, sortIndices, comparer);
+                MergeSortIndices(buffer, buffer.Count, sortIndices, sortComparer);
                 bufferSortIndices.Add(sortIndices);
             }
             // Simple merge sort to build the full column's sort indices
@@ -237,7 +236,7 @@ ValueTuple<string, int> GetFirstNonNullValueStartingAtIndex(int stringBufferInde
                 return (value, startIndex);
             }
 
-            SortedDictionary<string, List<ValueTuple<int, int>>> heapOfValueAndListOfTupleOfSortAndBufferIndex = new SortedDictionary<string, List<ValueTuple<int, int>>>(comparer);
+            SortedDictionary<string, List<ValueTuple<int, int>>> heapOfValueAndListOfTupleOfSortAndBufferIndex = new SortedDictionary<string, List<ValueTuple<int, int>>>(sortComparer);
             List<List<string>> buffers = _stringBuffers;
             for (int i = 0; i < buffers.Count; i++)
             {
@@ -266,7 +265,6 @@ ValueTuple<string, int> GetFirstNonNullValueStartingAtIndex(int stringBufferInde
             PopulateColumnSortIndicesWithHeap(heapOfValueAndListOfTupleOfSortAndBufferIndex,
                 columnSortIndices,
                 columnNullIndices,
-                ascending,
                 putNullValuesLast,
                 getBufferSortIndex,
                 getValueAtBuffer,

src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.Sort.cs

@@ -21,6 +21,7 @@ public partial class PrimitiveDataFrameColumn<T> : DataFrameColumn
         protected internal override PrimitiveDataFrameColumn<long> GetSortIndices(bool ascending = true, bool putNullValuesLast = true)
         {
             var comparer = Comparer<T>.Default;
+            IComparer<T> sortComparer = ascending ? comparer : Comparer<T>.Create((a, b) => comparer.Compare(b, a));
 
             List<List<int>> bufferSortIndices = new List<List<int>>(_columnContainer.Buffers.Count);
             var columnNullIndices = new Int64DataFrameColumn("NullIndices", NullCount);
@@ -35,8 +36,8 @@ protected internal override PrimitiveDataFrameColumn<long> GetSortIndices(bool a
                 {
                     sortIndices[i] = i;
                 }
-                IntrospectiveSort(buffer.ReadOnlySpan, buffer.Length, sortIndices, comparer);
-                // Bug fix: QuickSort is not stable. When PrimitiveDataFrameColumn has null values and default values, they move around
+                MergeSortIndices(buffer.ReadOnlySpan, buffer.Length, sortIndices, sortComparer);
+                // Stable sort preserves relative order of equal elements and null/default values
                 List<int> nonNullSortIndices = new List<int>();
                 for (int i = 0; i < sortIndices.Length; i++)
                 {
@@ -78,7 +79,7 @@ ValueTuple<T, int> GetFirstNonNullValueAndBufferIndexStartingAtIndex(int bufferI
                 return (value, startIndex);
             }
 
-            SortedDictionary<T, List<ValueTuple<int, int>>> heapOfValueAndListOfTupleOfSortAndBufferIndex = new SortedDictionary<T, List<ValueTuple<int, int>>>(comparer);
+            SortedDictionary<T, List<ValueTuple<int, int>>> heapOfValueAndListOfTupleOfSortAndBufferIndex = new SortedDictionary<T, List<ValueTuple<int, int>>>(sortComparer);
             IList<ReadOnlyDataFrameBuffer<T>> buffers = _columnContainer.Buffers;
             for (int i = 0; i < buffers.Count; i++)
             {
@@ -107,7 +108,6 @@ ValueTuple<T, int> GetFirstNonNullValueAndBufferIndexStartingAtIndex(int bufferI
             PopulateColumnSortIndicesWithHeap(heapOfValueAndListOfTupleOfSortAndBufferIndex,
                 columnSortIndices,
                 columnNullIndices,
-                ascending,
                 putNullValuesLast,
                 getBufferSortIndex,
                 getValueAndBufferSortIndexAtBuffer,