fix(client): make SF recovery segment sort O(N log N) unconditionally

sortByBaseSeq was a median-of-three Lomuto quicksort. Median-of-three
covers the readdir orders a healthy slot produces (lexicographic
enumeration yields already-sorted baseSeqs, hashed directory order is
effectively random), but Lomuto partitioning is O(N^2) on organ-pipe,
duplicate-heavy and median-of-three-killer orders. Exact simulation at
the documented 16K-segment ceiling: 22.6M comparisons for organ-pipe,
50.7M for Musser's med3-killer permutation, 134M (full N^2/2) for
mass-duplicate baseSeqs -- versus ~221K on the healthy paths. Such
orders are only reachable through corrupted-yet-parseable or
operator-copied headers, and recovery validation rejects those slots --
but the quadratic stall lands BEFORE validation gets to reject them.

The sort is now an introsort: the median-of-three fast path is
unchanged, and each root-to-leaf partition path carries a budget of
2*floor(log2(N)) passes. Loop-on-larger iterations count against the
budget, so bad-split chains cannot hide in the loop. Ranges that
exhaust it fall back to in-place heapsort (still Long.compareUnsigned,
zero allocation), capping every adversarial pattern at ~3.8*N*log2(N)
comparisons (organ-pipe 773K, duplicates 507K, med3-killer 861K at
N=16384). Recursion depth stays under log2(N).

The sortComparisons counter ticks +2 per sift-down level so it remains
a strict upper bound in the fallback. New adversarial test drives the
sort directly with in-memory segments at N=16384 across organ-pipe,
all-duplicate, few-distinct, med3-killer and sign-bit-key patterns,
asserting unsigned order plus comparisons < 8*N*log2(N) -- >2x headroom
over the worst measured pattern, ~12x below the mildest quadratic
blow-up. The existing sorted-input regression test is unchanged and
still passes.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentRing.java

@@ -71,8 +71,9 @@ public final class SegmentRing implements QuietCloseable {
     // openExisting() recovery on this JVM. Used by SegmentRingTest to
     // assert the sort stays O(N log N) without relying on wall-clock time
     // (CI runner variance makes elapsed-millisecond bounds flaky). Cheap
-    // in production: one volatile-free add per partition pass, dwarfed by
-    // the mmap I/O the recovery does on every segment.
+    // in production: one volatile-free add per partition pass (plus two per
+    // sift level in the rare heapsort fallback), dwarfed by the mmap I/O
+    // the recovery does on every segment.
     private static long sortComparisons;
     // References copied while compacting the logical sealed-segment head.
     // Test-only operation count for deterministic trim-complexity assertions.
@@ -1110,9 +1111,10 @@ public static long getNextSealedComparisons() {
      * {@link #sortByBaseSeq} since the last {@link #resetSortComparisons()}
      * (or process start). The count is incremented once per partition pass
      * for the median-of-three pivot pick plus once per element compared
-     * against the pivot, so a clean run on N segments adds roughly
-     * {@code 3 + (hi - lo - 1)} per recursive frame, summing to O(N log N).
-     * Exposed for {@code SegmentRingTest} to detect O(N²) regressions
+     * against the pivot ({@code 3 + (hi - lo - 1)} per pass), and by two per
+     * sift-down level when a range falls back to heapsort, so it strictly
+     * upper-bounds the true compare count and sums to O(N log N) on every
+     * input. Exposed for {@code SegmentRingTest} to detect O(N²) regressions
      * deterministically.
      */
     @TestOnly
@@ -1145,15 +1147,46 @@ public static void resetTrimMovedReferences() {
     }
 
     /**
-     * In-place quicksort over {@code list[lo, hi)} keyed by ascending
-     * {@code baseSeq}. Median-of-three pivot avoids the pathological O(N²)
-     * on already-sorted input that lexicographic readdir produces (our
-     * filenames are zero-padded hex of {@code baseSeq}). Recursion depth is
-     * bounded by ~2 log₂(N) -- for the documented 16K-segment ceiling, well
-     * under the JVM default stack.
+     * Drives the recovery-time baseSeq sort over the whole list. Exposed so
+     * {@code SegmentRingTest} can feed adversarial orders (organ-pipe, mass
+     * duplicates, median-of-three killer, unsigned-boundary keys) straight
+     * into the sort and assert comparison bounds without staging thousands
+     * of segment files on disk.
+     */
+    @TestOnly
+    public static void sortByBaseSeqForTest(ObjList<MmapSegment> list) {
+        sortByBaseSeq(list, 0, list.size());
+    }
+
+    /**
+     * In-place introsort over {@code list[lo, hi)} keyed by ascending
+     * unsigned {@code baseSeq}. Median-of-three quicksort handles the readdir
+     * orders a healthy slot produces (lexicographic enumeration of the
+     * generation-numbered filenames yields already-sorted baseSeqs; hashed
+     * directory order is effectively random), and a partition-pass budget of
+     * 2·⌊log₂(N)⌋ demotes any range that keeps splitting badly to in-place
+     * heapsort. Without that budget, Lomuto with a median-of-three pivot is
+     * O(N²) on organ-pipe, duplicate-heavy and median-of-three-killer orders
+     * -- reachable only through corrupted-yet-parseable or operator-copied
+     * headers, but at the documented 16K-segment ceiling that is 10⁷..10⁸
+     * comparisons of startup stall before recovery validation gets to
+     * reject the slot, so the fallback makes O(N log N) unconditional.
+     * Recursion depth stays under log₂(N) (recurse on the smaller side,
+     * loop on the larger), well within the JVM default stack.
      */
     private static void sortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi) {
+        int n = hi - lo;
+        if (n > 1) {
+            sortByBaseSeq(list, lo, hi, 2 * (31 - Integer.numberOfLeadingZeros(n)));
+        }
+    }
+
+    private static void sortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi, int budget) {
         while (hi - lo > 1) {
+            if (budget-- == 0) {
+                heapSortByBaseSeq(list, lo, hi);
+                return;
+            }
             int mid = (lo + hi) >>> 1;
             long a = list.get(lo).baseSeq();
             long b = list.get(mid).baseSeq();
@@ -1183,17 +1216,61 @@ private static void sortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi) {
             }
             swap(list, store, hi - 1);
             // Recurse on the smaller partition; loop on the larger to keep
-            // recursion depth bounded by log₂(N).
+            // recursion depth bounded by log₂(N). Children inherit the
+            // remaining pass budget: it counts passes along a root-to-leaf
+            // path, so a chain of bad splits exhausts it after ~2 log₂(N)
+            // levels no matter how the work is divided.
             if (store - lo < hi - store - 1) {
-                sortByBaseSeq(list, lo, store);
+                sortByBaseSeq(list, lo, store, budget);
                 lo = store + 1;
             } else {
-                sortByBaseSeq(list, store + 1, hi);
+                sortByBaseSeq(list, store + 1, hi, budget);
                 hi = store;
             }
         }
     }
 
+    /**
+     * In-place heapsort over {@code list[lo, hi)} keyed by ascending unsigned
+     * {@code baseSeq}: the introsort fallback for ranges whose partition-pass
+     * budget ran out. Guaranteed O(N log N) for any key distribution and any
+     * initial order; no allocation.
+     */
+    private static void heapSortByBaseSeq(ObjList<MmapSegment> list, int lo, int hi) {
+        int n = hi - lo;
+        for (int root = (n >>> 1) - 1; root >= 0; root--) {
+            siftDownByBaseSeq(list, lo, root, n);
+        }
+        for (int end = n - 1; end > 0; end--) {
+            swap(list, lo, lo + end);
+            siftDownByBaseSeq(list, lo, 0, end);
+        }
+    }
+
+    private static void siftDownByBaseSeq(ObjList<MmapSegment> list, int lo, int root, int heapSize) {
+        while (true) {
+            int child = (root << 1) + 1;
+            if (child >= heapSize) {
+                return;
+            }
+            // At most two unsigned compares per level (sibling pick + parent
+            // test); bump the counter by the constant 2 up front -- same
+            // cheap-upper-bound convention as the partition pass.
+            sortComparisons += 2;
+            if (child + 1 < heapSize
+                    && Long.compareUnsigned(list.get(lo + child).baseSeq(),
+                    list.get(lo + child + 1).baseSeq()) < 0) {
+                child++;
+            }
+            if (Long.compareUnsigned(list.get(lo + root).baseSeq(),
+                    list.get(lo + child).baseSeq()) >= 0) {
+                return;
+            }
+            swap(list, lo + root, lo + child);
+            root = child;
+        }
+    }
+
     private static void swap(ObjList<MmapSegment> list, int i, int j) {
         if (i == j) return;
         MmapSegment tmp = list.get(i);

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentRingTest.java

@@ -686,6 +686,101 @@ public void testLargeSegmentCountReopensInOrder() throws Exception {
         });
     }
 
+    /**
+     * Adversarial-order companion to
+     * {@link #testLargeSegmentCountReopensInOrder}: that test covers the
+     * already-sorted readdir order median-of-three was chosen for; this one
+     * covers the orders median-of-three does NOT defend. On the
+     * pre-introsort quicksort, exact simulation at N=16384 measured ~22.6M
+     * comparisons for organ-pipe, ~134M (full N²/2) for mass-duplicate
+     * baseSeqs and ~50.7M for Musser's median-of-three-killer permutation.
+     * The heapsort fallback caps the worst of these at ~3.8·N·log₂(N)
+     * (~860K), so the 8·N·log₂(N) bound below keeps >2x headroom against
+     * harmless implementation drift while sitting ~26x under the mildest
+     * quadratic blow-up. Also pins unsigned key ordering: baseSeqs with the
+     * sign bit set must sort above {@code Long.MAX_VALUE}, not below zero.
+     * <p>
+     * A healthy client cannot produce these orders (recovery validation
+     * rejects duplicate or non-contiguous baseSeqs moments after the sort),
+     * but corrupted-yet-parseable headers or operator file copies feed the
+     * sort BEFORE validation runs, so the sort itself must stay log-linear.
+     * Sorts in-memory segments directly: staging 16K adversarial header
+     * files on disk per pattern would dominate the test's runtime without
+     * adding coverage.
+     */
+    @Test
+    public void testAdversarialSegmentOrdersSortLogLinear() throws Exception {
+        TestUtils.assertMemoryLeak(() -> {
+            final int n = 16384;
+            final long bound = 8L * n * (long) (Math.log(n) / Math.log(2));
+
+            // Organ pipe: 0,1,...,n/2-1,n/2-1,...,1,0.
+            long[] organPipe = new long[n];
+            for (int i = 0; i < n / 2; i++) {
+                organPipe[i] = i;
+                organPipe[n - 1 - i] = i;
+            }
+            assertAdversarialSortWithinBound("organ-pipe", organPipe, bound);
+
+            long[] allDuplicates = new long[n];
+            for (int i = 0; i < n; i++) {
+                allDuplicates[i] = 42;
+            }
+            assertAdversarialSortWithinBound("all-duplicates", allDuplicates, bound);
+
+            long[] fewDistinct = new long[n];
+            for (int i = 0; i < n; i++) {
+                fewDistinct[i] = i % 4;
+            }
+            assertAdversarialSortWithinBound("few-distinct", fewDistinct, bound);
+
+            // Musser's median-of-three killer permutation of 1..n.
+            long[] med3Killer = new long[n];
+            int half = n / 2;
+            for (int i = 1; i <= half; i++) {
+                if (i % 2 == 1) {
+                    med3Killer[i - 1] = i;
+                    med3Killer[i] = half + i;
+                }
+                med3Killer[half + i - 1] = 2L * i;
+            }
+            assertAdversarialSortWithinBound("median-of-three-killer", med3Killer, bound);
+
+            // High-bit keys interleaved with small ones: exercises the
+            // unsigned comparison contract alongside the comparison bound.
+            long[] unsignedMix = new long[n];
+            for (int i = 0; i < n; i++) {
+                unsignedMix[i] = (i % 2 == 0) ? (0x8000000000000000L | i) : i;
+            }
+            assertAdversarialSortWithinBound("unsigned-mix", unsignedMix, bound);
+        });
+    }
+
+    private static void assertAdversarialSortWithinBound(String label, long[] baseSeqs, long bound) {
+        final long segSize = MmapSegment.HEADER_SIZE + MmapSegment.FRAME_HEADER_SIZE + 1;
+        ObjList<MmapSegment> list = new ObjList<>();
+        try {
+            for (long baseSeq : baseSeqs) {
+                list.add(MmapSegment.createInMemory(baseSeq, segSize));
+            }
+            SegmentRing.resetSortComparisons();
+            SegmentRing.sortByBaseSeqForTest(list);
+            long comparisons = SegmentRing.getSortComparisons();
+            for (int i = 1, size = list.size(); i < size; i++) {
+                assertTrue(label + ": unsigned baseSeq order violated at index " + i,
+                        Long.compareUnsigned(list.get(i - 1).baseSeq(), list.get(i).baseSeq()) <= 0);
+            }
+            assertTrue(label + " sort took " + comparisons + " comparisons (expected < " + bound
+                            + " = 8 * N * log2(N) for N=" + baseSeqs.length
+                            + "); regression suggests the introsort heapsort fallback stopped engaging",
+                    comparisons < bound);
+        } finally {
+            for (int i = 0, size = list.size(); i < size; i++) {
+                list.get(i).close();
+            }
+        }
+    }
+
     @Test
     public void testRemovingAcknowledgedPrefixMovesLinearReferences() throws Exception {
         TestUtils.assertMemoryLeak(() -> {