test(group): oracles for the fixes in #540 and the coverage gaps

test/rfl/group/count_distinct_str_many_groups.rfl

@@ -0,0 +1,15 @@
+;; count (distinct STR) over many groups with a tiny string vocabulary.
+;; Before the fix this crashed (SIGSEGV) once the group count was large.
+;; Rows i and i+65536 share a group; the text index is shifted by the row's
+;; block so a two-row group sees two distinct strings and a one-row group one.
+(set I (til 100000))
+(set TX (% (+ (% I 4) (div I 65536)) 4))
+(set T (table [k text] (list (as 'I32 (% (* I 17) 65536)) (at ["pooled long alpha value" "beta" "gamma" "delta"] TX))))
+(set R (xasc (select {from: T by: k n: (count text) s: (count (distinct text))}) 'k))
+(count R) -- 65536
+(min (at R 's)) -- 1
+(max (at R 's)) -- 2
+(sum (at R 'n)) -- 100000
+;; every group's distinct count equals its row count
+(count (select {from: R where: (!= n s)})) -- 0
+(count (select {from: R where: (== s 2)})) -- 34464

test/rfl/group/first_last_parallel_oracle.rfl

@@ -0,0 +1,27 @@
+;; first/last on a group larger than the parallel split grain, checked against
+;; the rows themselves rather than against another grouping path.  With the
+;; suite's two workers the grain is total/2, so a 280k-row group splits.
+(set N 300000)
+(set K (as 'I64 (== 0 (% (til N) 15))))
+(set V (+ 1 (% (* 7919 (til N)) 100003)))
+(set W (/ (as 'F64 (% (* 31 (til N)) 977)) 4))
+(set W (+ W (at [0.0 0.0 0Nf 0.0 0.0] (% (til N) 5))))
+(set T (table [k v w] (list K V W)))
+(set R (xasc (select {from: T by: k f: (first v) l: (last v) fw: (first w) lw: (last w) n: (count v)}) 'k))
+(at R 'n) -- [280000 20000]
+;; group 0 is every row with i%15 != 0: first is row 1, last is row N-1
+(at (at R 'f) 0) -- (at V 1)
+(at (at R 'l) 0) -- (at V (- N 1))
+;; group 1 is every row with i%15 == 0: first is row 0, last is the top multiple of 15
+(at (at R 'f) 1) -- (at V 0)
+(at (at R 'l) 1) -- (at V (* 15 (div (- N 1) 15)))
+;; first/last of a nullable column report the row's value (the probed rows are non-null; row 2 of every five is null)
+(at (at R 'fw) 0) -- (at W 1)
+(at (at R 'lw) 0) -- (at W (- N 1))
+(at (at R 'fw) 1) -- (at W 0)
+(at (at R 'lw) 1) -- (at W (* 15 (div (- N 1) 15)))
+;; a selection keeps row order: first/last come from the selected rows
+(set S (xasc (select {from: T where: (> v 50000) by: k f: (first v) l: (last v)}) 'k))
+(set SEL (select {from: T where: (> v 50000)}))
+(at (at S 'f) 0) -- (first (at (select {from: SEL where: (== k 0)}) 'v))
+(at (at S 'l) 0) -- (last (at (select {from: SEL where: (== k 0)}) 'v))

test/rfl/group/list_key_count_distinct.rfl

@@ -0,0 +1,14 @@
+;; LIST-typed keys with count (distinct) and a streaming aggregate beside it.
+;; Before the fix the key emit tried to build a LIST result column through the
+;; typed-vector constructor and failed with a type error.
+(set I (til 100000))
+(set G (list [1 2] [3 4] ['a 'b] ["pooled list string" "x"]))
+(set T (table [k v] (list (at G (% I 4)) (% I 7))))
+(set R (select {from: T by: k s: (count (distinct v)) n: (count v) t: (sum v)}))
+(count R) -- 4
+(asc (at R 'n)) -- [25000 25000 25000 25000]
+(at R 's) -- [7 7 7 7]
+(sum (at R 't)) -- (sum (% I 7))
+(set S (select {from: T where: (< v 3) by: k s: (count (distinct v))}))
+(count S) -- 4
+(at S 's) -- [3 3 3 3]

test/rfl/group/topk_full_size_k.rfl

@@ -0,0 +1,28 @@
+;; top/bot with K >= the group's value count must return the values SORTED
+;; (top descending, bot ascending), also when nulls leave fewer than K values.
+;; Before the fix the partially filled heap was emitted in heap order.
+
+(set T (table [k v] (list [1 1 1 1 1 2 2] [5.0 3.0 9.0 1.0 7.0 2.0 8.0])))
+(set R (xasc (select {from: T by: k b: (bot v 1024) t: (top v 1024) b3: (bot v 3) t3: (top v 3)}) 'k))
+(at (at R 'b) 0) -- [1.0 3.0 5.0 7.0 9.0]
+(at (at R 't) 0) -- [9.0 7.0 5.0 3.0 1.0]
+(at (at R 'b) 1) -- [2.0 8.0]
+(at (at R 't) 1) -- [8.0 2.0]
+(at (at R 'b3) 1) -- [2.0 8.0]
+(at (at R 't3) 1) -- [8.0 2.0]
+
+;; nulls are skipped; the survivors must still come out sorted
+(set N (table [k v] (list [1 1 1 1 1 1 2 2 2] [5 0Nl 0Nl 9 0Nl 1 0Nl 0Nl 3])))
+(set RN (xasc (select {from: N by: k t: (top v 8) b: (bot v 8) t2: (top v 2)}) 'k))
+(at (at RN 't) 0) -- [9 5 1]
+(at (at RN 'b) 0) -- [1 5 9]
+(at (at RN 't2) 0) -- [9 5]
+(at (at RN 't) 1) -- [3]
+
+;; a group large enough for the parallel consumer: every list sorted
+(set B (table [k w] (list (take [1] 200000) (/ (as 'F64 (% (* 31 (til 200000)) 977)) 4))))
+(set RB (select {from: B by: k b: (bot w 1024) t: (top w 1024)}))
+(set LB (at (at RB 'b) 0))
+(set LT (at (at RB 't) 0))
+(min (>= (- (at LB (+ 1 (til 1023))) (at LB (til 1023))) 0)) -- true
+(min (<= (- (at LT (+ 1 (til 1023))) (at LT (til 1023))) 0)) -- true

test/rfl/group/topk_k_limits.rfl

@@ -0,0 +1,12 @@
+;; top/bot K is admitted for 1..1024 and rejected outside that range,
+;; on a small table and past the parallel threshold alike.
+(set T (table [k v] (list [1 1 2] [3 1 2])))
+(at (at (select {from: T by: k t: (top v 1)}) 't) 0) -- [3]
+(at (at (select {from: T by: k t: (top v 1024)}) 't) 0) -- [3 1]
+(try (select {from: T by: k t: (top v 0)}) (fn [e] 'rejected)) -- 'rejected
+(try (select {from: T by: k t: (top v 1025)}) (fn [e] 'rejected)) -- 'rejected
+(try (select {from: T by: k t: (bot v -1)}) (fn [e] 'rejected)) -- 'rejected
+(set B (table [k v] (list (take [1] 200000) (% (til 200000) 977))))
+(count (at (at (select {from: B by: k t: (top v 1024)}) 't) 0)) -- 1024
+(count (at (at (select {from: B by: k b: (bot v 1)}) 'b) 0)) -- 1
+(try (select {from: B by: k t: (top v 1025)}) (fn [e] 'rejected)) -- 'rejected

test/test_agg_contract.c

@@ -1099,6 +1099,32 @@ static test_result_t test_wide_count_distinct(void) {
         TEST_ASSERT_EQ_I(ray_table_get_col(out, ray_sym_intern("m", 1))->type, RAY_F64);
         ray_release(out);
 
+        /* Key-dependent values: an even key sees every vocabulary entry across
+         * the four blocks, an odd key only entry 0, so a count landing on the
+         * wrong key is visible (the uniform table above cannot tell). */
+        snprintf(script, sizeof(script),
+            "(set t2 (table [i k v] (list i (as 'I32 (%% i 65536)) (at %s (* (as 'I64 (/ i 65536)) (- 1 (%% i 2)))))))",
+            vocabularies[kind]);
+        setup = ray_eval_str(script);
+        TEST_ASSERT_TRUE(setup && !RAY_IS_ERR(setup)); ray_release(setup);
+        for (int shape = 0; shape < 4; shape++) {
+            int selected = shape & 1;
+            snprintf(script, sizeof(script), "(select {from:t2 by:k s:(count (distinct v)) %s %s})",
+                selected ? "where:(< i 196608)" : "", shape >= 2 ? "total:(sum i)" : "");
+            ray_t* keyed = ray_eval_str(script);
+            TEST_ASSERT_FMT(keyed && !RAY_IS_ERR(keyed), "keyed wide count distinct failed, type %d", types[kind]);
+            TEST_ASSERT_EQ_I(ray_table_nrows(keyed), 65536);
+            ray_t* keys = ray_table_get_col(keyed, ray_sym_intern("k", 1));
+            ray_t* cnt = ray_table_get_col(keyed, ray_sym_intern("s", 1));
+            TEST_ASSERT_NOT_NULL(keys); TEST_ASSERT_NOT_NULL(cnt);
+            TEST_ASSERT_EQ_I(keys->type, RAY_I32);
+            for (int64_t g = 0; g < cnt->len; g++) {
+                int32_t key = ((int32_t*)ray_data(keys))[g];
+                int64_t expect = (key & 1) ? 1 : (selected ? 2 : 3);
+                TEST_ASSERT_EQ_I(((int64_t*)ray_data(cnt))[g], expect);
+            }
+            ray_release(keyed);
+        }
     }
     PASS();
 }