diff --git a/rayforce-sys/Cargo.toml b/rayforce-sys/Cargo.toml index 10019e8..c631dfd 100644 --- a/rayforce-sys/Cargo.toml +++ b/rayforce-sys/Cargo.toml @@ -12,6 +12,7 @@ build = "build.rs" [build-dependencies] bindgen = "0.70" +cc = "1" [lib] doctest = false diff --git a/rayforce-sys/build.rs b/rayforce-sys/build.rs index 0f5b6ee..cb516bf 100644 --- a/rayforce-sys/build.rs +++ b/rayforce-sys/build.rs @@ -26,6 +26,17 @@ fn main() { sanitize_libclang_path(); build_core_lib(&core); + // --- vendored KDB+ IPC client (csrc/kdb_ipc.c) --- + // Compiled here and linked BEFORE librayforce so its undefined `ray_*` + // symbols resolve from the core archive. (Staging spot until it moves into + // the engine proper.) + cc::Build::new() + .file("csrc/kdb_ipc.c") + .include(&include) + .warnings(false) + .compile("rkdb"); + println!("cargo:rerun-if-changed=csrc/kdb_ipc.c"); + // --- linking --- println!("cargo:rustc-link-search=native={}", core.display()); println!("cargo:rustc-link-lib=static=rayforce"); diff --git a/rayforce-sys/csrc/kdb_ipc.c b/rayforce-sys/csrc/kdb_ipc.c new file mode 100644 index 0000000..ffdec42 --- /dev/null +++ b/rayforce-sys/csrc/kdb_ipc.c @@ -0,0 +1,1048 @@ +#define _GNU_SOURCE +/* + * raypy_kdb.c — KDB+ IPC client. + * + * Sends raw query strings (or any rayforce object) to a KDB+ server using + * the documented KDB+ wire format and returns the response decoded into + * rayforce types. Blocking sockets, no event loop. + * + * Supported KDB+ wire types + * atoms: -KB -KG -KH -KI -KJ -KE -KF -KC -KS -KP -KM -KD -KN -KU -KV + * -KT -UU + * vectors: KB UU KG KH KI KJ KE KF KC KS KP KM KD KZ KN KU KV KT + * nested: 0 (general list), XT (table), XD (dict) + * error: -128 + */ + +/* Vendored from rayforce-py's rayforce/capi/raypy_kdb.c (the KDB+ IPC client). + * The 3 Python entry points are replaced with plain-C functions (rkdb_connect / + * rkdb_send / rkdb_close) callable from Rust via rayforce-sys; everything else — + * the wire serialize/deserialize, handshake, decompression — is unchanged. Two + * small helpers (`ray_scalar_elem_size`, `raypy_build_table`) are inlined so the + * file depends only on the public engine header. */ +#include + +#include +#include +#include +#include +#include +#include +#include + +/* v2 marks a dict as a RAY_LIST carrying this attr (from rayforce-py). */ +#ifndef RAY_ATTR_DICT +#define RAY_ATTR_DICT 0x02 +#endif + +/* type → scalar element byte size (from rayforce-py rayforce_c.h). */ +static inline size_t ray_scalar_elem_size(int8_t type) { + switch (type < 0 ? -type : type) { + case RAY_BOOL: + case RAY_U8: + return 1; + case RAY_I16: + return 2; + case RAY_I32: + case RAY_DATE: + case RAY_TIME: + case RAY_F32: + return 4; + case RAY_I64: + case RAY_F64: + case RAY_TIMESTAMP: + return 8; + case RAY_GUID: + return 16; + default: + return 0; + } +} + +/* Build a v2 table from column-name sym ids + column vectors (from rayforce-py + * raypy_init_from_py.c). */ +static ray_t *raypy_build_table(const int64_t *col_ids, ray_t *const *cols, + int64_t ncols) { + ray_t *tbl = ray_table_new(ncols); + if (tbl == NULL) + return ray_error("table_new failed", NULL); + if (RAY_IS_ERR(tbl)) + return tbl; + for (int64_t i = 0; i < ncols; i++) { + if (cols[i] == NULL) { + ray_release(tbl); + return ray_error("table column is null", NULL); + } + ray_t *res = ray_table_add_col(tbl, col_ids[i], cols[i]); + if (res == NULL || RAY_IS_ERR(res)) { + ray_release(tbl); + return res ? res : ray_error("table_add_col failed", NULL); + } + tbl = res; + } + return tbl; +} + +/* ================================================================ + * KDB+ wire-format constants + * ================================================================ */ +#define KDB_KB 1 /* boolean */ +#define KDB_UU 2 /* guid (16B) */ +#define KDB_KG 4 /* byte */ +#define KDB_KH 5 /* short */ +#define KDB_KI 6 /* int */ +#define KDB_KJ 7 /* long */ +#define KDB_KE 8 /* real (f32) */ +#define KDB_KF 9 /* float (f64) */ +#define KDB_KC 10 /* char */ +#define KDB_KS 11 /* symbol */ +#define KDB_KP 12 /* timestamp */ +#define KDB_KM 13 /* month */ +#define KDB_KD 14 /* date */ +#define KDB_KZ 15 /* datetime */ +#define KDB_KN 16 /* timespan */ +#define KDB_KU 17 /* minute */ +#define KDB_KV 18 /* second */ +#define KDB_KT 19 /* time */ +#define KDB_XT 98 /* table */ +#define KDB_XD 99 /* dict */ +#define KDB_ERR (-128) + +#define KDB_MSG_SYNC 1 + +typedef struct { + uint8_t endianness; + uint8_t msgtype; + uint8_t compressed; + uint8_t reserved; + uint32_t size; +} kdb_header_t; + +#define KDB_MAX_CONNS 256 +static int kdb_conn_fd[KDB_MAX_CONNS]; +static int kdb_conn_init = 0; + +static void kdb_init_table(void) { + if (kdb_conn_init) + return; + for (int i = 0; i < KDB_MAX_CONNS; i++) + kdb_conn_fd[i] = -1; + kdb_conn_init = 1; +} + +static int kdb_alloc_slot(int fd) { + kdb_init_table(); + for (int i = 0; i < KDB_MAX_CONNS; i++) { + if (kdb_conn_fd[i] == -1) { + kdb_conn_fd[i] = fd; + return i; + } + } + return -1; +} + +/* ================================================================ + * Socket helpers + * ================================================================ */ + +static ssize_t kdb_recv_all(int fd, void *buf, size_t n) { + size_t total = 0; + uint8_t *p = (uint8_t *)buf; + while (total < n) { + ssize_t r = recv(fd, p + total, n - total, 0); + if (r == 0) + return -1; + if (r < 0) { + if (errno == EINTR) + continue; + return -1; + } + total += (size_t)r; + } + return (ssize_t)total; +} + +static ssize_t kdb_send_all(int fd, const void *buf, size_t n) { + size_t total = 0; + const uint8_t *p = (const uint8_t *)buf; + while (total < n) { + ssize_t r = send(fd, p + total, n - total, 0); + if (r <= 0) { + if (r < 0 && errno == EINTR) + continue; + return -1; + } + total += (size_t)r; + } + return (ssize_t)total; +} + +static int kdb_open_socket(const char *host, int port) { + char service[16]; + snprintf(service, sizeof(service), "%d", port); + + struct addrinfo hints = {0}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + struct addrinfo *res = NULL; + if (getaddrinfo(host, service, &hints, &res) != 0 || res == NULL) + return -1; + + int fd = -1; + for (struct addrinfo *p = res; p != NULL; p = p->ai_next) { + fd = socket(p->ai_family, p->ai_socktype, p->ai_protocol); + if (fd < 0) + continue; + if (connect(fd, p->ai_addr, p->ai_addrlen) == 0) + break; + close(fd); + fd = -1; + } + freeaddrinfo(res); + return fd; +} + +/* ================================================================ + * Wire-format size + serialize (rayforce v2 → KDB+) + * ================================================================ */ + +static int8_t kdb_type_of(int8_t ray_type) { + int sign = (ray_type < 0) ? -1 : 1; + int t = (ray_type < 0) ? -ray_type : ray_type; + switch (t) { + case RAY_BOOL: + return (int8_t)(sign * KDB_KB); + case RAY_U8: + return (int8_t)(sign * KDB_KG); + case RAY_I16: + return (int8_t)(sign * KDB_KH); + case RAY_I32: + return (int8_t)(sign * KDB_KI); + case RAY_I64: + return (int8_t)(sign * KDB_KJ); + case RAY_F32: + return (int8_t)(sign * KDB_KE); + case RAY_F64: + return (int8_t)(sign * KDB_KF); + case RAY_DATE: + return (int8_t)(sign * KDB_KD); + case RAY_TIME: + return (int8_t)(sign * KDB_KT); + case RAY_TIMESTAMP: + return (int8_t)(sign * KDB_KP); + case RAY_GUID: + return (int8_t)(sign * KDB_UU); + case RAY_SYM: + return (int8_t)(sign * KDB_KS); + case RAY_STR: + return (int8_t)(sign * KDB_KC); + case RAY_LIST: + return 0; + case RAY_TABLE: + return KDB_XT; + case RAY_DICT: + return KDB_XD; + case RAY_ERROR: + return KDB_ERR; + default: + return 0; + } +} + +/* Forward */ +static int64_t kdb_size_obj(ray_t *obj); +static int64_t kdb_ser_obj(uint8_t *buf, ray_t *obj); +static ray_t *kdb_des_obj(uint8_t **buf, int64_t *len); + +/* Build a v2 table from a RAY_SYM-vec of column names and a RAY_LIST of + * column vectors. Consumes references to `keys` and `vals`. */ +static ray_t *kdb_make_table(ray_t *keys, ray_t *vals) { + if (keys == NULL || vals == NULL || keys->type != RAY_SYM || + vals->type != RAY_LIST || keys->len != vals->len) { + if (keys) + ray_release(keys); + if (vals) + ray_release(vals); + return ray_error("kdb: malformed table — expected (sym-vec, list)", NULL); + } + ray_t *tbl = raypy_build_table((const int64_t *)ray_data(keys), + (ray_t *const *)ray_data(vals), keys->len); + ray_release(keys); + ray_release(vals); + return tbl; +} + +static int64_t kdb_size_obj(ray_t *obj) { + if (obj == NULL || obj == RAY_NULL_OBJ) + return 1 + 1 + 4; /* type + attrs + len(0) */ + + int8_t t = obj->type; + + /* Atoms */ + if (t < 0) { + int abs_t = -t; + switch (abs_t) { + case RAY_BOOL: + case RAY_U8: + return 1 + 1; + case RAY_I16: + return 1 + 2; + case RAY_I32: + case RAY_DATE: + case RAY_TIME: + case RAY_F32: + return 1 + 4; + case RAY_I64: + case RAY_TIMESTAMP: + case RAY_F64: + return 1 + 8; + case RAY_GUID: + return 1 + 16; + case RAY_SYM: { + ray_t *s = ray_sym_str(obj->i64); + int64_t n = s ? (int64_t)ray_str_len(s) : 0; + if (s) + ray_release(s); + return 1 + n + 1; /* null-terminated */ + } + case RAY_STR: { + /* Single char goes as -KC atom; multi-char as KC vector. */ + int64_t n = (int64_t)ray_str_len(obj); + if (n == 1) + return 1 + 1; + return 1 + 1 + 4 + n; + } + } + return 0; + } + + /* Vectors / containers */ + if (t == RAY_LIST) { + int64_t size = 1 + 1 + 4; + ray_t **elems = (ray_t **)ray_data(obj); + for (int64_t i = 0; i < obj->len; i++) + size += kdb_size_obj(elems[i]); + return size; + } + if (t == RAY_SYM) { + int64_t size = 1 + 1 + 4; + int64_t *ids = (int64_t *)ray_data(obj); + for (int64_t i = 0; i < obj->len; i++) { + ray_t *s = ray_sym_str(ids[i]); + size += s ? (int64_t)ray_str_len(s) + 1 : 1; + if (s) + ray_release(s); + } + return size; + } + if (t == RAY_TABLE) { + /* Wire form: XT byte + attrs(0) + XD marker + KS-vec of names + + * list-0 of column vectors. v2 RAY_TABLE is opaque — extract via the + * public accessors instead of indexing ray_data directly. */ + int64_t ncols = ray_table_ncols(obj); + int64_t names = 1 + 1 + 4; /* KS type + attrs + count */ + for (int64_t i = 0; i < ncols; i++) { + ray_t *s = ray_sym_str(ray_table_col_name(obj, i)); + names += s ? (int64_t)ray_str_len(s) + 1 : 1; + if (s) + ray_release(s); + } + int64_t cols = 1 + 1 + 4; /* list type + attrs + count */ + for (int64_t i = 0; i < ncols; i++) + cols += kdb_size_obj(ray_table_get_col_idx(obj, i)); + return 3 + names + cols; /* XT + attrs + XD */ + } + if (t == RAY_ERROR) { + const char *msg = ray_err_code(obj); + int64_t n = msg ? (int64_t)strlen(msg) : 0; + return 1 + n + 1; /* type byte + msg + null */ + } + if (t == RAY_NULL) + return 1 + 1 + 4; + + int esz = (int)ray_scalar_elem_size(t); + if (esz == 0) + return 0; + return 1 + 1 + 4 + obj->len * esz; +} + +static int64_t kdb_ser_obj(uint8_t *buf, ray_t *obj) { + uint8_t *start = buf; + + if (obj == NULL || obj == RAY_NULL_OBJ) { + *buf++ = 101; /* identity / null */ + *buf++ = 0; + return buf - start; + } + + int8_t t = obj->type; + *buf++ = (uint8_t)kdb_type_of(t); + + if (t < 0) { + int abs_t = -t; + switch (abs_t) { + case RAY_BOOL: + case RAY_U8: + *buf++ = obj->u8; + return buf - start; + case RAY_I16: + memcpy(buf, &obj->i16, 2); + return buf + 2 - start; + case RAY_I32: + case RAY_DATE: + case RAY_TIME: + memcpy(buf, &obj->i32, 4); + return buf + 4 - start; + case RAY_F32: { + float f = (float)obj->f64; + memcpy(buf, &f, 4); + return buf + 4 - start; + } + case RAY_I64: + case RAY_TIMESTAMP: + memcpy(buf, &obj->i64, 8); + return buf + 8 - start; + case RAY_F64: + memcpy(buf, &obj->f64, 8); + return buf + 8 - start; + case RAY_GUID: + memcpy(buf, ray_data(obj), 16); + return buf + 16 - start; + case RAY_SYM: { + ray_t *s = ray_sym_str(obj->i64); + size_t n = s ? ray_str_len(s) : 0; + if (s) { + memcpy(buf, ray_str_ptr(s), n); + ray_release(s); + } + buf[n] = 0; + return buf + n + 1 - start; + } + case RAY_STR: { + size_t n = ray_str_len(obj); + if (n == 1) { + *buf++ = (uint8_t)ray_str_ptr(obj)[0]; + return buf - start; + } + /* RAY_STR atom (n>1) → KC vector */ + start[0] = (uint8_t)KDB_KC; + *buf++ = 0; /* attrs */ + uint32_t len32 = (uint32_t)n; + memcpy(buf, &len32, 4); + buf += 4; + memcpy(buf, ray_str_ptr(obj), n); + return buf + n - start; + } + } + return -1; + } + + /* Containers/vectors */ + if (t == RAY_LIST) { + *buf++ = 0; + uint32_t len32 = (uint32_t)obj->len; + memcpy(buf, &len32, 4); + buf += 4; + ray_t **elems = (ray_t **)ray_data(obj); + for (int64_t i = 0; i < obj->len; i++) { + int64_t r = kdb_ser_obj(buf, elems[i]); + if (r < 0) + return -1; + buf += r; + } + return buf - start; + } + if (t == RAY_SYM) { + *buf++ = 0; + uint32_t len32 = (uint32_t)obj->len; + memcpy(buf, &len32, 4); + buf += 4; + int64_t *ids = (int64_t *)ray_data(obj); + for (int64_t i = 0; i < obj->len; i++) { + ray_t *s = ray_sym_str(ids[i]); + size_t n = s ? ray_str_len(s) : 0; + if (s) { + memcpy(buf, ray_str_ptr(s), n); + ray_release(s); + } + buf[n] = 0; + buf += n + 1; + } + return buf - start; + } + if (t == RAY_TABLE) { + int64_t ncols = ray_table_ncols(obj); + *buf++ = 0; /* attrs */ + *buf++ = (uint8_t)KDB_XD; /* table is dict-of-cols on the wire */ + + /* keys: KS vector of column names */ + *buf++ = (uint8_t)KDB_KS; + *buf++ = 0; /* attrs */ + uint32_t kn = (uint32_t)ncols; + memcpy(buf, &kn, 4); + buf += 4; + for (int64_t i = 0; i < ncols; i++) { + ray_t *s = ray_sym_str(ray_table_col_name(obj, i)); + size_t n = s ? ray_str_len(s) : 0; + if (s) { + memcpy(buf, ray_str_ptr(s), n); + ray_release(s); + } + buf[n] = 0; + buf += n + 1; + } + + /* values: general list of column vectors */ + *buf++ = 0; /* list type */ + *buf++ = 0; /* attrs */ + memcpy(buf, &kn, 4); + buf += 4; + for (int64_t i = 0; i < ncols; i++) { + int64_t r = kdb_ser_obj(buf, ray_table_get_col_idx(obj, i)); + if (r < 0) + return -1; + buf += r; + } + return buf - start; + } + /* RAY_DICT (type code 99) is never produced by v2; dicts present as + * RAY_LIST + RAY_ATTR_DICT and are serialized through the RAY_LIST path + * above (losing the dict-ness on the wire). */ + if (t == RAY_ERROR) { + const char *msg = ray_err_code(obj); + size_t n = msg ? strlen(msg) : 0; + if (n) + memcpy(buf, msg, n); + buf[n] = 0; + return buf + n + 1 - start; + } + if (t == RAY_NULL) { + *buf++ = 0; + memset(buf, 0, 4); + return buf + 4 - start; + } + + int esz = (int)ray_scalar_elem_size(t); + if (esz == 0) + return -1; + *buf++ = 0; + uint32_t len32 = (uint32_t)obj->len; + memcpy(buf, &len32, 4); + buf += 4; + size_t n = (size_t)obj->len * esz; + memcpy(buf, ray_data(obj), n); + return buf + n - start; +} + +/* ================================================================ + * Wire-format deserialize (KDB+ → rayforce v2) + * ================================================================ */ + +#define KDB_NEED(n) \ + do { \ + if (*len < (int64_t)(n)) \ + return ray_error("kdb: buffer underflow", NULL); \ + } while (0) + +/* Decode a width-byte atom and re-tag as the requested ray_type. Works + * because v2's atom union shares storage by width: BOOL/U8 → u8; + * I16 → i16; I32/DATE/TIME → i32; I64/TIMESTAMP → i64. */ +static ray_t *kdb_des_atom_i(uint8_t **buf, int64_t *len, int8_t ray_type, + int width) { + KDB_NEED(width); + ray_t *o = NULL; + switch (width) { + case 1: + o = ray_u8(**buf); + break; + case 2: { + int16_t v; + memcpy(&v, *buf, 2); + o = ray_i16(v); + break; + } + case 4: { + int32_t v; + memcpy(&v, *buf, 4); + o = ray_i32(v); + break; + } + case 8: { + int64_t v; + memcpy(&v, *buf, 8); + o = ray_i64(v); + break; + } + } + if (o) + o->type = -ray_type; /* tag with the requested ray atom type */ + *buf += width; + *len -= width; + return o; +} + +/* Read a kdb vector header: 1 byte attrs + 4 bytes int32 length. + * Advances *buf by 5 and decrements *len. Returns -1 on buffer underflow. */ +static int kdb_read_vec_header(uint8_t **buf, int64_t *len, int32_t *out_n) { + if (*len < 5) + return -1; + (*buf)++; + *len -= 1; /* attrs */ + memcpy(out_n, *buf, 4); + *buf += 4; + *len -= 4; + return 0; +} + +static ray_t *kdb_des_vec_i(uint8_t **buf, int64_t *len, int8_t ray_type, + int width) { + int32_t n; + if (kdb_read_vec_header(buf, len, &n) < 0) + return ray_error("kdb: buffer underflow", NULL); + if (n < 0) + return ray_error("kdb: negative vector length", NULL); + int64_t bytes = (int64_t)n * width; + if (*len < bytes) + return ray_error("kdb: buffer underflow (vec body)", NULL); + + ray_t *vec = ray_vec_new(ray_type, n); + if (vec == NULL || RAY_IS_ERR(vec)) { + if (vec) + ray_release(vec); + return ray_error("kdb: vector alloc failed", NULL); + } + memcpy(ray_data(vec), *buf, bytes); + vec->len = n; + *buf += bytes; + *len -= bytes; + return vec; +} + +static ray_t *kdb_des_obj(uint8_t **buf, int64_t *len) { + if (*len < 1) + return ray_error("kdb: buffer underflow (type)", NULL); + + int8_t type = (int8_t)**buf; + (*buf)++; + (*len)--; + + switch (type) { + /* Atoms (negative type). The deserializer reads raw bytes then re-tags. */ + case -KDB_KB: + return kdb_des_atom_i(buf, len, RAY_BOOL, 1); + case -KDB_KG: + return kdb_des_atom_i(buf, len, RAY_U8, 1); + case -KDB_KH: + return kdb_des_atom_i(buf, len, RAY_I16, 2); + case -KDB_KI: + return kdb_des_atom_i(buf, len, RAY_I32, 4); + case -KDB_KJ: + return kdb_des_atom_i(buf, len, RAY_I64, 8); + case -KDB_KP: + case -KDB_KN: + return kdb_des_atom_i(buf, len, RAY_TIMESTAMP, 8); + case -KDB_KD: + case -KDB_KM: + return kdb_des_atom_i(buf, len, RAY_DATE, 4); + case -KDB_KT: + case -KDB_KU: + case -KDB_KV: + return kdb_des_atom_i(buf, len, RAY_TIME, 4); + case -KDB_KE: { + KDB_NEED(4); + float f; + memcpy(&f, *buf, 4); + *buf += 4; + *len -= 4; + return ray_f64((double)f); + } + case -KDB_KF: { + KDB_NEED(8); + double f; + memcpy(&f, *buf, 8); + *buf += 8; + *len -= 8; + return ray_f64(f); + } + case -KDB_KC: { + KDB_NEED(1); + char c = (char)**buf; + *buf += 1; + *len -= 1; + return ray_str(&c, 1); + } + case -KDB_KS: { + int64_t n = 0; + while (n < *len && (*buf)[n] != '\0') + n++; + if (n >= *len) + return ray_error("kdb: symbol not null-terminated", NULL); + int64_t id = ray_sym_intern((const char *)*buf, (size_t)n); + *buf += n + 1; + *len -= n + 1; + return (id < 0) ? ray_error("kdb: symbol intern failed", NULL) + : ray_sym(id); + } + case -KDB_UU: { + KDB_NEED(16); + ray_t *g = ray_guid(*buf); + *buf += 16; + *len -= 16; + return g; + } + + /* Vectors */ + case KDB_KB: + return kdb_des_vec_i(buf, len, RAY_BOOL, 1); + case KDB_KG: + return kdb_des_vec_i(buf, len, RAY_U8, 1); + case KDB_KH: + return kdb_des_vec_i(buf, len, RAY_I16, 2); + case KDB_KI: + return kdb_des_vec_i(buf, len, RAY_I32, 4); + case KDB_KJ: + return kdb_des_vec_i(buf, len, RAY_I64, 8); + case KDB_KP: + case KDB_KN: + return kdb_des_vec_i(buf, len, RAY_TIMESTAMP, 8); + case KDB_KD: + case KDB_KM: + return kdb_des_vec_i(buf, len, RAY_DATE, 4); + case KDB_KT: + case KDB_KU: + case KDB_KV: + return kdb_des_vec_i(buf, len, RAY_TIME, 4); + case KDB_KZ: + return kdb_des_vec_i(buf, len, RAY_TIMESTAMP, 8); + case KDB_KE: { + /* Real (4-byte float) vector — convert to F64 vec for v2. */ + int32_t n; + if (kdb_read_vec_header(buf, len, &n) < 0) + return ray_error("kdb: buffer underflow", NULL); + if (n < 0) + return ray_error("kdb: negative real-vec length", NULL); + int64_t bytes = (int64_t)n * 4; + if (*len < bytes) + return ray_error("kdb: buffer underflow (real-vec)", NULL); + ray_t *vec = ray_vec_new(RAY_F64, n); + if (vec == NULL || RAY_IS_ERR(vec)) { + if (vec) + ray_release(vec); + return ray_error("kdb: vector alloc failed", NULL); + } + double *out = (double *)ray_data(vec); + for (int32_t i = 0; i < n; i++) { + float f; + memcpy(&f, *buf + i * 4, 4); + out[i] = (double)f; + } + vec->len = n; + *buf += bytes; + *len -= bytes; + return vec; + } + case KDB_KF: + return kdb_des_vec_i(buf, len, RAY_F64, 8); + case KDB_UU: + return kdb_des_vec_i(buf, len, RAY_GUID, 16); + case KDB_KC: { + /* KC vector → RAY_STR atom of length n */ + int32_t n; + if (kdb_read_vec_header(buf, len, &n) < 0) + return ray_error("kdb: buffer underflow", NULL); + if (n < 0) + return ray_error("kdb: negative char-vec length", NULL); + if (*len < n) + return ray_error("kdb: buffer underflow (char-vec)", NULL); + ray_t *s = ray_str((const char *)*buf, (size_t)n); + *buf += n; + *len -= n; + return s; + } + case KDB_KS: { + int32_t n; + if (kdb_read_vec_header(buf, len, &n) < 0) + return ray_error("kdb: buffer underflow", NULL); + if (n < 0) + return ray_error("kdb: negative symbol-vec length", NULL); + ray_t *vec = ray_sym_vec_new(RAY_SYM_W64, n); + if (vec == NULL || RAY_IS_ERR(vec)) { + if (vec) + ray_release(vec); + return ray_error("kdb: symbol vector alloc failed", NULL); + } + int64_t *ids = (int64_t *)ray_data(vec); + for (int32_t i = 0; i < n; i++) { + int64_t k = 0; + while (k < *len && (*buf)[k] != '\0') + k++; + if (k >= *len) { + ray_release(vec); + return ray_error("kdb: symbol not null-terminated in vec", NULL); + } + int64_t id = ray_sym_intern((const char *)*buf, (size_t)k); + if (id < 0) { + ray_release(vec); + return ray_error("kdb: symbol intern failed", NULL); + } + ids[i] = id; + *buf += k + 1; + *len -= k + 1; + } + vec->len = n; + return vec; + } + + case 0: { /* general list */ + int32_t n; + if (kdb_read_vec_header(buf, len, &n) < 0) + return ray_error("kdb: buffer underflow", NULL); + if (n < 0) + return ray_error("kdb: negative list length", NULL); + ray_t *list = ray_list_new(0); + for (int32_t i = 0; i < n; i++) { + ray_t *elem = kdb_des_obj(buf, len); + if (elem == NULL || RAY_IS_ERR(elem)) { + ray_release(list); + return elem ? elem : ray_error("kdb: list element decode failed", NULL); + } + list = ray_list_append(list, elem); + ray_release(elem); /* append retains its own ref; drop ours */ + if (list == NULL || RAY_IS_ERR(list)) + return list ? list : ray_error("kdb: list append failed", NULL); + } + return list; + } + + case KDB_XT: { /* table = attrs(0) + dict_marker(99) + keys + values */ + KDB_NEED(2); + (*buf) += 2; + *len -= 2; + ray_t *keys = kdb_des_obj(buf, len); + if (keys == NULL || RAY_IS_ERR(keys)) + return keys; + ray_t *vals = kdb_des_obj(buf, len); + if (vals == NULL || RAY_IS_ERR(vals)) { + ray_release(keys); + return vals; + } + return kdb_make_table(keys, vals); + } + + case KDB_XD: { /* dict = keys + values; could be a keyed table */ + ray_t *keys = kdb_des_obj(buf, len); + if (keys == NULL || RAY_IS_ERR(keys)) + return keys; + ray_t *vals = kdb_des_obj(buf, len); + if (vals == NULL || RAY_IS_ERR(vals)) { + ray_release(keys); + return vals; + } + /* Plain dict — represent as RAY_LIST + RAY_ATTR_DICT. */ + ray_t *dict = ray_list_new(0); + dict = ray_list_append(dict, keys); + ray_release(keys); /* append retains; drop our ref */ + dict = ray_list_append(dict, vals); + ray_release(vals); + if (dict && !RAY_IS_ERR(dict)) + dict->attrs |= RAY_ATTR_DICT; + return dict; + } + + case KDB_ERR: { + /* Error string is NUL-terminated on the wire; ensure the terminator is + * within the remaining buffer before handing it to ray_error (#H4 OOB). */ + const char *s = (const char *)*buf; + int64_t i = 0; + while (i < *len && s[i] != '\0') + i++; + if (i == *len) + return ray_error("kdb: malformed error frame", NULL); + return ray_error(s, NULL); + } + + default: + return ray_error("kdb: unsupported wire type", NULL); + } +} + +/* ================================================================ + * IPC decompression for compressed KDB+ responses. + * ================================================================ */ + +static int kdb_decompress(const uint8_t *src, int64_t src_len, + uint8_t **out_buf, int64_t *out_len) { + if (src_len < 4) + return -1; + + uint32_t header_size; + memcpy(&header_size, src, 4); + int64_t out_size = (int64_t)header_size - (int64_t)sizeof(kdb_header_t); + if (out_size <= 0) + return -1; + + uint32_t buffer[256] = {0}; + uint8_t *result = (uint8_t *)malloc((size_t)out_size); + if (result == NULL) + return -1; + + int64_t i = 0, n = 0, f = 0, s = 0, p = 0, d = 4; + while (s < out_size) { + if (i == 0) { + if (d >= src_len) { + free(result); + return -1; + } + f = src[d++]; + i = 1; + } + if (f & i) { + if (d >= src_len) { + free(result); + return -1; + } + int64_t r = buffer[src[d++]]; + result[s++] = result[r++]; + result[s++] = result[r++]; + if (d >= src_len) { + free(result); + return -1; + } + n = src[d++]; + for (int64_t m = 0; m < n; m++) + result[s + m] = result[r + m]; + } else { + if (d >= src_len) { + free(result); + return -1; + } + result[s++] = src[d++]; + } + while (p < s - 1) { + int64_t pp = p++; + buffer[result[pp] ^ result[p]] = (uint32_t)pp; + } + if (f & i) { + s += n; + p = s; + } + i *= 2; + if (i == 256) + i = 0; + } + *out_buf = result; + *out_len = out_size; + return 0; +} + +/* ================================================================ + * Plain-C entry points (callable from Rust via rayforce-sys). + * ================================================================ */ + +/* Connect + KDB+ handshake. Returns a connection slot (>= 0) or -1 on failure. */ +int64_t rkdb_connect(const char *host, int port) { + if (host == NULL || port < 1 || port > 65535) + return -1; + + int fd = kdb_open_socket(host, port); + if (fd < 0) + return -1; + + /* KDB+ handshake: send {0x03, 0x00}, read 1 byte. */ + uint8_t hello[2] = {0x03, 0x00}; + if (kdb_send_all(fd, hello, 2) < 0 || kdb_recv_all(fd, hello, 1) < 0) { + close(fd); + return -1; + } + + int slot = kdb_alloc_slot(fd); + if (slot < 0) { + close(fd); + return -1; + } + return (int64_t)slot; +} + +void rkdb_close(int64_t slot) { + kdb_init_table(); + if (slot < 0 || slot >= KDB_MAX_CONNS || kdb_conn_fd[slot] < 0) + return; + close(kdb_conn_fd[(int)slot]); + kdb_conn_fd[(int)slot] = -1; +} + +/* Send `msg` (any rayforce object, typically a RAY_STR query) to the KDB+ + * server on `slot` and return the response decoded into a rayforce object. + * Returns a RAY_ERROR object on any failure (never NULL). */ +ray_t *rkdb_send(int64_t slot, ray_t *msg) { + kdb_init_table(); + if (slot < 0 || slot >= KDB_MAX_CONNS || kdb_conn_fd[slot] < 0) + return ray_error("kdb: invalid handle", NULL); + int fd = kdb_conn_fd[(int)slot]; + + int64_t body_size = kdb_size_obj(msg); + if (body_size <= 0) + return ray_error("kdb: cannot serialize message", NULL); + + size_t total = sizeof(kdb_header_t) + (size_t)body_size; + uint8_t *buf = (uint8_t *)malloc(total); + if (buf == NULL) + return ray_error("kdb: out of memory", NULL); + + int64_t written = kdb_ser_obj(buf + sizeof(kdb_header_t), msg); + if (written < 0) { + free(buf); + return ray_error("kdb: serialization failed", NULL); + } + kdb_header_t header = { + .endianness = 1, + .msgtype = KDB_MSG_SYNC, + .compressed = 0, + .reserved = 0, + .size = (uint32_t)(written + sizeof(kdb_header_t)), + }; + memcpy(buf, &header, sizeof(header)); + + if (kdb_send_all(fd, buf, header.size) < 0) { + free(buf); + return ray_error("kdb: send failed", NULL); + } + free(buf); + + /* Receive response. */ + if (kdb_recv_all(fd, &header, sizeof(header)) < 0) + return ray_error("kdb: recv header failed", NULL); + int64_t body_len = (int64_t)header.size - (int64_t)sizeof(header); + if (body_len <= 0) + return ray_error("kdb: empty response body", NULL); + + uint8_t *body = (uint8_t *)malloc((size_t)body_len); + if (body == NULL) + return ray_error("kdb: out of memory", NULL); + if (kdb_recv_all(fd, body, (size_t)body_len) < 0) { + free(body); + return ray_error("kdb: recv body failed", NULL); + } + + uint8_t *decoded = body; + int64_t decoded_len = body_len; + uint8_t *decompressed = NULL; + if (header.compressed) { + if (kdb_decompress(body, body_len, &decompressed, &decoded_len) < 0) { + free(body); + return ray_error("kdb: decompression failed", NULL); + } + decoded = decompressed; + } + + uint8_t *cursor = decoded; + int64_t remaining = decoded_len; + ray_t *result = kdb_des_obj(&cursor, &remaining); + free(body); + if (decompressed) + free(decompressed); + if (result == NULL) + return ray_error("kdb: deserialization returned null", NULL); + return result; +} diff --git a/rayforce-sys/src/lib.rs b/rayforce-sys/src/lib.rs index 8aeebaf..dc86ab3 100644 --- a/rayforce-sys/src/lib.rs +++ b/rayforce-sys/src/lib.rs @@ -14,3 +14,15 @@ #![allow(clippy::all)] include!(concat!(env!("OUT_DIR"), "/bindings.rs")); + +// Vendored KDB+ IPC client (csrc/kdb_ipc.c). Not part of the engine bindings; +// hand-declared here until it moves into the engine proper. +extern "C" { + /// Connect + KDB+ handshake. Returns a connection slot (>= 0) or -1. + pub fn rkdb_connect(host: *const ::std::os::raw::c_char, port: ::std::os::raw::c_int) -> i64; + /// Send a rayforce object (typically a RAY_STR query) and return the decoded + /// response, or a RAY_ERROR object on failure (never null). + pub fn rkdb_send(slot: i64, msg: *mut ray_t) -> *mut ray_t; + /// Close a connection slot (no-op if invalid). + pub fn rkdb_close(slot: i64); +} diff --git a/rayforce/src/ipc.rs b/rayforce/src/ipc.rs index ba35adb..42cd52a 100644 --- a/rayforce/src/ipc.rs +++ b/rayforce/src/ipc.rs @@ -34,8 +34,9 @@ impl TcpClient { CString::new(password).map_err(|_| RayError::binding("password contains NUL"))?; unsafe { ensure_poll(); + // engine added a 5th arg (timeout_ms) after commit a7294066; 0 = block. let handle = - sys::ray_ipc_connect(host_c.as_ptr(), port, user_c.as_ptr(), pass_c.as_ptr()); + sys::ray_ipc_connect(host_c.as_ptr(), port, user_c.as_ptr(), pass_c.as_ptr(), 0); if handle < 0 { return Err(RayError::binding(format!( "connect to {host}:{port} failed" diff --git a/rayforce/src/kdb.rs b/rayforce/src/kdb.rs new file mode 100644 index 0000000..a1703ee --- /dev/null +++ b/rayforce/src/kdb.rs @@ -0,0 +1,60 @@ +//! KDB+ IPC client — connect to a kdb+ server, send a query, get the response +//! decoded into a rayforce [`Value`]. +//! +//! The wire protocol (handshake, serialize/deserialize, decompression) is a C +//! client vendored into `rayforce-sys` (`csrc/kdb_ipc.c`, originally from +//! `rayforce-py`) — staged there until it moves into the engine proper. Sync +//! request/reply only (no async/subscribe), which is all the RevoLT-style +//! pull-by-sequence pattern needs. Requires a live [`crate::Runtime`] (decoded +//! responses are engine-allocated objects). +//! +//! ```no_run +//! use rayforce::{Runtime, kdb::KdbConnection}; +//! let _rt = Runtime::new().unwrap(); +//! let conn = KdbConnection::connect("localhost", 5010).unwrap(); +//! let fills = conn.execute("select from fixmsgs where i > 0").unwrap(); +//! ``` + +use std::ffi::CString; + +use rayforce_sys as sys; + +use crate::error::{check, RayError, Result}; +use crate::value::Value; + +/// An open connection to a kdb+ server. Closed on drop. +pub struct KdbConnection { + slot: i64, +} + +impl KdbConnection { + /// Open a TCP connection and perform the KDB+ handshake. + pub fn connect(host: &str, port: u16) -> Result { + let host_c = CString::new(host).map_err(|_| RayError::binding("kdb host contains NUL"))?; + let slot = unsafe { sys::rkdb_connect(host_c.as_ptr(), i32::from(port)) }; + if slot < 0 { + return Err(RayError::binding(format!( + "kdb: connect to {host}:{port} failed" + ))); + } + Ok(KdbConnection { slot }) + } + + /// Send a query string for remote evaluation; return the response decoded + /// into a [`Value`] (atoms, vectors, tables, …). A kdb-side error surfaces + /// as `Err`. + pub fn execute(&self, query: &str) -> Result { + let q = Value::string(query); + // rkdb_send borrows `q` (serializes it), returns an owned result or a + // RAY_ERROR object (never null); `check` converts the error object. + let res = unsafe { sys::rkdb_send(self.slot, q.as_ptr()) }; + let ok = unsafe { check(res)? }; + Ok(unsafe { Value::from_owned(ok) }) + } +} + +impl Drop for KdbConnection { + fn drop(&mut self) { + unsafe { sys::rkdb_close(self.slot) }; + } +} diff --git a/rayforce/src/lib.rs b/rayforce/src/lib.rs index 311d593..5089095 100644 --- a/rayforce/src/lib.rs +++ b/rayforce/src/lib.rs @@ -16,6 +16,7 @@ mod dict; mod error; mod expr; mod ipc; +pub mod kdb; mod list; mod ops; mod query; @@ -32,6 +33,7 @@ pub use expr::{ avg, col, count, distinct, first, last, lit, max, median, min, sum, Expr, IntoExpr, }; pub use ipc::TcpClient; +pub use kdb::KdbConnection; pub use ops::Operation; pub use query::{Select, Update}; pub use runtime::{eval, eval_value, get_global, is_live, set_global, Runtime}; diff --git a/rayforce/src/raw.rs b/rayforce/src/raw.rs index 8bbc47f..a3d38b9 100644 --- a/rayforce/src/raw.rs +++ b/rayforce/src/raw.rs @@ -42,6 +42,29 @@ pub(crate) unsafe fn set_attrs(v: Raw, a: u8) { (*v).__bindgen_anon_1.as_mut().attrs = a; } +/// Resolution domain of a `RAY_SYM` vector (mirrors the engine's static-inline +/// `ray_sym_vec_domain`, which bindgen does not emit): follow a slice to its +/// parent, read the `sym_domain` aux field, and fall back to the runtime domain +/// when unset. A `RAY_SYM` cell is a *position* in this domain — for FILE +/// domains (e.g. a loaded splayed column) positions are not runtime intern ids +/// and must be mapped via `ray_sym_domain_runtime_lut`. +#[inline] +pub(crate) unsafe fn sym_domain(v: Raw) -> *mut sys::ray_sym_domain_s { + let h = hdr(v); + if h.attrs & sys::RAY_ATTR_SLICE as u8 != 0 { + let parent = h.__bindgen_anon_1.__bindgen_anon_1.slice_parent; + if !parent.is_null() { + return sym_domain(parent); + } + } + let d = h.__bindgen_anon_1.__bindgen_anon_3.sym_domain; + if d.is_null() { + sys::ray_sym_runtime_domain() + } else { + d + } +} + #[inline] pub(crate) unsafe fn rc(v: Raw) -> u32 { hdr(v).rc diff --git a/rayforce/src/vector.rs b/rayforce/src/vector.rs index cad4e93..77bb576 100644 --- a/rayforce/src/vector.rs +++ b/rayforce/src/vector.rs @@ -223,9 +223,18 @@ impl Value { sys::RAY_TIMESTAMP => Value::timestamp_nanos(*(base.add(idx * 8) as *const i64)), sys::RAY_GUID => Value::guid(&*(base.add(idx * 16) as *const [u8; 16])), sys::RAY_SYM => { - // Box directly from the id — lossless and avoids a - // string round-trip / re-intern. - let id = sys::ray_vec_get_sym_id(self.as_ptr(), idx as i64); + // A cell is a *position* in the vector's symbol domain, not a + // runtime intern id. For the runtime domain the LUT is NULL + // and positions pass through; for FILE domains (loaded + // splayed columns) the LUT maps position -> runtime id. + let pos = sys::ray_vec_get_sym_id(self.as_ptr(), idx as i64); + let dom = raw::sym_domain(self.as_ptr()); + let lut = sys::ray_sym_domain_runtime_lut(dom); + let id = if lut.is_null() { + pos + } else { + *lut.offset(pos as isize) + }; Value::from_owned(crate::error::check(sys::ray_sym(id))?) } sys::RAY_STR => { diff --git a/rayforce/tests/kdb.rs b/rayforce/tests/kdb.rs new file mode 100644 index 0000000..7d9567a --- /dev/null +++ b/rayforce/tests/kdb.rs @@ -0,0 +1,112 @@ +//! Verify the vendored KDB+ IPC client against a mock kdb server (raw TCP that +//! speaks the kdb wire protocol). No `q` binary required. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::thread; + +use rayforce::{kdb::KdbConnection, Runtime, Table}; + +// --- kdb wire-format builders (server side) ------------------------------- + +/// Long (i64) vector: type 7, attrs, int32 len, data. +fn long_vec(vals: &[i64]) -> Vec { + let mut b = vec![7u8, 0u8]; + b.extend_from_slice(&(vals.len() as i32).to_le_bytes()); + for v in vals { + b.extend_from_slice(&v.to_le_bytes()); + } + b +} + +/// Symbol vector: type 11, attrs, int32 len, NUL-terminated names. +fn sym_vec(syms: &[&str]) -> Vec { + let mut b = vec![11u8, 0u8]; + b.extend_from_slice(&(syms.len() as i32).to_le_bytes()); + for s in syms { + b.extend_from_slice(s.as_bytes()); + b.push(0); + } + b +} + +/// Table: type 98, attrs(0), dict-marker(99), KS-keys, general-list of columns. +fn table(names: &[&str], cols: &[Vec]) -> Vec { + let mut b = vec![98u8, 0u8, 99u8]; + b.extend(sym_vec(names)); + b.push(0u8); // general list + b.push(0u8); // attrs + b.extend_from_slice(&(cols.len() as i32).to_le_bytes()); + for c in cols { + b.extend_from_slice(c); + } + b +} + +/// Wrap a serialized object as a kdb response message (8-byte header + body). +fn msg(body: &[u8]) -> Vec { + let size = (8 + body.len()) as u32; + let mut b = vec![1u8, 2u8, 0u8, 0u8]; // little-endian, response, uncompressed + b.extend_from_slice(&size.to_le_bytes()); + b.extend_from_slice(body); + b +} + +/// Bind a mock kdb server that answers one request with `response`; returns the port. +fn spawn_mock(response: Vec) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + thread::spawn(move || { + if let Ok((mut sock, _)) = listener.accept() { + // handshake: client sends 2 bytes, server replies with 1. + let mut hs = [0u8; 2]; + let _ = sock.read_exact(&mut hs); + let _ = sock.write_all(&[3u8]); + // read the query message (header then body) and discard it. + let mut hdr = [0u8; 8]; + if sock.read_exact(&mut hdr).is_ok() { + let size = u32::from_le_bytes([hdr[4], hdr[5], hdr[6], hdr[7]]) as usize; + let mut body = vec![0u8; size.saturating_sub(8)]; + let _ = sock.read_exact(&mut body); + } + let _ = sock.write_all(&response); + } + }); + port +} + +#[test] +fn kdb_pulls_a_table() { + let _rt = Runtime::new().unwrap(); + + let response = msg(&table( + &["seq", "sym"], + &[long_vec(&[1, 2, 3]), sym_vec(&["AAPL", "MSFT", "GOOG"])], + )); + let port = spawn_mock(response); + + let conn = KdbConnection::connect("127.0.0.1", port).unwrap(); + let v = conn.execute("select from fixmsgs where i > 0").unwrap(); + + let t = Table::from_value(v).unwrap(); + assert_eq!(t.shape(), (3, 2)); + assert_eq!( + t.column("seq").unwrap().as_slice::().unwrap(), + &[1, 2, 3] + ); + let sym = t.column("sym").unwrap(); + assert_eq!(sym.get(0).unwrap().as_sym().unwrap(), "AAPL"); + assert_eq!(sym.get(2).unwrap().as_sym().unwrap(), "GOOG"); +} + +#[test] +fn kdb_surfaces_server_error() { + let _rt = Runtime::new().unwrap(); + // kdb error frame: type -128 then a NUL-terminated message. + let mut body = vec![(-128i8) as u8]; + body.extend_from_slice(b"type\0"); + let port = spawn_mock(msg(&body)); + + let conn = KdbConnection::connect("127.0.0.1", port).unwrap(); + assert!(conn.execute("1+`a").is_err()); +} diff --git a/rayforce/tests/kdb_real.rs b/rayforce/tests/kdb_real.rs new file mode 100644 index 0000000..e86590d --- /dev/null +++ b/rayforce/tests/kdb_real.rs @@ -0,0 +1,53 @@ +//! End-to-end check of the vendored KDB+ IPC client against a *real* `q` +//! server (not the mock in `kdb.rs`). Gated on `RAYFORCE_KDB_ADDR=host:port` +//! so it's a no-op when no kdb endpoint is available (CI, no `q` binary). +//! +//! Start a server first, e.g.: +//! QHOME=~/lynx/q/m64 ~/lynx/q/m64/m64/q fixsrv.q -q +//! then: +//! RAYFORCE_KDB_ADDR=127.0.0.1:5010 cargo test -p rayforce --test kdb_real -- --nocapture + +use rayforce::{kdb::KdbConnection, Runtime, Table}; + +fn addr() -> Option<(String, u16)> { + let a = std::env::var("RAYFORCE_KDB_ADDR").ok()?; + let (h, p) = a.rsplit_once(':')?; + Some((h.to_string(), p.parse().ok()?)) +} + +#[test] +fn real_q_roundtrips_atoms_vectors_and_tables() { + let Some((host, port)) = addr() else { + eprintln!("RAYFORCE_KDB_ADDR unset — skipping real-q e2e"); + return; + }; + let _rt = Runtime::new().unwrap(); + let conn = KdbConnection::connect(&host, port).unwrap(); + + // scalar + let v = conn.execute("ping 41").unwrap(); + assert_eq!(v.format(), "42"); + + // vector + let v = conn.execute("exec seq from fixmsgs").unwrap(); + assert_eq!(v.as_slice::().unwrap(), &[1, 2, 3, 4, 5]); + + // full table + let t = Table::from_value(conn.execute("select from fixmsgs").unwrap()).unwrap(); + assert_eq!(t.shape(), (5, 4)); + assert_eq!( + t.column("seq").unwrap().as_slice::().unwrap(), + &[1, 2, 3, 4, 5] + ); + let sym = t.column("sym").unwrap(); + assert_eq!(sym.get(0).unwrap().as_sym().unwrap(), "AAPL"); + assert_eq!(sym.get(4).unwrap().as_sym().unwrap(), "TSLA"); + + // RevoLT-style pull-by-sequence: only rows past a cursor + let t = Table::from_value(conn.execute("select from fixmsgs where seq > 3").unwrap()).unwrap(); + assert_eq!(t.shape(), (2, 4)); + assert_eq!(t.column("seq").unwrap().as_slice::().unwrap(), &[4, 5]); + + // kdb-side error surfaces as Err + assert!(conn.execute("1+`a").is_err()); +} diff --git a/rayforce/tests/table.rs b/rayforce/tests/table.rs index df90248..83c8435 100644 --- a/rayforce/tests/table.rs +++ b/rayforce/tests/table.rs @@ -122,3 +122,42 @@ fn splayed_roundtrip() { let _ = std::fs::remove_dir_all(&base); } + +#[test] +fn splayed_sym_values_roundtrip() { + // A SYM column loaded from a splayed table must resolve to its original + // values. The cells are positions in the column's *file* symbol domain; + // resolving them against the runtime domain (the pre-fix behaviour) yields + // garbage (e.g. "+"). `splayed_roundtrip` above only checks an i64 column, + // so it never exercised symbol resolution. + let _rt = Runtime::new().unwrap(); + let t = Table::new( + &["k", "v"], + &[ + Value::sym_vec(&["abcdef123456", "xyz"]), + Value::vec(&[1i64, 2]), + ], + ) + .unwrap(); + + let base = std::env::temp_dir().join(format!("rayforce_rs_splay_sym_{}", std::process::id())); + let _ = std::fs::remove_dir_all(&base); + let dir = base.join("t"); + std::fs::create_dir_all(&dir).unwrap(); + let dir_str = dir.to_str().unwrap(); + + // /.sym dotfile convention (no explicit sym path). + t.save_splayed(dir_str, None).unwrap(); + let loaded = Table::load_splayed(dir_str, None).unwrap(); + + assert_eq!(loaded.shape(), (2, 2)); + let k = loaded.column("k").unwrap(); + assert_eq!(k.get(0).unwrap().as_sym().unwrap(), "abcdef123456"); + assert_eq!(k.get(1).unwrap().as_sym().unwrap(), "xyz"); + assert_eq!( + loaded.column("v").unwrap().as_slice::().unwrap(), + &[1, 2] + ); + + let _ = std::fs::remove_dir_all(&base); +}