diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e6c09c1d..1d0878f4 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/src/app/repl.c b/src/app/repl.c index daa223ad..32e5e65a 100644 --- a/src/app/repl.c +++ b/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; diff --git a/src/io/csv.c b/src/io/csv.c index 0c4e9277..18f49a46 100644 --- a/src/io/csv.c +++ b/src/io/csv.c @@ -43,6 +43,7 @@ #include "core/numparse.h" #include "core/pool.h" #include "core/platform.h" /* ray_vm_map_fd_ro / ray_vm_unmap_file (tracked) */ +#include "core/qstats.h" /* suppress progress for internal finalize tasks */ #include "lang/format.h" #include "ops/hash.h" #include "ops/idxop.h" /* attach per-chunk zone index after load */ @@ -719,6 +720,12 @@ static int64_t build_row_offsets(const char* buf, size_t buf_size, * Only scans for \n; pure \r line endings (old Mac) treated as single row. * Empty lines are preserved as rows (for NULL handling). */ for (;;) { + if (RAY_UNLIKELY((n & 0xFFFF) == 0 && ray_interrupted())) { + scratch_free(hdr); + *offsets_out = NULL; + *hdr_out = NULL; + return -1; + } const char* nl = (const char*)memchr(p, '\n', (size_t)(end - p)); if (!nl) break; p = nl + 1; @@ -739,7 +746,17 @@ static int64_t build_row_offsets(const char* buf, size_t buf_size, /* Slow path: track quote parity, byte-by-byte. * Empty lines preserved as rows (for NULL handling). */ bool in_quote = false; + size_t checked = 0; while (p < end) { + if (RAY_UNLIKELY(++checked == 65536)) { + checked = 0; + if (ray_interrupted()) { + scratch_free(hdr); + *offsets_out = NULL; + *hdr_out = NULL; + return -1; + } + } char c = *p; if (c == '"') { in_quote = !in_quote; @@ -795,6 +812,10 @@ static int64_t build_row_offsets_limited(const char* buf, size_t buf_size, if (RAY_LIKELY(!data_has_quotes)) { for (;;) { + if (RAY_UNLIKELY((n & 0xFFFF) == 0 && ray_interrupted())) { + scratch_free(hdr); + return -1; + } const char* nl = (const char*)memchr(p, '\n', (size_t)(end - p)); if (!nl) { p = end; @@ -820,7 +841,15 @@ static int64_t build_row_offsets_limited(const char* buf, size_t buf_size, } } else { bool in_quote = false; + size_t checked = 0; while (p < end) { + if (RAY_UNLIKELY(++checked == 65536)) { + checked = 0; + if (ray_interrupted()) { + scratch_free(hdr); + return -1; + } + } char c = *p; if (c == '"') { in_quote = !in_quote; @@ -898,6 +927,8 @@ static bool csv_intern_strings(csv_strref_t** str_refs, int n_cols, return false; /* OOM: cannot grow sym table */ for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) + return false; if (refs[r].ptr == NULL) { /* Empty/missing field → sym 0 (the canonical empty * symbol). SYM columns are no-null by design. */ @@ -920,11 +951,15 @@ static bool csv_intern_strings(csv_strref_t** str_refs, int n_cols, static void csv_free_escaped_strrefs(csv_strref_t** str_refs, int n_cols, const csv_type_t* col_types, int64_t n_rows, - const char* buf, size_t buf_size) { + const char* buf, size_t buf_size, + const uint8_t* row_done, + const bool* col_had_escaped) { const char* buf_end = buf + buf_size; for (int c = 0; c < n_cols; c++) { if (col_types[c] != CSV_TYPE_STR || !str_refs[c]) continue; + if (col_had_escaped && !col_had_escaped[c]) continue; for (int64_t r = 0; r < n_rows; r++) { + if (row_done && !row_done[r]) continue; const char* p = str_refs[c][r].ptr; if (p && (p < buf || p >= buf_end)) ray_sys_free((void*)p); @@ -949,6 +984,8 @@ static bool csv_fill_str_cols(csv_strref_t** str_refs, int n_cols, * wouldn't fit in the u32 offset field. */ uint64_t pool_bytes = 0; for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) + return false; if (refs[r].ptr == NULL) continue; uint32_t l = refs[r].len; if (l > RAY_STR_INLINE_MAX) pool_bytes += l; @@ -967,6 +1004,8 @@ static bool csv_fill_str_cols(csv_strref_t** str_refs, int n_cols, uint32_t pool_off = 0; for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) + return false; memset(&dst[r], 0, sizeof(ray_str_t)); if (refs[r].ptr == NULL) continue; const char* p = refs[r].ptr; @@ -1037,6 +1076,8 @@ typedef struct { void** col_data; /* non-const: workers write parsed values into columns */ csv_strref_t** str_refs; /* [n_cols] — strref arrays for string columns, NULL for others */ bool* worker_had_null; /* [n_workers * n_cols] */ + bool* worker_had_escaped; /* [n_workers * n_cols] */ + uint8_t* row_done; /* [n_rows], marks initialized strref rows */ } csv_par_ctx_t; static void csv_parse_fn(void* arg, uint32_t worker_id, @@ -1045,8 +1086,11 @@ static void csv_parse_fn(void* arg, uint32_t worker_id, char esc_buf[8192]; const char* buf_end = ctx->buf + ctx->buf_size; bool* my_had_null = &ctx->worker_had_null[(size_t)worker_id * (size_t)ctx->n_cols]; + bool* my_had_escaped = &ctx->worker_had_escaped[(size_t)worker_id * (size_t)ctx->n_cols]; for (int64_t row = start; row < end_row; row++) { + if (RAY_UNLIKELY(((row - start) & 1023) == 0 && ray_interrupted())) + return; const char* p = ctx->buf + ctx->row_offsets[row]; const char* row_end = (row + 1 < ctx->n_rows) ? ctx->buf + ctx->row_offsets[row + 1] @@ -1188,6 +1232,7 @@ static void csv_parse_fn(void* arg, uint32_t worker_id, * (freed below) — both die before csv_fill_str_cols * reads the strref. Persist escaped fields. */ if (fld < ctx->buf || fld >= buf_end) { + my_had_escaped[c] = true; if (dyn_esc && fld == dyn_esc) { dyn_esc = NULL; /* transfer ownership */ } else { @@ -1205,6 +1250,7 @@ static void csv_parse_fn(void* arg, uint32_t worker_id, } if (RAY_UNLIKELY(dyn_esc != NULL)) ray_sys_free(dyn_esc); } + if (ctx->row_done) ctx->row_done[row] = 1; } } @@ -1212,18 +1258,22 @@ static void csv_parse_fn(void* arg, uint32_t worker_id, * Serial parse fallback (small files or no thread pool) * -------------------------------------------------------------------------- */ -static void csv_parse_serial(const char* buf, size_t buf_size, +static bool csv_parse_serial(const char* buf, size_t buf_size, const int64_t* row_offsets, int64_t n_rows, int n_cols, char delim, const csv_type_t* col_types, const int8_t* resolved_types, void** col_data, csv_strref_t** str_refs, - bool* col_had_null) { + bool* col_had_null, + bool* col_had_escaped, + uint8_t* row_done) { char esc_buf[8192]; const char* buf_end = buf + buf_size; for (int64_t row = 0; row < n_rows; row++) { + if (RAY_UNLIKELY((row & 1023) == 0 && ray_interrupted())) + return false; const char* p = buf + row_offsets[row]; const char* row_end = (row + 1 < n_rows) ? buf + row_offsets[row + 1] @@ -1363,6 +1413,7 @@ static void csv_parse_serial(const char* buf, size_t buf_size, * (freed below) — both die before csv_fill_str_cols * reads the strref. Persist escaped fields. */ if (fld < buf || fld >= buf_end) { + col_had_escaped[c] = true; if (dyn_esc && fld == dyn_esc) { dyn_esc = NULL; /* transfer ownership */ } else { @@ -1380,7 +1431,9 @@ static void csv_parse_serial(const char* buf, size_t buf_size, } if (RAY_UNLIKELY(dyn_esc != NULL)) ray_sys_free(dyn_esc); } + if (row_done) row_done[row] = 1; } + return true; } /* Per-column elem size for the hash-attach cap. Mirrors the integer @@ -1520,7 +1573,7 @@ static int8_t csv_auto_width_to_ray(csv_type_t w) { /* Accumulate per-column min/max/has_null over a block of rows for the columns * flagged RAY_CSV_AUTO_TAG in resolved_types; non-AUTO columns are skipped. * Mirrors the field-walk (and short-row null handling) of csv_parse_serial. */ -static void csv_auto_scan_rows(const char* buf, const char* buf_end, +static bool csv_auto_scan_rows(const char* buf, const char* buf_end, const int64_t* row_offsets, int64_t n_rows, int ncols, char delim, const int8_t* resolved_types, @@ -1528,6 +1581,8 @@ static void csv_auto_scan_rows(const char* buf, const char* buf_end, bool* col_had_null) { char esc_buf[8192]; for (int64_t row = 0; row < n_rows; row++) { + if (RAY_UNLIKELY((row & 1023) == 0 && ray_interrupted())) + return false; const char* p = buf + row_offsets[row]; const char* row_end = (row + 1 < n_rows) ? buf + row_offsets[row + 1] : buf_end; @@ -1557,44 +1612,48 @@ static void csv_auto_scan_rows(const char* buf, const char* buf_end, if (RAY_UNLIKELY(dyn_esc != NULL)) ray_sys_free(dyn_esc); } } + return true; } /* Resolve every RAY_CSV_AUTO_TAG column in resolved_types[] to a concrete * width using a stats scan over the already-built whole-file row_offsets. * No-op when the schema carries no `INT` column. */ -static void csv_resolve_auto_in_place(const char* buf, size_t file_size, +static bool csv_resolve_auto_in_place(const char* buf, size_t file_size, const int64_t* row_offsets, int64_t n_rows, int ncols, char delim, int8_t* resolved_types) { bool any = false; for (int c = 0; c < ncols; c++) if (resolved_types[c] == RAY_CSV_AUTO_TAG) any = true; - if (!any) return; + if (!any) return true; int64_t col_min[CSV_MAX_COLS], col_max[CSV_MAX_COLS]; bool col_had_null[CSV_MAX_COLS]; for (int c = 0; c < ncols; c++) { col_min[c] = INT64_MAX; col_max[c] = INT64_MIN; col_had_null[c] = false; } - csv_auto_scan_rows(buf, buf + file_size, row_offsets, n_rows, ncols, delim, - resolved_types, col_min, col_max, col_had_null); + if (!csv_auto_scan_rows(buf, buf + file_size, row_offsets, n_rows, ncols, + delim, resolved_types, col_min, col_max, + col_had_null)) + return false; for (int c = 0; c < ncols; c++) if (resolved_types[c] == RAY_CSV_AUTO_TAG) resolved_types[c] = csv_auto_width_to_ray( csv_resolve_int_width(col_min[c], col_max[c], col_had_null[c])); + return true; } /* Same as csv_resolve_auto_in_place, but for the streaming writers which never * build whole-file offsets: stream the file in chunks accumulating stats, * then resolve. Bounded memory (one chunk of offsets at a time). */ -static void csv_resolve_auto_streamed(const char* buf, size_t file_size, +static bool csv_resolve_auto_streamed(const char* buf, size_t file_size, size_t data_offset, int ncols, char delim, bool data_has_quotes, int8_t* resolved_types) { bool any = false; for (int c = 0; c < ncols; c++) if (resolved_types[c] == RAY_CSV_AUTO_TAG) any = true; - if (!any) return; + if (!any) return true; int64_t col_min[CSV_MAX_COLS], col_max[CSV_MAX_COLS]; bool col_had_null[CSV_MAX_COLS]; @@ -1604,6 +1663,7 @@ static void csv_resolve_auto_streamed(const char* buf, size_t file_size, size_t off = data_offset; while (off < file_size) { + if (ray_interrupted()) return false; ray_t* hdr = NULL; int64_t* roff = NULL; size_t next = off; @@ -1611,10 +1671,13 @@ static void csv_resolve_auto_streamed(const char* buf, size_t file_size, CSV_PART_ROWS_DEFAULT, data_has_quotes, &roff, &hdr, &next); - if (cnt <= 0) { scratch_free(hdr); break; } - csv_auto_scan_rows(buf, buf + file_size, roff, cnt, ncols, delim, - resolved_types, col_min, col_max, col_had_null); + if (cnt < 0) { scratch_free(hdr); return false; } + if (cnt == 0) { scratch_free(hdr); break; } + bool scan_ok = csv_auto_scan_rows(buf, buf + file_size, roff, cnt, + ncols, delim, resolved_types, + col_min, col_max, col_had_null); scratch_free(hdr); + if (!scan_ok) return false; if (next <= off) break; off = next; } @@ -1623,6 +1686,7 @@ static void csv_resolve_auto_streamed(const char* buf, size_t file_size, if (resolved_types[c] == RAY_CSV_AUTO_TAG) resolved_types[c] = csv_auto_width_to_ray( csv_resolve_int_width(col_min[c], col_max[c], col_had_null[c])); + return true; } static ray_t* csv_materialize_rows(const char* buf, size_t file_size, @@ -1661,6 +1725,8 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, bool col_had_null[CSV_MAX_COLS]; if (ncols > 0) memset(col_had_null, 0, (size_t)ncols * sizeof(bool)); + bool col_had_escaped[CSV_MAX_COLS]; + memset(col_had_escaped, 0, sizeof(col_had_escaped)); csv_type_t parse_types[CSV_MAX_COLS]; for (int c = 0; c < ncols; c++) { @@ -1695,6 +1761,14 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, ray_t* str_ref_hdrs[CSV_MAX_COLS]; memset(str_ref_bufs, 0, sizeof(str_ref_bufs)); memset(str_ref_hdrs, 0, sizeof(str_ref_hdrs)); + ray_t* row_done_hdr = NULL; + uint8_t* row_done = has_text_cols + ? (uint8_t*)scratch_calloc(&row_done_hdr, (size_t)n_rows) + : NULL; + if (has_text_cols && !row_done) { + for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); + return NULL; + } for (int c = 0; c < ncols; c++) { if (parse_types[c] == CSV_TYPE_STR) { size_t sz = (size_t)n_rows * sizeof(csv_strref_t); @@ -1702,6 +1776,7 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, if (!str_ref_bufs[c]) { for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); for (int j = 0; j < c; j++) scratch_free(str_ref_hdrs[j]); + scratch_free(row_done_hdr); return NULL; } } @@ -1713,12 +1788,15 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, if (use_parallel) { uint32_t n_workers = ray_pool_total_workers(pool); - size_t whn_sz = (size_t)n_workers * (size_t)ncols * sizeof(bool); - bool* worker_had_null_buf = (bool*)ray_sys_alloc(whn_sz); - if (!worker_had_null_buf) { + size_t worker_flags_sz = (size_t)n_workers * (size_t)ncols * sizeof(bool); + bool* worker_flags = (bool*)ray_sys_alloc(worker_flags_sz * 2); + if (!worker_flags) { use_parallel = false; } else { - memset(worker_had_null_buf, 0, whn_sz); + memset(worker_flags, 0, worker_flags_sz * 2); + bool* worker_had_null_buf = worker_flags; + bool* worker_had_escaped_buf = worker_flags + + (size_t)n_workers * (size_t)ncols; csv_par_ctx_t ctx = { .buf = buf, @@ -1732,6 +1810,8 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, .col_data = col_data, .str_refs = str_ref_bufs, .worker_had_null = worker_had_null_buf, + .worker_had_escaped = worker_had_escaped_buf, + .row_done = row_done, }; ray_pool_dispatch(pool, csv_parse_fn, &ctx, n_rows); @@ -1740,19 +1820,31 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, for (int c = 0; c < ncols; c++) { if (worker_had_null_buf[(size_t)w * (size_t)ncols + (size_t)c]) col_had_null[c] = true; + if (worker_had_escaped_buf[(size_t)w * (size_t)ncols + (size_t)c]) + col_had_escaped[c] = true; } } - ray_sys_free(worker_had_null_buf); + ray_sys_free(worker_flags); } } if (!use_parallel) { - csv_parse_serial(buf, file_size, row_offsets, n_rows, - ncols, delimiter, parse_types, resolved_types, col_data, - str_ref_bufs, col_had_null); + (void)csv_parse_serial(buf, file_size, row_offsets, n_rows, + ncols, delimiter, parse_types, resolved_types, + col_data, str_ref_bufs, col_had_null, + col_had_escaped, row_done); } } + if (ray_interrupted()) { + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, row_done, col_had_escaped); + for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); + for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); + return ray_error("cancel", "interrupted"); + } + if (has_text_cols) { csv_finalize_ctx_t fctx = { .str_refs = str_ref_bufs, @@ -1768,16 +1860,21 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, }; ray_pool_t* fpool = ray_pool_get(); if (fpool && ray_pool_total_workers(fpool) >= 2) { + uint32_t qmode = ray_qstats_mode(); + ray_qstats_set_mode(qmode & ~RAY_QS_PROGRESS); ray_pool_dispatch_n(fpool, csv_finalize_task, &fctx, 2); + ray_qstats_set_mode(qmode); } else { csv_finalize_task(&fctx, 0, 0, 1); csv_finalize_task(&fctx, 0, 1, 2); } - if (!fctx.fill_ok || !fctx.intern_ok) { - csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, buf, file_size); + if (!fctx.fill_ok || !fctx.intern_ok || ray_interrupted()) { + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, row_done, col_had_escaped); for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); - return NULL; + return ray_interrupted() ? ray_error("cancel", "interrupted") : NULL; } } @@ -1790,8 +1887,14 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, sym_max_ids[c] = max_id; } - csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, buf, file_size); + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, NULL, col_had_escaped); for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); + if (ray_interrupted()) { + for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); + return ray_error("cancel", "interrupted"); + } for (int c = 0; c < ncols; c++) { ray_t* vec = col_vecs[c]; @@ -1815,10 +1918,24 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, void* dst = ray_data(narrow); if (new_w == RAY_SYM_W8) { uint8_t* d = (uint8_t*)dst; - for (int64_t r = 0; r < n_rows; r++) d[r] = (uint8_t)src[r]; + for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) { + ray_release(narrow); + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } + d[r] = (uint8_t)src[r]; + } } else { uint16_t* d = (uint16_t*)dst; - for (int64_t r = 0; r < n_rows; r++) d[r] = (uint16_t)src[r]; + for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) { + ray_release(narrow); + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } + d[r] = (uint16_t)src[r]; + } } narrow->attrs |= (col_vecs[c]->attrs & RAY_ATTR_HAS_NULLS); ray_release(col_vecs[c]); @@ -1837,14 +1954,26 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, * stays as well — it's the entropy signal we just measured). See * csv_should_attach_hash for the selectivity + memory cap. */ for (int c = 0; c < ncols; c++) { + if (ray_interrupted()) { + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } ray_t* v = col_vecs[c]; if (!v || RAY_IS_ERR(v)) continue; if (v->len < (1 << 16)) continue; /* < one chunk, skip */ ray_t* r = ray_index_attach_chunk_zone(&v, 16); + if (ray_interrupted()) { + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } if (r && !RAY_IS_ERR(r)) col_vecs[c] = v; /* attach succeeded */ /* On failure the original column stays in col_vecs[c]; ignore. */ } for (int c = 0; c < ncols; c++) { + if (ray_interrupted()) { + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } ray_t* v = col_vecs[c]; if (!csv_should_attach_hash(v)) continue; /* ray_index_attach_hash drops any existing index on the @@ -1853,6 +1982,10 @@ static ray_t* csv_materialize_rows(const char* buf, size_t file_size, * is known to be high-entropy, chunk-skip never fires * anyway, so the chunk_zone is dead weight. */ ray_t* r = ray_index_attach_hash(&v); + if (ray_interrupted()) { + for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); + return ray_error("cancel", "interrupted"); + } if (r && !RAY_IS_ERR(r)) col_vecs[c] = v; } @@ -1905,6 +2038,8 @@ static const char* csv_skip_matching_header(const char* p, const char* buf_end, ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, const int8_t* col_types_in, int32_t n_types, const int64_t* col_names_in, int32_t n_names) { + if (ray_interrupted()) return ray_error("cancel", "interrupted"); + /* ---- 1. Open file and get size ---- */ int fd = open(path, O_RDONLY); if (fd < 0) return ray_error("io", NULL); @@ -1960,6 +2095,18 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, return ray_error("range", "csv read: header has too many columns, got %lld (max %d)", (long long)ncols, CSV_MAX_COLS); } + if (col_types_in && n_types != ncols) { + ray_vm_unmap_file(buf, file_size); + return ray_error("length", + "csv read: schema has %lld types but file has %d columns", + (long long)n_types, ncols); + } + if (col_names_in && n_names != ncols) { + ray_vm_unmap_file(buf, file_size); + return ray_error("length", + "csv read: schema has %lld names but file has %d columns", + (long long)n_names, ncols); + } /* ---- 5. Parse header row ---- */ const char* p = buf; @@ -1980,7 +2127,7 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, * lines are null data rows. */ if (p < buf_end && *p == '\r') p++; if (p < buf_end && *p == '\n') p++; - } else if (col_names_in && n_names >= ncols) { + } else if (col_names_in) { for (int c = 0; c < ncols; c++) col_name_ids[c] = col_names_in[c]; const char* after = csv_skip_matching_header(buf, buf_end, delimiter, @@ -2002,6 +2149,11 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, int64_t n_rows = build_row_offsets(buf, file_size, data_offset, &row_offsets, &row_offsets_hdr); + if (n_rows < 0) { + ray_vm_unmap_file(buf, file_size); + return ray_error("cancel", "interrupted"); + } + if (n_rows == 0) { /* Empty file → empty table */ ray_t* tbl = ray_table_new(ncols); @@ -2019,15 +2171,18 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, /* ---- 7. Resolve column types ---- */ int8_t resolved_types[CSV_MAX_COLS]; - if (col_types_in && n_types >= ncols) { + if (col_types_in) { /* Explicit types provided by caller — validate against known types */ for (int c = 0; c < ncols; c++) { int8_t t = col_types_in[c]; if (t < RAY_BOOL || (t >= RAY_TYPE_COUNT && t != RAY_CSV_AUTO_TAG) || t == RAY_TABLE) { - /* Invalid type constant — fall through to error */ - goto fail_offsets; + scratch_free(row_offsets_hdr); + ray_vm_unmap_file(buf, file_size); + return ray_error("type", + "csv read: unsupported type code %d in column %d", + (int)t, c); } resolved_types[c] = t; } @@ -2036,16 +2191,14 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, ncols, delimiter, esc_buf, resolved_types)) goto fail_offsets; - } else { - /* col_types_in provided but too short — error */ - goto fail_offsets; } /* `INT` schema columns: resolve to a concrete narrowest width over all * rows before allocating vectors, so the explicit-width path below sees a * real type. */ - csv_resolve_auto_in_place(buf, file_size, row_offsets, n_rows, - ncols, delimiter, resolved_types); + if (!csv_resolve_auto_in_place(buf, file_size, row_offsets, n_rows, + ncols, delimiter, resolved_types)) + goto fail_offsets_cancel; /* ---- 8. Allocate column vectors ---- */ ray_t* col_vecs[CSV_MAX_COLS]; @@ -2070,6 +2223,9 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, bool col_had_null[CSV_MAX_COLS]; if (ncols > 0) memset(col_had_null, 0, (size_t)ncols * sizeof(bool)); + bool col_had_escaped[CSV_MAX_COLS]; + memset(col_had_escaped, 0, sizeof(col_had_escaped)); + /* Build csv_type_t array for parse functions (maps td types → csv types) */ csv_type_t parse_types[CSV_MAX_COLS]; for (int c = 0; c < ncols; c++) { @@ -2106,6 +2262,14 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, ray_t* str_ref_hdrs[CSV_MAX_COLS]; memset(str_ref_bufs, 0, sizeof(str_ref_bufs)); memset(str_ref_hdrs, 0, sizeof(str_ref_hdrs)); + ray_t* row_done_hdr = NULL; + uint8_t* row_done = has_text_cols + ? (uint8_t*)scratch_calloc(&row_done_hdr, (size_t)n_rows) + : NULL; + if (has_text_cols && !row_done) { + for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); + goto fail_offsets; + } for (int c = 0; c < ncols; c++) { if (parse_types[c] == CSV_TYPE_STR) { size_t sz = (size_t)n_rows * sizeof(csv_strref_t); @@ -2113,6 +2277,7 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, if (!str_ref_bufs[c]) { for (int j = 0; j < ncols; j++) ray_release(col_vecs[j]); for (int j = 0; j < c; j++) scratch_free(str_ref_hdrs[j]); + scratch_free(row_done_hdr); goto fail_offsets; } } @@ -2124,12 +2289,15 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, if (use_parallel) { uint32_t n_workers = ray_pool_total_workers(pool); - size_t whn_sz = (size_t)n_workers * (size_t)ncols * sizeof(bool); - bool* worker_had_null_buf = (bool*)ray_sys_alloc(whn_sz); - if (!worker_had_null_buf) { + size_t worker_flags_sz = (size_t)n_workers * (size_t)ncols * sizeof(bool); + bool* worker_flags = (bool*)ray_sys_alloc(worker_flags_sz * 2); + if (!worker_flags) { use_parallel = false; } else { - memset(worker_had_null_buf, 0, whn_sz); + memset(worker_flags, 0, worker_flags_sz * 2); + bool* worker_had_null_buf = worker_flags; + bool* worker_had_escaped_buf = worker_flags + + (size_t)n_workers * (size_t)ncols; csv_par_ctx_t ctx = { .buf = buf, @@ -2143,6 +2311,8 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, .col_data = col_data, .str_refs = str_ref_bufs, .worker_had_null = worker_had_null_buf, + .worker_had_escaped = worker_had_escaped_buf, + .row_done = row_done, }; ray_pool_dispatch(pool, csv_parse_fn, &ctx, n_rows); @@ -2152,19 +2322,24 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, for (int c = 0; c < ncols; c++) { if (worker_had_null_buf[(size_t)w * (size_t)ncols + (size_t)c]) col_had_null[c] = true; + if (worker_had_escaped_buf[(size_t)w * (size_t)ncols + (size_t)c]) + col_had_escaped[c] = true; } } - ray_sys_free(worker_had_null_buf); + ray_sys_free(worker_flags); } } if (!use_parallel) { - csv_parse_serial(buf, file_size, row_offsets, n_rows, - ncols, delimiter, parse_types, resolved_types, col_data, - str_ref_bufs, col_had_null); + (void)csv_parse_serial(buf, file_size, row_offsets, n_rows, + ncols, delimiter, parse_types, resolved_types, + col_data, str_ref_bufs, col_had_null, + col_had_escaped, row_done); } } + if (ray_interrupted()) goto fail_parsed_cancel; + /* ---- 9b. Materialize RAY_STR columns AND batch-intern sym columns ---- * These two phases touch disjoint columns and (after the GUID fix) * intern_strings is the only one that mutates the global sym table. @@ -2185,14 +2360,24 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, }; ray_pool_t* fpool = ray_pool_get(); if (fpool && ray_pool_total_workers(fpool) >= 2) { + /* Finalization has two coarse internal tasks, not rows. Letting + * the generic pool progress pump expose them resets a completed + * n_rows parse to 0/2 (visibly 100% -> 50%). Keep profiling + * active, but preserve the row progress until query completion. */ + uint32_t qmode = ray_qstats_mode(); + ray_qstats_set_mode(qmode & ~RAY_QS_PROGRESS); ray_pool_dispatch_n(fpool, csv_finalize_task, &fctx, 2); + ray_qstats_set_mode(qmode); } else { csv_finalize_task(&fctx, 0, 0, 1); csv_finalize_task(&fctx, 0, 1, 2); } - if (!fctx.fill_ok || !fctx.intern_ok) { - csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, buf, file_size); + if (!fctx.fill_ok || !fctx.intern_ok || ray_interrupted()) { + if (ray_interrupted()) goto fail_parsed_cancel; + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, NULL, col_had_escaped); for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); goto fail_offsets; } @@ -2202,14 +2387,19 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, if (resolved_types[c] != RAY_SYM) continue; uint32_t* ids = (uint32_t*)col_data[c]; int64_t max_id = 0; - for (int64_t r = 0; r < n_rows; r++) + for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) + goto fail_parsed_cancel; if ((int64_t)ids[r] > max_id) max_id = ids[r]; + } sym_max_ids[c] = max_id; } /* Free heap-allocated escaped string copies, then strref buffers */ - csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, buf, file_size); + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, NULL, col_had_escaped); for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); /* ---- 9c. Set HAS_NULLS for columns that saw a null ---- * @@ -2239,10 +2429,22 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, void* dst = ray_data(narrow); if (new_w == RAY_SYM_W8) { uint8_t* d = (uint8_t*)dst; - for (int64_t r = 0; r < n_rows; r++) d[r] = (uint8_t)src[r]; + for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) { + ray_release(narrow); + goto fail_cols_cancel; + } + d[r] = (uint8_t)src[r]; + } } else { /* RAY_SYM_W16 */ uint16_t* d = (uint16_t*)dst; - for (int64_t r = 0; r < n_rows; r++) d[r] = (uint16_t)src[r]; + for (int64_t r = 0; r < n_rows; r++) { + if (RAY_UNLIKELY((r & 1023) == 0 && ray_interrupted())) { + ray_release(narrow); + goto fail_cols_cancel; + } + d[r] = (uint16_t)src[r]; + } } narrow->attrs |= (col_vecs[c]->attrs & RAY_ATTR_HAS_NULLS); ray_release(col_vecs[c]); @@ -2258,16 +2460,20 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, * Second pass upgrades high-entropy columns to a hash index; * see csv_should_attach_hash. */ for (int c = 0; c < ncols; c++) { + if (ray_interrupted()) goto fail_cols_cancel; ray_t* v = col_vecs[c]; if (!v || RAY_IS_ERR(v)) continue; if (v->len < (1 << 16)) continue; ray_t* r = ray_index_attach_chunk_zone(&v, 16); + if (ray_interrupted()) goto fail_cols_cancel; if (r && !RAY_IS_ERR(r)) col_vecs[c] = v; } for (int c = 0; c < ncols; c++) { + if (ray_interrupted()) goto fail_cols_cancel; ray_t* v = col_vecs[c]; if (!csv_should_attach_hash(v)) continue; ray_t* r = ray_index_attach_hash(&v); + if (ray_interrupted()) goto fail_cols_cancel; if (r && !RAY_IS_ERR(r)) col_vecs[c] = v; } @@ -2291,6 +2497,17 @@ ray_t* ray_read_csv_named_opts(const char* path, char delimiter, bool header, return result; /* Error paths */ +fail_parsed_cancel: + csv_free_escaped_strrefs(str_ref_bufs, ncols, parse_types, n_rows, + buf, file_size, row_done, col_had_escaped); + for (int c = 0; c < ncols; c++) scratch_free(str_ref_hdrs[c]); + scratch_free(row_done_hdr); +fail_cols_cancel: + for (int c = 0; c < ncols; c++) ray_release(col_vecs[c]); +fail_offsets_cancel: + scratch_free(row_offsets_hdr); + ray_vm_unmap_file(buf, file_size); + return ray_error("cancel", "interrupted"); fail_offsets: scratch_free(row_offsets_hdr); fail_unmap: @@ -2434,6 +2651,7 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool const int8_t* col_types_in, int32_t n_types, const int64_t* col_names_in, int32_t n_names, const char* dir, int64_t rows_per_chunk) { + if (ray_interrupted()) return RAY_ERR_CANCEL; if (!path || !dir) return RAY_ERR_DOMAIN; if (rows_per_chunk <= 0) rows_per_chunk = CSV_PART_ROWS_DEFAULT; @@ -2481,6 +2699,11 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool ray_vm_unmap_file(buf, file_size); return RAY_ERR_RANGE; } + if ((col_types_in && n_types != ncols) || + (col_names_in && n_names != ncols)) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_LENGTH; + } const char* p = buf; char esc_buf[8192]; @@ -2497,7 +2720,7 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool } if (p < buf_end && *p == '\r') p++; if (p < buf_end && *p == '\n') p++; - } else if (col_names_in && n_names >= ncols) { + } else if (col_names_in) { for (int c = 0; c < ncols; c++) col_name_ids[c] = col_names_in[c]; const char* after = csv_skip_matching_header(buf, buf_end, delimiter, @@ -2514,7 +2737,7 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool size_t data_offset = (size_t)(p - buf); bool data_has_quotes = memchr(buf + data_offset, '"', file_size - data_offset) != NULL; int8_t resolved_types[CSV_MAX_COLS]; - if (col_types_in && n_types >= ncols) { + if (col_types_in) { for (int c = 0; c < ncols; c++) { int8_t t = col_types_in[c]; if (t < RAY_BOOL || @@ -2534,6 +2757,10 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool &sample_offsets, &sample_offsets_hdr, NULL); + if (sample_n < 0) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_CANCEL; + } bool infer_ok = csv_infer_types_from_offsets( buf, buf_end, sample_offsets, sample_n, ncols, delimiter, esc_buf, resolved_types); @@ -2542,9 +2769,6 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool ray_vm_unmap_file(buf, file_size); return RAY_ERR_OOM; } - } else { - ray_vm_unmap_file(buf, file_size); - return RAY_ERR_TYPE; } for (int c = 0; c < ncols; c++) { @@ -2576,8 +2800,11 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool /* `INT` schema columns: resolve to a concrete narrowest width over the * whole file before opening the fixed-width streaming column writers. */ - csv_resolve_auto_streamed(buf, file_size, data_offset, ncols, delimiter, - data_has_quotes, resolved_types); + if (!csv_resolve_auto_streamed(buf, file_size, data_offset, ncols, + delimiter, data_has_quotes, resolved_types)) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_CANCEL; + } err = ray_mkdir_p(dir); if (err != RAY_OK) { @@ -2660,7 +2887,7 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool &row_offsets_hdr, &next_offset); if (cnt <= 0) { scratch_free(row_offsets_hdr); - err = RAY_ERR_IO; + err = (cnt < 0) ? RAY_ERR_CANCEL : RAY_ERR_IO; break; } } @@ -2670,8 +2897,9 @@ ray_err_t ray_csv_save_splayed_named_opts(const char* path, char delimiter, bool resolved_types); scratch_free(row_offsets_hdr); if (!tbl || RAY_IS_ERR(tbl)) { + err = (tbl && RAY_IS_ERR(tbl)) ? ray_err_from_obj(tbl) + : RAY_ERR_OOM; if (tbl) ray_release(tbl); - err = RAY_ERR_OOM; break; } @@ -2738,6 +2966,7 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool const int64_t* col_names_in, int32_t n_names, const char* root, const char* table_name, int64_t rows_per_part) { + if (ray_interrupted()) return RAY_ERR_CANCEL; if (!path || !root || !table_name) return RAY_ERR_DOMAIN; if (rows_per_part <= 0) rows_per_part = CSV_PART_ROWS_DEFAULT; bool trace = getenv("RAY_CSV_TRACE") != NULL; @@ -2786,6 +3015,11 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool ray_vm_unmap_file(buf, file_size); return RAY_ERR_RANGE; } + if ((col_types_in && n_types != ncols) || + (col_names_in && n_names != ncols)) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_LENGTH; + } const char* p = buf; char esc_buf[8192]; @@ -2802,7 +3036,7 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool } if (p < buf_end && *p == '\r') p++; if (p < buf_end && *p == '\n') p++; - } else if (col_names_in && n_names >= ncols) { + } else if (col_names_in) { for (int c = 0; c < ncols; c++) col_name_ids[c] = col_names_in[c]; const char* after = csv_skip_matching_header(buf, buf_end, delimiter, @@ -2825,7 +3059,7 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool } int8_t resolved_types[CSV_MAX_COLS]; - if (col_types_in && n_types >= ncols) { + if (col_types_in) { for (int c = 0; c < ncols; c++) { int8_t t = col_types_in[c]; if (t < RAY_BOOL || @@ -2845,6 +3079,10 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool &sample_offsets, &sample_offsets_hdr, NULL); + if (sample_n < 0) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_CANCEL; + } bool infer_ok = csv_infer_types_from_offsets( buf, buf_end, sample_offsets, sample_n, ncols, delimiter, esc_buf, resolved_types); @@ -2853,15 +3091,15 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool ray_vm_unmap_file(buf, file_size); return RAY_ERR_OOM; } - } else { - ray_vm_unmap_file(buf, file_size); - return RAY_ERR_TYPE; } /* `INT` schema columns: resolve to a concrete narrowest width over the * whole file so every partition writes the same column type. */ - csv_resolve_auto_streamed(buf, file_size, data_offset, ncols, delimiter, - data_has_quotes, resolved_types); + if (!csv_resolve_auto_streamed(buf, file_size, data_offset, ncols, + delimiter, data_has_quotes, resolved_types)) { + ray_vm_unmap_file(buf, file_size); + return RAY_ERR_CANCEL; + } err = ray_mkdir_p(root); if (err != RAY_OK) { @@ -2887,7 +3125,7 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool fprintf(stderr, "csv.parted: row-offset failure part=%" PRId64 " offset=%zu\n", part, chunk_offset); scratch_free(row_offsets_hdr); - err = RAY_ERR_IO; + err = (cnt < 0) ? RAY_ERR_CANCEL : RAY_ERR_IO; break; } } @@ -2895,9 +3133,10 @@ ray_err_t ray_csv_save_parted_named_opts(const char* path, char delimiter, bool ray_t* tbl = csv_materialize_rows(buf, file_size, row_offsets, cnt, ncols, delimiter, col_name_ids, resolved_types); if (!tbl || RAY_IS_ERR(tbl)) { + err = (tbl && RAY_IS_ERR(tbl)) ? ray_err_from_obj(tbl) + : RAY_ERR_OOM; if (tbl) ray_release(tbl); scratch_free(row_offsets_hdr); - err = RAY_ERR_OOM; if (trace) fprintf(stderr, "csv.parted: materialize failure part=%" PRId64 " rows=%" PRId64 "\n", part, cnt); diff --git a/src/ops/builtins.c b/src/ops/builtins.c index b2140d86..d32ce9af 100644 --- a/src/ops/builtins.c +++ b/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; diff --git a/src/ops/idxop.c b/src/ops/idxop.c index 13076964..97dd5ff0 100644 --- a/src/ops/idxop.c +++ b/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"); } /* -------------------------------------------------------------------------- diff --git a/test/rfl/io/csv_types.rfl b/test/rfl/io/csv_types.rfl index b6b27019..de067b59 100644 --- a/test/rfl/io/csv_types.rfl +++ b/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 diff --git a/test/test_csv.c b/test/test_csv.c index 47c50945..c0f1bb61 100644 --- a/test/test_csv.c +++ b/test/test_csv.c @@ -25,6 +25,7 @@ #include #include #include "mem/heap.h" +#include "core/qstats.h" #include "io/csv.h" #include "table/sym.h" #include @@ -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 }, };