fix(csv): harden import cancellation and schema handling

.github/workflows/ci.yml

@@ -89,6 +89,10 @@ jobs:
   # work-stealing ring real contention to race against.  Linux + clang only.
   tsan:
     runs-on: ubuntu-latest
+    env:
+      # Keep Clang off transient hosted-runner ISA features that become
+      # -Werror diagnostics before the TSan suites can run.
+      RAY_MARCH: x86-64
     steps:
       - uses: actions/checkout@v4
       - name: TSan concurrency suites

src/app/repl.c

@@ -147,6 +147,10 @@ static void render_progress_full(int64_t done, int64_t total,
     if (total > 0) {
         pct = (double)done / (double)total;
         if (pct > 1.0) pct = 1.0;
+        /* A dispatch can finish before its enclosing query's cleanup and
+         * finalization phases.  Reserve 100% for the final snapshot (which
+         * clears the line) so the live bar never claims completion early. */
+        if (pct >= 1.0) pct = 0.99;
         int sub = (int)(pct * sub_total);
         full = sub / 8;
         frac = sub % 8;

src/ops/builtins.c

@@ -505,8 +505,13 @@ static int8_t resolve_type_name(int64_t sym_id) {
     else if (len == 3 && memcmp(name, "F64", 3) == 0) result = RAY_F64;
     else if (len == 2 && memcmp(name, "B8", 2) == 0) result = RAY_BOOL;
     else if (len == 2 && memcmp(name, "U8", 2) == 0) result = RAY_U8;
+    else if (len == 3 && memcmp(name, "SYM", 3) == 0) result = RAY_SYM;
     else if (len == 6 && memcmp(name, "SYMBOL", 6) == 0) result = RAY_SYM;
     else if (len == 3 && memcmp(name, "STR", 3) == 0) result = RAY_STR;
+    /* CSV cells are scalar text, so the historical LIST schema spelling is
+     * materialized as the native packed STR column rather than a boxed list.
+     * This keeps schemas copied from older table displays loadable. */
+    else if (len == 4 && memcmp(name, "LIST", 4) == 0) result = RAY_STR;
     else if (len == 3 && memcmp(name, "F32", 3) == 0) result = RAY_F32;
     else if (len == 4 && memcmp(name, "DATE", 4) == 0) result = RAY_DATE;
     else if (len == 4 && memcmp(name, "TIME", 4) == 0) result = RAY_TIME;

src/ops/idxop.c

@@ -453,6 +453,7 @@ static ray_err_t chunk_zone_scan_int(ray_t* v, ray_index_t* ix,
     const uint8_t* base = (const uint8_t*)ray_data(v);
 
     for (uint32_t g = 0; g < n_chunks; g++) {
+        if (RAY_UNLIKELY(ray_interrupted())) return RAY_ERR_CANCEL;
         int64_t s = (int64_t)g * csz;
         int64_t e = s + csz; if (e > n) e = n;
         int64_t mn = INT64_MAX, mx = INT64_MIN;
@@ -491,6 +492,7 @@ static ray_err_t chunk_zone_scan_float(ray_t* v, ray_index_t* ix,
     const uint8_t* base = (const uint8_t*)ray_data(v);
 
     for (uint32_t g = 0; g < n_chunks; g++) {
+        if (RAY_UNLIKELY(ray_interrupted())) return RAY_ERR_CANCEL;
         int64_t s = (int64_t)g * csz;
         int64_t e = s + csz; if (e > n) e = n;
         double mn = INFINITY, mx = -INFINITY;
@@ -965,6 +967,12 @@ ray_t* ray_index_attach_hash(ray_t** vp) {
     int64_t n_keys = 0, n_groups = 0;
 
     for (int64_t i = 0; i < n; i++) {
+        if (RAY_UNLIKELY((i & 0xFFFF) == 0 && ray_interrupted())) {
+            scratch_free(btab_hdr);
+            scratch_free(rgid_hdr);
+            ray_release(gkeys);
+            return ray_error("cancel", "interrupted");
+        }
         if (ray_vec_is_null(v, i)) { rgid[i] = -1; continue; }  /* SYM is null-free → always false */
         int64_t kw = (v->type == RAY_SYM)
             ? ray_read_sym(base, i, RAY_SYM, v->attrs)  /* domain id, width-aware */
@@ -1003,14 +1011,26 @@ ray_t* ray_index_attach_hash(ray_t** vp) {
     int64_t* of = (int64_t*)ray_data(offs);
     int64_t* rw = (int64_t*)ray_data(rows);
     memset(of, 0, (size_t)(n_groups + 1) * sizeof(int64_t));
-    for (int64_t i = 0; i < n; i++)
+    for (int64_t i = 0; i < n; i++) {
+        if (RAY_UNLIKELY((i & 0xFFFF) == 0 && ray_interrupted()))
+            goto hash_cancel_csr;
         if (rgid[i] >= 0) of[rgid[i] + 1]++;
-    for (int64_t g = 0; g < n_groups; g++)
+    }
+    for (int64_t g = 0; g < n_groups; g++) {
+        if (RAY_UNLIKELY((g & 0xFFFF) == 0 && ray_interrupted()))
+            goto hash_cancel_csr;
         of[g + 1] += of[g];
-    for (int64_t i = 0; i < n; i++)
+    }
+    for (int64_t i = 0; i < n; i++) {
+        if (RAY_UNLIKELY((i & 0xFFFF) == 0 && ray_interrupted()))
+            goto hash_cancel_csr;
         if (rgid[i] >= 0) rw[of[rgid[i]]++] = i;
-    for (int64_t g = n_groups; g > 0; g--)
+    }
+    for (int64_t g = n_groups; g > 0; g--) {
+        if (RAY_UNLIKELY((g & 0xFFFF) == 0 && ray_interrupted()))
+            goto hash_cancel_csr;
         of[g] = of[g - 1];
+    }
     of[0] = 0;
     scratch_free(rgid_hdr);
 
@@ -1027,6 +1047,11 @@ ray_t* ray_index_attach_hash(ray_t** vp) {
     int64_t* tbl = (int64_t*)ray_data(table);
     memset(tbl, 0, (size_t)cap * sizeof(int64_t));
     for (int64_t g = 0; g < n_groups; g++) {
+        if (RAY_UNLIKELY((g & 0xFFFF) == 0 && ray_interrupted())) {
+            ray_release(table); ray_release(gkeys);
+            ray_release(offs);  ray_release(rows);
+            return ray_error("cancel", "interrupted");
+        }
         uint64_t slot = mix64((uint64_t)gk[g]) & mask;
         while (tbl[slot] != 0) slot = (slot + 1) & mask;
         tbl[slot] = g + 1;
@@ -1049,6 +1074,13 @@ ray_t* ray_index_attach_hash(ray_t** vp) {
     ix->u.hash.order_sym = -1;
 
     return attach_finalize(v, idx);
+
+hash_cancel_csr:
+    scratch_free(rgid_hdr);
+    ray_release(gkeys);
+    ray_release(offs);
+    ray_release(rows);
+    return ray_error("cancel", "interrupted");
 }
 
 /* --------------------------------------------------------------------------

test/rfl/io/csv_types.rfl

@@ -353,6 +353,21 @@
 (set Tcardlostr (.csv.read [STR] "rf_csv_card_lo.csv"))
 (type (at Tcardlostr 'k)) -- 'STR
 
+;; ════════════════════════════════════════════════════════════════════════════
+;; Explicit schema aliases and exact width validation.
+;; SYM is the canonical name printed by `type`; SYMBOL remains the long alias.
+;; Historical LIST text columns materialize as packed STR columns.  A schema
+;; must describe every CSV column exactly instead of failing later as OOM.
+;; ════════════════════════════════════════════════════════════════════════════
+(.sys.exec "printf 's,text\nfoo,hello\nbar,world\n' > rf_test_csv_types_alias.csv") -- 0
+(set Talias (.csv.read [SYM LIST] "rf_test_csv_types_alias.csv"))
+(type (at Talias 's)) -- 'SYM
+(type (at Talias 'text)) -- 'STR
+(at (at Talias 'text) 0) -- "hello"
+(.csv.read [SYM] "rf_test_csv_types_alias.csv") !- length
+(.csv.read [SYM LIST I64] "rf_test_csv_types_alias.csv") !- length
+(.sys.exec "rm -f rf_test_csv_types_alias.csv") -- 0
+
 ;; ── cleanup ─────────────────────────────────────────────────────────────────
 (.sys.exec "rm -f rf_csv_card_*.csv") -- 0
 (.sys.exec "rm -f rf_test_csv_types_*.csv") -- 0

test/test_csv.c

@@ -25,6 +25,7 @@
 #include <rayforce.h>
 #include <rayforce.h>
 #include "mem/heap.h"
+#include "core/qstats.h"
 #include "io/csv.h"
 #include "table/sym.h"
 #include <stdio.h>
@@ -1545,6 +1546,103 @@ static test_result_t test_csv_resolve_int_width(void) {
     PASS();
 }
 
+typedef struct {
+    int calls;
+    bool cancel_on_first;
+    bool went_backwards;
+    uint64_t last_done;
+    uint64_t last_total;
+} csv_progress_probe_t;
+
+static void csv_progress_probe_cb(const ray_progress_t* p, void* user) {
+    csv_progress_probe_t* probe = (csv_progress_probe_t*)user;
+    if (p->final || p->rows_total == 0) return;
+    if (probe->last_total != 0 &&
+        p->rows_done * probe->last_total < probe->last_done * p->rows_total)
+        probe->went_backwards = true;
+    probe->last_done = p->rows_done;
+    probe->last_total = p->rows_total;
+    probe->calls++;
+    if (probe->cancel_on_first && probe->calls == 1)
+        ray_request_interrupt();
+}
+
+/* Cancellation must be observed while pool workers are inside the CSV row
+ * loop, and cleanup must only inspect initialized strrefs.  Sparse escaped
+ * fields exercise ownership cleanup without turning the fixture into an
+ * allocation benchmark. */
+static test_result_t test_csv_interrupt_mid_parse(void) {
+    ray_heap_init();
+    (void)ray_sym_init();
+
+    FILE* f = fopen(TMP_CSV, "w");
+    TEST_ASSERT_NOT_NULL(f);
+    fputs("id,payload\n", f);
+    for (int i = 0; i < 200000; i++) {
+        if ((i % 1000) == 0)
+            fprintf(f, "%d,\"escaped \"\"value\"\" %d\"\n", i, i);
+        else
+            fprintf(f, "%d,payload_%06d\n", i, i);
+    }
+    fclose(f);
+
+    csv_progress_probe_t probe = {.cancel_on_first = true};
+    ray_progress_set_callback(csv_progress_probe_cb, &probe, 1, 1);
+    ray_qstats_set_mode(RAY_QS_PROGRESS);
+    int8_t schema[] = {RAY_I32, RAY_STR};
+    ray_t* loaded = ray_read_csv_opts(TMP_CSV, ',', true, schema, 2);
+    bool got_cancel = loaded && RAY_IS_ERR(loaded) &&
+                      strcmp(ray_err_code(loaded), "cancel") == 0;
+
+    if (loaded) ray_release(loaded);
+    ray_progress_end();
+    ray_progress_set_callback(NULL, NULL, 0, 0);
+    ray_qstats_set_mode(0);
+    ray_clear_interrupt();
+    unlink(TMP_CSV);
+    ray_sym_destroy();
+    ray_heap_destroy();
+
+    TEST_ASSERT_TRUE(probe.calls > 0);
+    TEST_ASSERT_TRUE(got_cancel);
+    PASS();
+}
+
+/* The row parser is followed by a two-task text finalizer.  That internal
+ * dispatch must not replace n_rows progress with 0/2 and make the visible
+ * percentage jump from complete back to 50%. */
+static test_result_t test_csv_progress_never_goes_backwards(void) {
+    ray_heap_init();
+    (void)ray_sym_init();
+
+    FILE* f = fopen(TMP_CSV, "w");
+    TEST_ASSERT_NOT_NULL(f);
+    fputs("payload,symbol\n", f);
+    for (int i = 0; i < 100000; i++)
+        fprintf(f, "long_payload_%06d_for_progress,unique_symbol_%06d\n", i, i);
+    fclose(f);
+
+    csv_progress_probe_t probe = {0};
+    ray_progress_set_callback(csv_progress_probe_cb, &probe, 1, 1);
+    ray_qstats_set_mode(RAY_QS_PROGRESS);
+    int8_t schema[] = {RAY_STR, RAY_SYM};
+    ray_t* loaded = ray_read_csv_opts(TMP_CSV, ',', true, schema, 2);
+    bool read_ok = loaded && !RAY_IS_ERR(loaded);
+
+    if (loaded) ray_release(loaded);
+    ray_progress_end();
+    ray_progress_set_callback(NULL, NULL, 0, 0);
+    ray_qstats_set_mode(0);
+    unlink(TMP_CSV);
+    ray_sym_destroy();
+    ray_heap_destroy();
+
+    TEST_ASSERT_TRUE(read_ok);
+    TEST_ASSERT_TRUE(probe.calls > 0);
+    TEST_ASSERT_FALSE(probe.went_backwards);
+    PASS();
+}
+
 const test_entry_t csv_entries[] = {
     { "csv/roundtrip_i64", test_csv_roundtrip_i64, NULL, NULL },
     { "csv/roundtrip_guid", test_csv_guid_roundtrip, NULL, NULL },
@@ -1597,5 +1695,7 @@ const test_entry_t csv_entries[] = {
     { "csv/explicit_u8_schema_serial",
                                   test_csv_explicit_u8_schema_serial,      NULL, NULL },
     { "csv/resolve_int_width",    test_csv_resolve_int_width,              NULL, NULL },
+    { "csv/interrupt_mid_parse",  test_csv_interrupt_mid_parse,             NULL, NULL },
+    { "csv/progress_monotonic",   test_csv_progress_never_goes_backwards,   NULL, NULL },
     { NULL, NULL, NULL, NULL },
 };