pivot, asof/window-join: flatten parted input instead of out-of-bounds read (fixes two segfaults)

src/ops/join.c

@@ -2860,7 +2860,34 @@ static bool asof_hash_group_match(uint32_t n_eq,
     return ok;
 }
 
+static ray_t* exec_window_join_flat(ray_graph_t* g, ray_op_t* op,
+                                    ray_t* left_table, ray_t* right_table);
+
+/* Parted-input guard for asof-join / window-join — mirrors
+ * join_with_parted_guard but flatten-only (window-join has no broadcast path,
+ * and its op ext is the asof-window ext, not the equi-join ext).  A raw
+ * RAY_PARTED/MAPCOMMON column indexed to nrows OOB-reads → SIGSEGV; flatten
+ * any parted side first.  The parted-asof STREAMING path (query.c) feeds this
+ * kernel FLAT per-day tables via OP_WINDOW_JOIN, so the guard is inert there
+ * and only fires for a bare parted table handed directly to (asof-join …). */
 ray_t* exec_window_join(ray_graph_t* g, ray_op_t* op,
+                        ray_t* left_table, ray_t* right_table) {
+    ray_t *lf = NULL, *rf = NULL;
+    if (join_table_has_parted_col(left_table)) {
+        lf = join_flatten_parted(left_table);
+        if (!lf || RAY_IS_ERR(lf)) return lf ? lf : ray_error("oom", NULL);
+    }
+    if (join_table_has_parted_col(right_table)) {
+        rf = join_flatten_parted(right_table);
+        if (!rf || RAY_IS_ERR(rf)) { if (lf) ray_release(lf); return rf ? rf : ray_error("oom", NULL); }
+    }
+    ray_t* r = exec_window_join_flat(g, op, lf ? lf : left_table, rf ? rf : right_table);
+    if (lf) ray_release(lf);
+    if (rf) ray_release(rf);
+    return r;
+}
+
+static ray_t* exec_window_join_flat(ray_graph_t* g, ray_op_t* op,
                                ray_t* left_table, ray_t* right_table) {
     ray_op_ext_t* ext = find_ext(g, op->id);
     if (!ext) return ray_error("nyi", NULL);

src/ops/pivot.c

@@ -276,9 +276,56 @@ static inline uint64_t pivot_hash_null_words(uint64_t h, const int64_t* nullw,
  * pivot values into separate output columns.
  * ============================================================================ */
 
+/* Segment count of a parted table = length of any parted column's segs array
+ * (mirrors build_segment_table exec.c). */
+static int64_t pivot_parted_seg_count(ray_t* tbl) {
+    int64_t nc = ray_table_ncols(tbl);
+    for (int64_t c = 0; c < nc; c++) {
+        ray_t* col = ray_table_get_col_idx(tbl, c);
+        if (col && RAY_IS_PARTED(col->type)) return col->len;
+    }
+    return 0;
+}
+
+static bool pivot_table_has_parted_col(ray_t* tbl) {
+    int64_t nc = ray_table_ncols(tbl);
+    for (int64_t c = 0; c < nc; c++) {
+        ray_t* col = ray_table_get_col_idx(tbl, c);
+        if (col && (RAY_IS_PARTED(col->type) || col->type == RAY_MAPCOMMON))
+            return true;
+    }
+    return false;
+}
+
 ray_t* exec_pivot(ray_graph_t* g, ray_op_t* op, ray_t* tbl) {
     if (!tbl || RAY_IS_ERR(tbl)) return tbl;
 
+    /* Parted-input guard: exec_pivot fetches columns via ray_table_get_col and
+     * indexes them by row up to ray_table_nrows.  A raw RAY_PARTED/MAPCOMMON
+     * column (whose data is a tiny segs array, not nrows elements) would be
+     * indexed out of bounds → OOB read / SIGSEGV.  Flatten (concat all
+     * segments) and recompute on the flat table, mirroring exec_window /
+     * join_with_parted_guard.  A flat table has no parted column, so the
+     * recursive call falls straight through this guard (one level). */
+    if (pivot_table_has_parted_col(tbl)) {
+        int64_t nseg = pivot_parted_seg_count(tbl);
+        ray_t* flat = NULL;
+        for (int64_t s = 0; s < nseg; s++) {
+            ray_t* seg = build_segment_table(tbl, (int32_t)s);
+            if (!seg || RAY_IS_ERR(seg)) { if (flat) ray_release(flat); return seg ? seg : ray_error("oom", NULL); }
+            if (!flat) flat = seg;
+            else {
+                ray_t* m = ray_result_merge(flat, seg);
+                ray_release(flat); ray_release(seg);
+                if (!m || RAY_IS_ERR(m)) return m ? m : ray_error("oom", NULL);
+                flat = m;
+            }
+        }
+        ray_t* r = exec_pivot(g, op, flat);
+        ray_release(flat);
+        return r;
+    }
+
     ray_op_ext_t* ext = find_ext(g, op->id);
     if (!ext) return ray_error("nyi", NULL);
 

test/rfl/join/asof_parted_guard.rfl

@@ -0,0 +1,30 @@
+;; asof_parted_guard.rfl — a BARE parted table (not a (select {from:...})
+;; operand) handed directly to (asof-join ...) must flatten and compute, not
+;; out-of-bounds read.  exec_window_join fetched time/key columns via
+;; ray_table_get_col and indexed to ray_table_nrows; a raw RAY_PARTED column
+;; was indexed OOB → SIGSEGV exit 139 (confirmed bare-parted on a 40M-row
+;; table).  The parted-asof STREAMING path (asof of select-operands) is
+;; separate and feeds this kernel flat per-day tables; this guards the eager
+;; path for a bare parted table.  Result must equal the flat-oracle asof.
+(.sys.exec "rm -rf /tmp/rfl_asofg_t /tmp/rfl_asofg_q")
+(set T1 (table [sym time price] (list [a b] [2024.01.01D10:00:03.000000000 2024.01.01D10:00:05.000000000] (as 'F64 [10.0 20.0]))))
+(set Q1 (table [sym time bid] (list [a a b] [2024.01.01D10:00:00.000000000 2024.01.01D10:00:02.000000000 2024.01.01D10:00:01.000000000] (as 'F64 [99.0 100.0 50.0]))))
+(set T2 (table [sym time price] (list [a] [2024.01.02D10:00:02.000000000] (as 'F64 [30.0]))))
+(set Q2 (table [sym time bid] (list [a] [2024.01.02D10:00:01.000000000] (as 'F64 [150.0]))))
+(.db.splayed.set "/tmp/rfl_asofg_t/2024.01.01/trade/" T1 "/tmp/rfl_asofg_t/.sym")
+(.db.splayed.set "/tmp/rfl_asofg_t/2024.01.02/trade/" T2 "/tmp/rfl_asofg_t/.sym")
+(.db.splayed.set "/tmp/rfl_asofg_q/2024.01.01/quote/" Q1 "/tmp/rfl_asofg_q/.sym")
+(.db.splayed.set "/tmp/rfl_asofg_q/2024.01.02/quote/" Q2 "/tmp/rfl_asofg_q/.sym")
+(set PT (.db.parted.get "/tmp/rfl_asofg_t" 'trade))
+(set PQ (.db.parted.get "/tmp/rfl_asofg_q" 'quote))
+(count PT) -- 3
+;; flat oracle: partitions concatenated in day order.
+(set FT (table [sym time price] (list [a b a] [2024.01.01D10:00:03.000000000 2024.01.01D10:00:05.000000000 2024.01.02D10:00:02.000000000] (as 'F64 [10.0 20.0 30.0]))))
+(set FQ (table [sym time bid] (list [a a b a] [2024.01.01D10:00:00.000000000 2024.01.01D10:00:02.000000000 2024.01.01D10:00:01.000000000 2024.01.02D10:00:01.000000000] (as 'F64 [99.0 100.0 50.0 150.0]))))
+;; bare parted asof must not crash and must equal the flat asof.
+(set RP (asof-join ['sym 'time] PT PQ))
+(set RF (asof-join ['sym 'time] FT FQ))
+(count RP) -- 3
+(count RP) -- (count RF)
+(at RP 'bid) -- (at RF 'bid)
+(at RP 'price) -- (at RF 'price)

test/rfl/pivot/pivot_parted_guard.rfl

@@ -0,0 +1,24 @@
+;; pivot_parted_guard.rfl — a parted table fed to (pivot ...) must flatten and
+;; compute, not out-of-bounds read.  exec_pivot fetches columns via
+;; ray_table_get_col and indexes them to ray_table_nrows; a raw RAY_PARTED
+;; column (data = tiny segs array, not nrows elements) was indexed OOB →
+;; SIGSEGV exit 139 (confirmed on a 40M-row parted table).  The guard flattens
+;; (concat segments) then recomputes; result must equal the flat-oracle pivot.
+(.sys.exec "rm -rf /tmp/rfl_pivot_parted")
+(set T1 (table [sym cat qty] (list (as 'SYM ['a 'b 'a]) (as 'SYM ['x 'y 'x]) (as 'I64 [1 2 3]))))
+(set T2 (table [sym cat qty] (list (as 'SYM ['b 'a]) (as 'SYM ['y 'x]) (as 'I64 [4 5]))))
+(.db.splayed.set "/tmp/rfl_pivot_parted/2024.01.01/t/" T1 "/tmp/rfl_pivot_parted/.sym")
+(.db.splayed.set "/tmp/rfl_pivot_parted/2024.01.02/t/" T2 "/tmp/rfl_pivot_parted/.sym")
+(set P (.db.parted.get "/tmp/rfl_pivot_parted" 't))
+(count P) -- 5
+;; flat oracle: the same rows concatenated in partition order.
+(set F (table [sym cat qty] (list (as 'SYM ['a 'b 'a 'b 'a]) (as 'SYM ['x 'y 'x 'y 'x]) (as 'I64 [1 2 3 4 5]))))
+;; parted pivot must not crash and must equal the flat pivot.
+(set RP (pivot P 'sym 'cat 'qty sum))
+(set RF (pivot F 'sym 'cat 'qty sum))
+(count RP) -- (count RF)
+(count RP) -- 2
+(cols RP) -- (cols RF)
+(at RP 'sym) -- (at RF 'sym)
+(at RP 'x) -- (at RF 'x)
+(at RP 'y) -- (at RF 'y)