Fix query engine edge cases, NT scaling regressions, and add JIT warmup

Edge cases fixed:
- E1: Multi-column dict-encoded join key unification (silent wrong results)
- E2: HAVING alias fallback uses last-index-of to handle underscored names
- E3: OR/IN predicates only materialize referenced columns, not all

Code quality (O1-O10):
- Extract shared column helpers into query/columns.clj (DRY)
- Extract partition-sizes/boundaries helpers in window.clj
- Extract date-trunc/date-add helpers in expression.clj
- Remove dead bindings, stale comments, NTILE dead code
- Add ColumnOpsVar.java for JIT-isolated VARIANCE/CORR group-by

NT scaling fixes:
- Window functions respect PARALLEL_THRESHOLD with avgPartSize<8 guard
- LIKE broadcast loop made sequential (FJP overhead dominated)
- Expression pre-computation moved inside group-by fallback path
- Benchmark WIN/DS execution order swapped for fair JIT warmup

JIT warmup for production safety:
- q/jit-warmup! exercises all 11 Java code paths with tiny data (0.2s)
- Called on server startup and before benchmark tiers
- Prevents deoptimization cliffs from mixed analytical workloads

Author: whilo

CLAUDE.md

@@ -56,6 +56,7 @@ After modifying Java files in `src-java/`:
 javac --add-modules jdk.incubator.vector -d target/classes \
   src-java/stratum/internal/ColumnOps.java \
   src-java/stratum/internal/ColumnOpsExt.java \
+  src-java/stratum/internal/ColumnOpsVar.java \
   src-java/stratum/internal/ColumnOpsChunked.java \
   src-java/stratum/internal/ColumnOpsChunkedSimd.java \
   src-java/stratum/internal/ColumnOpsAnalytics.java

bench/olap_bench.clj

@@ -2344,8 +2344,8 @@
   (let [q {:from {:orderkey (:orderkey data) :linenumber (:linenumber data) :price (:price data)}
            :window [{:op :row-number :partition-by [:orderkey] :order-by [[:linenumber :asc]] :as :rn}]}
         qc (assoc q :result :columns)
-        r (bench #(q/q qc))
         r-1t (bench-1t #(q/q qc))
+        r (bench #(q/q qc))
         cr (q/q qc)
         n (:n-rows cr)]
     (println (format "  Stratum 1T: %s  (%d rows)" (fmt-ms (:median r-1t)) n))
@@ -2367,8 +2367,8 @@
   (let [q {:from {:orderkey (:orderkey data) :linenumber (:linenumber data) :price (:price data)}
            :window [{:op :lag :col :price :offset 1 :partition-by [:orderkey] :order-by [[:linenumber :asc]] :as :prev_price}]}
         qc (assoc q :result :columns)
-        r (bench #(q/q qc))
         r-1t (bench-1t #(q/q qc))
+        r (bench #(q/q qc))
         cr (q/q qc)
         n (:n-rows cr)]
     (println (format "  Stratum 1T: %s  (%d rows)" (fmt-ms (:median r-1t)) n))
@@ -2390,8 +2390,8 @@
   (let [q {:from {:orderkey (:orderkey data) :shipdate (:shipdate data) :price (:price data)}
            :window [{:op :sum :col :price :partition-by [:orderkey] :order-by [[:shipdate :asc]] :as :running_sum}]}
         qc (assoc q :result :columns)
-        r (bench #(q/q qc))
         r-1t (bench-1t #(q/q qc))
+        r (bench #(q/q qc))
         cr (q/q qc)
         n (:n-rows cr)]
     (println (format "  Stratum 1T: %s  (%d rows)" (fmt-ms (:median r-1t)) n))
@@ -2493,8 +2493,8 @@
   (let [q {:from (dissoc data :n)
            :window [{:op :row-number :partition-by [:ss_store_sk] :order-by [[:ss_net_profit :asc]] :as :rn}]}
         qc (assoc q :result :columns)
-        r (bench #(q/q qc) :warmup 10 :iters 5)
         r-1t (bench-1t #(q/q qc) :warmup 10 :iters 5)
+        r (bench #(q/q qc) :warmup 10 :iters 5)
         cr (q/q qc)
         n (:n-rows cr)]
     (println (format "  Stratum 1T: %s  (%d rows)" (fmt-ms (:median r-1t)) n))
@@ -2622,14 +2622,20 @@
       (when (seq query-args) (println (format "  Queries: %s" (str/join ", " query-args))))
       (println "================================================================"))
 
+    ;; JIT warmup — exercise all Java hot paths so cross-tier deoptimization
+    ;; doesn't distort benchmark results.
+    (let [t0 (System/nanoTime)]
+      (q/jit-warmup!)
+      (println (format "\nJIT warmup: %.1fs" (/ (- (System/nanoTime) t0) 1e9))))
+
     (let [all-results (atom {})]
 
       ;; === Tier 1: TPC-H / SSB (existing B1-B6) ===
       ;; Sort key: :shipdate (primary filter column for range predicates)
       (when (run-tier? "t1" "1" "tpch" "ssb")
         (println "\n=== Tier 1: TPC-H / SSB ===")
         (let [arrays (generate-arrays n)
-              conn (duckdb-setup-lineitem n)]
+              ^Connection conn (duckdb-setup-lineitem n)]
           (try
             (doseq [m modes]
               (println (format "\n  --- Input mode: %s ---" (name m)))

src-java/stratum/internal/ColumnOpsAnalytics.java

@@ -1048,7 +1048,7 @@ public static int[] windowArgSort(long[] partKeys, Object[] orderKeys,
                 pos[p]++;
             }
 
-            if (nParts >= 2 && length > 100_000) {
+            if (nParts >= 2 && length > ColumnOps.PARALLEL_THRESHOLD) {
                 // Batch partitions into nThreads coarse tasks (not one task per partition)
                 int nThreads = Math.min(POOL.getParallelism(), nParts);
                 int partsPerThread = (nParts + nThreads - 1) / nThreads;
@@ -1315,10 +1315,17 @@ private static int compareMulti(Object[] orderKeys, boolean[] orderDirs, int a,
      * Compute ROW_NUMBER window function over pre-sorted indices.
      */
     public static double[] windowRowNumber(long[] partKeys, int[] sortedIndices, int length) {
-        if (length < 200_000) return windowRowNumberSeq(partKeys, sortedIndices, length);
+        if (length < ColumnOps.PARALLEL_THRESHOLD) return windowRowNumberSeq(partKeys, sortedIndices, length);
         // Find partition boundaries in sorted order
         int[] partStarts = findPartitionBoundaries(partKeys, sortedIndices, length);
         int nParts = partStarts.length - 1;
+        // Only parallelize when partitions are large enough to amortize FJP overhead.
+        // With many tiny partitions (avg < 8 rows), the per-element work is trivial
+        // and parallel random writes to result[] cause cache contention.
+        // Threshold 8 keeps WIN benchmarks (avg 4 rows) sequential while allowing
+        // H2O-Q8 (avg 60 rows) to benefit from parallelism.
+        long avgPartSize = length / Math.max(nParts, 1);
+        if (avgPartSize < 8) return windowRowNumberSeq(partKeys, sortedIndices, length);
         double[] result = new double[length];
         int nThreads = Math.min(POOL.getParallelism(), Math.max(1, nParts));
         int partsPerThread = (nParts + nThreads - 1) / nThreads;
@@ -1381,10 +1388,13 @@ private static int[] findPartitionBoundaries(long[] partKeys, int[] sortedIndice
     public static double[] windowLagLead(long[] partKeys, int[] sortedIndices,
                                           Object values, int length,
                                           int offset, double defaultVal, boolean isLead) {
-        if (length < 200_000) return windowLagLeadSeq(partKeys, sortedIndices, values, length, offset, defaultVal, isLead);
+        if (length < ColumnOps.PARALLEL_THRESHOLD) return windowLagLeadSeq(partKeys, sortedIndices, values, length, offset, defaultVal, isLead);
 
         int[] partStarts = findPartitionBoundaries(partKeys, sortedIndices, length);
         int nParts = partStarts.length - 1;
+        long avgPartSize = length / Math.max(nParts, 1);
+        if (avgPartSize < 8) return windowLagLeadSeq(partKeys, sortedIndices, values, length, offset, defaultVal, isLead);
+
         double[] result = new double[length];
         boolean isLong = values instanceof long[];
         boolean isDouble = values instanceof double[];
@@ -1459,10 +1469,13 @@ private static double[] windowLagLeadSeq(long[] partKeys, int[] sortedIndices,
      */
     public static double[] windowRunningSum(long[] partKeys, int[] sortedIndices,
                                              Object values, int length) {
-        if (length < 200_000) return windowRunningSumSeq(partKeys, sortedIndices, values, length);
+        if (length < ColumnOps.PARALLEL_THRESHOLD) return windowRunningSumSeq(partKeys, sortedIndices, values, length);
 
         int[] partStarts = findPartitionBoundaries(partKeys, sortedIndices, length);
         int nParts = partStarts.length - 1;
+        long avgPartSize = length / Math.max(nParts, 1);
+        if (avgPartSize < 8) return windowRunningSumSeq(partKeys, sortedIndices, values, length);
+
         double[] result = new double[length];
         boolean isLong = values instanceof long[];
         boolean isDouble = values instanceof double[];