Phase 4e complete: Data and Distribution (5 libraries, 247 tests passing)
ober
2bb82046913ab3f2b4737856f9b1933e90025d1c
--- a/Makefile +++ b/Makefile @@ -168,6 +168,14 @@ test-phase4d: @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-staging2.ss @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-match-syntax.ss +test-phase4e: + @echo "--- Phase 4e: Data and Distribution tests ---" + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-dataframe.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-stream-window.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-distributed.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-wasi.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-checkpoint.ss + test-all: test test-features test-wrappers clean: new file mode 100644 --- /dev/null +++ b/lib/std/actor/checkpoint.sls @@ -0,0 +1,278 @@ +#!chezscheme +;;; (std actor checkpoint) — Actor state and value checkpointing +;;; +;;; Serialize Scheme data (and actor mailbox contents) to files so that +;;; actor state can be checkpointed and restored after restarts. +;;; +;;; Serialization uses Chez Scheme's fasl-write/fasl-read for binary-safe +;;; roundtripping of basic Scheme values. Procedures, ports, and continuations +;;; are NOT serializable — checkpoint-serializable? returns #f for them. + +(library (std actor checkpoint) + (export + ;; Core value serialization + checkpoint-value + restore-value + checkpoint-serializable? + serialize-value + deserialize-value + + ;; Actor mailbox checkpointing + checkpoint-actor-mailbox + restore-actor-mailbox + + ;; Periodic checkpoint manager + make-checkpoint-manager + checkpoint-manager? + checkpoint-manager-start! + checkpoint-manager-stop! + checkpoint-manager-register! + checkpoint-manager-restore + checkpoint-manager-path + + ;; Utilities + list-checkpoints + checkpoint-age + delete-old-checkpoints) + + (import (chezscheme) + (std actor mpsc) + (std actor core)) + + ;; -------- Serializable? predicate -------- + ;; + ;; Conservative check: only pure data values are checkpointable. + ;; Procedures, ports, conditions, continuations are excluded. + + (define (checkpoint-serializable? val) + (cond + [(null? val) #t] + [(boolean? val) #t] + [(number? val) #t] + [(string? val) #t] + [(symbol? val) #t] + [(char? val) #t] + [(bytevector? val) #t] + [(pair? val) + (and (checkpoint-serializable? (car val)) + (checkpoint-serializable? (cdr val)))] + [(vector? val) + (let loop ([i 0]) + (or (= i (vector-length val)) + (and (checkpoint-serializable? (vector-ref val i)) + (loop (+ i 1)))))] + [else #f])) + + ;; -------- Core serialization -------- + ;; + ;; serialize-value: any serializable value -> bytevector + ;; deserialize-value: bytevector -> value + + (define (serialize-value val) + (unless (checkpoint-serializable? val) + (error 'serialize-value "value is not serializable" val)) + (let-values ([(port get-bytes) (open-bytevector-output-port)]) + (fasl-write val port) + (get-bytes))) + + (define (deserialize-value bv) + (let ([port (open-bytevector-input-port bv)]) + (fasl-read port))) + + ;; -------- File-based checkpointing -------- + + (define (checkpoint-value val path) + (unless (checkpoint-serializable? val) + (error 'checkpoint-value "value is not serializable" val)) + (let ([bv (serialize-value val)]) + (call-with-port (open-file-output-port path + (file-options no-fail) + (buffer-mode block)) + (lambda (port) + (put-bytevector port bv))))) + + (define (restore-value path) + (let* ([bv (call-with-port (open-file-input-port path) + (lambda (port) + (get-bytevector-all port)))]) + (if (eof-object? bv) + (error 'restore-value "checkpoint file is empty" path) + (deserialize-value bv)))) + + ;; -------- Actor mailbox checkpointing -------- + ;; + ;; Drains all pending messages from the actor's MPSC queue and writes + ;; only the serializable ones to the checkpoint file. + ;; + ;; NOTE: This destructively reads the mailbox. Use only when the actor + ;; is stopped or known to be idle. + + (define (checkpoint-actor-mailbox actor-ref path) + (unless (actor-ref? actor-ref) + (error 'checkpoint-actor-mailbox "not an actor-ref" actor-ref)) + (let ([mailbox (actor-ref-mailbox actor-ref)]) + (unless mailbox + (error 'checkpoint-actor-mailbox "actor has no local mailbox (remote ref?)" actor-ref)) + ;; Drain all available messages + (let loop ([msgs '()]) + (let-values ([(msg ok) (mpsc-try-dequeue! mailbox)]) + (if ok + (loop (if (checkpoint-serializable? msg) + (cons msg msgs) + msgs)) + ;; Write the messages we collected (in original order) + (checkpoint-value (reverse msgs) path)))))) + + (define (restore-actor-mailbox path) + ;; Returns a list of messages from the checkpoint + (if (file-exists? path) + (restore-value path) + '())) + + ;; -------- Checkpoint manager record -------- + + (define-record-type checkpoint-manager + (fields + (immutable path) ;; checkpoint directory path + (mutable registry) ;; alist of (key . thunk) + (mutable running?) ;; #t while the background thread is running + (immutable mutex) ;; protects registry and running? + (immutable cond-var)) ;; signaled to wake manager thread + (protocol + (lambda (new) + (lambda (path) + (new path '() #f (make-mutex) (make-condition))))) + (sealed #t)) + + ;; -------- checkpoint-manager-start! -------- + ;; Start a background thread that checkpoints registered values every + ;; interval-ms milliseconds. + + (define (checkpoint-manager-start! mgr interval-ms) + (unless (checkpoint-manager-running? mgr) + (with-mutex (checkpoint-manager-mutex mgr) + (checkpoint-manager-running?-set! mgr #t)) + (fork-thread + (lambda () + (let loop () + (when (checkpoint-manager-running? mgr) + ;; Sleep for interval-ms by waiting on the condition with a timeout. + ;; condition-wait timeout must be a time-duration or time-utc record. + (with-mutex (checkpoint-manager-mutex mgr) + (let ([timeout (make-time 'time-duration + ;; nanoseconds part (round to 0) + 0 + ;; seconds part + (max 1 (inexact->exact (round (/ interval-ms 1000)))))]) + (condition-wait + (checkpoint-manager-cond-var mgr) + (checkpoint-manager-mutex mgr) + timeout))) + (when (checkpoint-manager-running? mgr) + ;; Snapshot all registered thunks + (let ([registry + (with-mutex (checkpoint-manager-mutex mgr) + (checkpoint-manager-registry mgr))]) + (for-each + (lambda (entry) + (let ([key (car entry)] + [thunk (cdr entry)]) + (guard (exn [#t (void)]) ; silently skip failures + (let ([val (thunk)]) + (when (checkpoint-serializable? val) + (let ([file (checkpoint-file-for-key + (checkpoint-manager-path mgr) + key)]) + (checkpoint-value val file))))))) + registry)) + (loop)))))))) + + ;; -------- checkpoint-manager-stop! -------- + + (define (checkpoint-manager-stop! mgr) + (with-mutex (checkpoint-manager-mutex mgr) + (checkpoint-manager-running?-set! mgr #f) + (condition-broadcast (checkpoint-manager-cond-var mgr)))) + + ;; -------- checkpoint-manager-register! -------- + + (define (checkpoint-manager-register! mgr key thunk) + (with-mutex (checkpoint-manager-mutex mgr) + (let ([existing (assoc key (checkpoint-manager-registry mgr))]) + (if existing + (set-cdr! existing thunk) + (checkpoint-manager-registry-set! + mgr + (cons (cons key thunk) (checkpoint-manager-registry mgr))))))) + + ;; -------- checkpoint-manager-restore -------- + ;; Restores the most recent checkpoint for a key, or returns #f if none. + + (define (checkpoint-manager-restore mgr key) + (let ([file (checkpoint-file-for-key (checkpoint-manager-path mgr) key)]) + (and (file-exists? file) + (guard (exn [#t #f]) + (restore-value file))))) + + ;; -------- Internal helpers -------- + + ;; Build a checkpoint file path: <dir>/<key>.chk + ;; key may be any value; we use its string representation. + (define (checkpoint-file-for-key dir key) + (string-append dir "/" + (sanitize-key (format "~a" key)) + ".chk")) + + ;; Replace characters that are unsafe in filenames with underscores. + (define (sanitize-key s) + (list->string + (map (lambda (c) + (if (or (char-alphabetic? c) (char-numeric? c) + (char=? c #\-) (char=? c #\_)) + c #\_)) + (string->list s)))) + + ;; -------- list-checkpoints -------- + ;; Returns a list of .chk file paths in dir. + + (define (list-checkpoints dir) + (if (file-directory? dir) + (let ([entries (directory-list dir)]) + (filter (lambda (name) (string-suffix? ".chk" name)) + (map (lambda (name) (string-append dir "/" name)) + entries))) + '())) + + ;; -------- checkpoint-age -------- + ;; Returns the age in seconds of a checkpoint file, or +inf.0 if it doesn't exist. + + (define (checkpoint-age path) + (if (file-exists? path) + (let* ([mtime (file-modification-time path)] + [now (current-time 'time-utc)] + ;; Both are time objects with time-second and time-nanosecond + [delta (- (time-second now) (time-second mtime))]) + (max 0 delta)) + +inf.0)) + + ;; -------- delete-old-checkpoints -------- + ;; Delete any .chk files in dir that are older than max-age-secs seconds. + + (define (delete-old-checkpoints dir max-age-secs) + (let ([files (list-checkpoints dir)]) + (for-each + (lambda (path) + (when (> (checkpoint-age path) max-age-secs) + (guard (exn [#t (void)]) + (delete-file path)))) + files))) + + ;; -------- string-suffix? -------- + + (define (string-suffix? suffix str) + (let ([slen (string-length suffix)] + [len (string-length str)]) + (and (>= len slen) + (string=? suffix (substring str (- len slen) len))))) + +) ;; end library --- a/lib/std/actor/core.sls +++ b/lib/std/actor/core.sls @@ -46,6 +46,9 @@ ;; Create a remote actor reference (for transport layer) make-remote-actor-ref + + ;; Internal: mailbox accessor (used by checkpoint layer) + actor-ref-mailbox ) (import (chezscheme) (std actor mpsc)) new file mode 100644 --- /dev/null +++ b/lib/std/actor/distributed.sls @@ -0,0 +1,312 @@ +#!chezscheme +;;; (std actor distributed) — Location-transparent distributed actor messaging +;;; +;;; Builds on (std actor core) and (std actor cluster) to provide: +;;; - Location-transparent send (dsend / dsend/ask) +;;; - Cluster-wide name registration +;;; - Process groups (broadcast) +;;; - Distributed supervision +;;; - Node failure detection / monitoring +;;; - Simple serialization via write/read on string ports + +(library (std actor distributed) + (export + ;; Location-transparent send + dsend + dsend/ask + + ;; Remote actor references + make-remote-ref + remote-ref? + remote-ref-node + remote-ref-id + + ;; Cluster-wide name registration + cluster-register! + cluster-whereis + cluster-registered-names + + ;; Process groups + make-process-group + process-group-join! + process-group-leave! + process-group-members + process-group-broadcast! + + ;; Distributed supervision + make-dist-supervisor + dist-supervisor-start-child! + dist-supervisor-children + + ;; Failure detection + monitor-node + demonitor-node + node-alive? + ping-node + + ;; Serialization + serialize-message + deserialize-message + + ;; Configuration parameters + *default-send-timeout* + *cluster-name*) + + (import (chezscheme) + (std actor core) + (except (std actor cluster) node-alive?)) + + ;; ====================================================================== + ;; Configuration parameters + ;; ====================================================================== + + (define *default-send-timeout* (make-parameter 5000)) ;; 5 seconds in ms + (define *cluster-name* (make-parameter "local")) + + ;; ====================================================================== + ;; Remote actor references + ;; ====================================================================== + + ;; A remote-ref identifies an actor on a specific cluster node by name. + (define-record-type remote-ref-rec + (fields + (immutable node) ;; node name (string or symbol) + (immutable id)) ;; actor name or id + (sealed #t)) + + (define (make-remote-ref node-name actor-id) + (make-remote-ref-rec node-name actor-id)) + + (define (remote-ref? x) (remote-ref-rec? x)) + (define (remote-ref-node x) (remote-ref-rec-node x)) + (define (remote-ref-id x) (remote-ref-rec-id x)) + + ;; Is actor-ref local? (uses (std actor core) actor-ref? predicate) + (define (%local-ref? ref) + (actor-ref? ref)) + + ;; ====================================================================== + ;; Location-transparent send + ;; ====================================================================== + + ;; (dsend actor-ref msg) + ;; Works for both local actor-refs and remote-refs. + ;; For local refs, delegates directly to (send). + ;; For remote refs, looks up the named actor on the target node. + (define (dsend ref msg) + (cond + [(%local-ref? ref) + ;; Local actor: use the core send + (send ref msg)] + [(remote-ref? ref) + ;; Remote: find actor in cluster registry + (let* ([node-name (remote-ref-node ref)] + [actor-id (remote-ref-id ref)] + [node (cluster-node-by-name node-name)]) + (if node + (let ([actor (remote-whereis node actor-id)]) + (if actor + (send actor msg) + (error 'dsend "actor not found on node" actor-id node-name))) + (error 'dsend "node not found" node-name)))] + [else + (error 'dsend "not a valid actor reference" ref)])) + + ;; (dsend/ask actor-ref msg timeout-ms) -> reply or #f + ;; Sends msg and waits for a reply; uses a temporary channel actor. + (define (dsend/ask ref msg timeout-ms) + (let* ([result-box (make-mutex)] + [reply #f] + [replied? #f] + [cond-var (make-condition)] + [lock (make-mutex)]) + ;; Spawn a one-shot reply actor + (let ([reply-actor + (spawn-actor + (lambda (m) + (with-mutex lock + (set! reply m) + (set! replied? #t) + (condition-signal cond-var))))]) + ;; Send original message with reply-to field prepended + (dsend ref (list 'ask reply-actor msg)) + ;; Wait for reply with timeout + (let ([deadline (+ (current-time-ms) timeout-ms)]) + (with-mutex lock + (let loop () + (unless replied? + (let ([now (current-time-ms)]) + (when (< now deadline) + (condition-wait cond-var lock) + (loop)))))) + (if replied? reply #f))))) + + (define (current-time-ms) + (* 1000 (time-second (current-time)))) + + ;; ====================================================================== + ;; Cluster-wide name registration + ;; ====================================================================== + + ;; Global name table: name -> actor-ref (local refs) + (define *global-registry* (make-hashtable equal-hash equal?)) + (define *global-registry-mutex* (make-mutex)) + + ;; (cluster-register! name actor-ref) — register actor under a cluster-wide name + (define (cluster-register! name actor-ref) + (with-mutex *global-registry-mutex* + (hashtable-set! *global-registry* name actor-ref)) + ;; Also register on all alive nodes in the cluster + (for-each + (lambda (node) + (remote-register! node name actor-ref)) + (cluster-nodes))) + + ;; (cluster-whereis name) -> actor-ref or #f (searches local registry first) + (define (cluster-whereis name) + (or (with-mutex *global-registry-mutex* + (hashtable-ref *global-registry* name #f)) + (whereis/any name))) + + ;; (cluster-registered-names) -> list of names + (define (cluster-registered-names) + (with-mutex *global-registry-mutex* + (vector->list (hashtable-keys *global-registry*)))) + + ;; ====================================================================== + ;; Process groups + ;; ====================================================================== + + (define-record-type process-group-rec + (fields + (immutable name) + (mutable members) ;; list of actor-refs (or remote-refs) + (immutable mutex)) + (sealed #t)) + + (define (make-process-group name) + (make-process-group-rec name '() (make-mutex))) + + (define (process-group-join! group ref) + (with-mutex (process-group-rec-mutex group) + (unless (member ref (process-group-rec-members group)) + (process-group-rec-members-set! group + (cons ref (process-group-rec-members group)))))) + + (define (process-group-leave! group ref) + (with-mutex (process-group-rec-mutex group) + (process-group-rec-members-set! group + (filter (lambda (r) (not (equal? r ref))) + (process-group-rec-members group))))) + + (define (process-group-members group) + (with-mutex (process-group-rec-mutex group) + (list-copy (process-group-rec-members group)))) + + (define (process-group-broadcast! group msg) + (for-each (lambda (ref) (dsend ref msg)) + (process-group-members group))) + + ;; ====================================================================== + ;; Distributed supervision + ;; ====================================================================== + + (define-record-type dist-sup-rec + (fields + (mutable children) ;; list of (id node-hint actor-ref) + (immutable mutex)) + (sealed #t)) + + (define (make-dist-supervisor) + (make-dist-sup-rec '() (make-mutex))) + + ;; (dist-supervisor-start-child! sup id proc [node-hint]) + ;; node-hint: a node name (string) or #f for local + (define (dist-supervisor-start-child! sup id proc . rest) + (let* ([node-hint (if (null? rest) #f (car rest))] + [actor + (if (or (not node-hint) + (equal? node-hint (*cluster-name*))) + ;; Start locally + (spawn-actor proc) + ;; Simulate remote placement: start locally but tag with node + (spawn-actor proc))]) + (with-mutex (dist-sup-rec-mutex sup) + (dist-sup-rec-children-set! sup + (cons (list id node-hint actor) + (filter (lambda (c) (not (equal? (car c) id))) + (dist-sup-rec-children sup))))) + actor)) + + (define (dist-supervisor-children sup) + (with-mutex (dist-sup-rec-mutex sup) + (map (lambda (c) + (list (list-ref c 0) (list-ref c 1) (list-ref c 2))) + (dist-sup-rec-children sup)))) + + ;; ====================================================================== + ;; Node failure detection + ;; ====================================================================== + + ;; Monitor table: node-name -> list of callbacks + (define *node-monitors* (make-hashtable equal-hash equal?)) + (define *node-monitors-mutex* (make-mutex)) + + ;; (monitor-node node-name callback) + ;; callback is called with node-name when node is detected as down + (define (monitor-node node-name callback) + (with-mutex *node-monitors-mutex* + (hashtable-update! *node-monitors* node-name + (lambda (cbs) (cons callback cbs)) '()))) + + ;; (demonitor-node node-name callback) + (define (demonitor-node node-name callback) + (with-mutex *node-monitors-mutex* + (hashtable-update! *node-monitors* node-name + (lambda (cbs) (filter (lambda (c) (not (eq? c callback))) cbs)) + '()))) + + ;; (node-alive? target-name) -> #t/#f + ;; Uses cluster-node-by-name (which searches alive nodes). + ;; Returns #t if the node exists in the cluster and is alive. + (define (node-alive? target-name) + ;; cluster-node-by-name searches (cluster-nodes) which already + ;; filters to only alive nodes. So if found, it's alive. + (if (cluster-node-by-name target-name) #t #f)) + + ;; (ping-node node-name timeout-ms) -> 'ok or 'timeout + ;; Uses cluster membership as a proxy for connectivity. + (define (ping-node node-name timeout-ms) + (if (node-alive? node-name) + 'ok + 'timeout)) + + ;; Internal: notify monitors of a failed node + (define (%notify-node-failure! node-name) + (let ([cbs (with-mutex *node-monitors-mutex* + (hashtable-ref *node-monitors* node-name '()))]) + (for-each (lambda (cb) (cb node-name)) cbs))) + + ;; ====================================================================== + ;; Serialization + ;; ====================================================================== + + ;; Serialize a message to a bytevector using write. + ;; Only works for data that is writable/readable (no procedures, etc.) + (define (serialize-message msg) + (let ([port (open-output-string)]) + (write msg port) + (string->utf8 (get-output-string port)))) + + ;; Deserialize a message from a bytevector. + (define (deserialize-message bv) + (let ([port (open-input-string (utf8->string bv))]) + (read port))) + + ;; Hook into cluster leave events so monitors fire automatically. + ;; Must be after all define forms (it's an expression, not a definition). + (on-node-leave + (lambda (node) + (%notify-node-failure! (node-name node)))) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/lib/std/dataframe.sls @@ -0,0 +1,653 @@ +#!chezscheme +;;; (std dataframe) — Tabular data with column-oriented storage +;;; +;;; A dataframe stores data as a vector of column vectors. +;;; All transformations return new dataframes (immutable API). +;;; +;;; Record layout: +;;; columns : vector of symbols (column names) +;;; data : vector of vectors (one per column, all same length) + +(library (std dataframe) + (export + ;; Creation + make-dataframe + dataframe? + dataframe-columns + dataframe-nrow + dataframe-ncol + ;; Access + dataframe-column + dataframe-row + dataframe-ref + dataframe-head + dataframe-tail + ;; Construction + dataframe-from-alists + dataframe-from-vectors + dataframe->alists + dataframe->vectors + ;; Transformation + dataframe-select + dataframe-drop + dataframe-filter + dataframe-map + dataframe-mutate + dataframe-rename + dataframe-sort + dataframe-join + dataframe-left-join + dataframe-append + ;; Aggregation + dataframe-group-by + dataframe-summarize + dataframe-count + ;; Stats + col-sum col-mean col-min col-max col-median col-std + ;; I/O + dataframe->csv-string + dataframe-from-csv-string + ;; Display + dataframe-display + dataframe-describe) + + (import (chezscheme)) + + ;; ====================================================================== + ;; Internal record + ;; ====================================================================== + + (define-record-type df-record + (fields + (immutable columns) ;; vector of symbols + (immutable data)) ;; vector of vectors (column-major) + (sealed #t)) + + (define (dataframe? x) (df-record? x)) + + ;; Build a df from parallel symbol-vector lists (already validated). + (define (%make-df cols data) + (make-df-record (list->vector cols) (list->vector data))) + + ;; ====================================================================== + ;; make-dataframe + ;; ====================================================================== + + ;; (make-dataframe columns data) + ;; columns : list of symbols + ;; data : list of lists — each inner list is one column's values + (define (make-dataframe columns data) + (unless (list? columns) + (error 'make-dataframe "columns must be a list of symbols" columns)) + (let* ([ncol (length columns)] + [vecs (map list->vector data)]) + (unless (= (length data) ncol) + (error 'make-dataframe "number of data lists must equal number of columns")) + (when (> ncol 0) + (let ([n (vector-length (car vecs))]) + (for-each (lambda (v) + (unless (= (vector-length v) n) + (error 'make-dataframe "all columns must have equal length"))) + vecs))) + (%make-df columns vecs))) + + ;; ====================================================================== + ;; Basic accessors + ;; ====================================================================== + + (define (dataframe-columns df) (vector->list (df-record-columns df))) + (define (dataframe-ncol df) (vector-length (df-record-columns df))) + (define (dataframe-nrow df) + (if (= (vector-length (df-record-data df)) 0) + 0 + (vector-length (vector-ref (df-record-data df) 0)))) + + ;; Find column index (or error). + (define (%col-index df col-name) + (let ([cols (df-record-columns df)]) + (let loop ([i 0]) + (cond + [(= i (vector-length cols)) + (error 'dataframe-column "column not found" col-name)] + [(eq? (vector-ref cols i) col-name) i] + [else (loop (+ i 1))])))) + + (define (dataframe-column df col-name) + (vector-copy (vector-ref (df-record-data df) (%col-index df col-name)))) + + (define (dataframe-row df i) + (let ([cols (df-record-columns df)] + [data (df-record-data df)]) + (let loop ([j 0] [acc '()]) + (if (= j (vector-length cols)) + (reverse acc) + (loop (+ j 1) + (cons (cons (vector-ref cols j) + (vector-ref (vector-ref data j) i)) + acc)))))) + + (define (dataframe-ref df row col) + (vector-ref (vector-ref (df-record-data df) (%col-index df col)) row)) + + ;; ====================================================================== + ;; Head / Tail + ;; ====================================================================== + + (define (dataframe-head df n) + (let* ([nrow (min n (dataframe-nrow df))] + [cols (dataframe-columns df)] + [data (df-record-data df)]) + (%make-df cols + (map (lambda (i) (vector-copy (vector-ref data i) 0 nrow)) + (iota (vector-length (df-record-columns df))))))) + + (define (dataframe-tail df n) + (let* ([nrow (dataframe-nrow df)] + [start (max 0 (- nrow n))] + [count (- nrow start)] + [cols (dataframe-columns df)] + [data (df-record-data df)]) + (%make-df cols + (map (lambda (i) (vector-copy (vector-ref data i) start count)) + (iota (vector-length (df-record-columns df))))))) + + ;; ====================================================================== + ;; Construction helpers + ;; ====================================================================== + + ;; (dataframe-from-alists (list alist ...)) — each alist is one row + (define (dataframe-from-alists alists) + (if (null? alists) + (%make-df '() '()) + (let* ([cols (map car (car alists))] + [nrow (length alists)] + [ncol (length cols)] + [vecs (make-vector ncol #f)]) + ;; Initialize column vectors + (let loop ([j 0]) + (when (< j ncol) + (vector-set! vecs j (make-vector nrow #f)) + (loop (+ j 1)))) + ;; Fill row by row + (let row-loop ([row alists] [i 0]) + (unless (null? row) + (for-each + (lambda (pair j) + (vector-set! (vector-ref vecs j) i (cdr pair))) + (car row) + (iota ncol)) + (row-loop (cdr row) (+ i 1)))) + (%make-df cols (vector->list vecs))))) + + ;; (dataframe-from-vectors col-names (list vec ...)) + (define (dataframe-from-vectors col-names vecs) + (%make-df col-names (map vector-copy vecs))) + + ;; (dataframe->alists df) — list of row alists + (define (dataframe->alists df) + (let ([nrow (dataframe-nrow df)]) + (map (lambda (i) (dataframe-row df i)) (iota nrow)))) + + ;; (dataframe->vectors df) — list of (col-name . vector) pairs + (define (dataframe->vectors df) + (map (lambda (col) (cons col (dataframe-column df col))) + (dataframe-columns df))) + + ;; ====================================================================== + ;; Selection and dropping + ;; ====================================================================== + + (define (dataframe-select df . col-names) + (let ([data (df-record-data df)]) + (%make-df col-names + (map (lambda (col) + (vector-copy (vector-ref data (%col-index df col)))) + col-names)))) + + (define (dataframe-drop df . col-names) + (let ([keep (filter (lambda (c) (not (memq c col-names))) + (dataframe-columns df))]) + (apply dataframe-select df keep))) + + ;; ====================================================================== + ;; Filter / Map / Mutate + ;; ====================================================================== + + ;; (dataframe-filter df pred) — pred takes a row alist + (define (dataframe-filter df pred) + (let* ([cols (dataframe-columns df)] + [data (df-record-data df)] + [ncol (length cols)] + [nrow (dataframe-nrow df)] + [new-rows '()]) + ;; Collect row indices satisfying pred + (let loop ([i 0] [acc '()]) + (if (= i nrow) + (let ([kept (reverse acc)]) + (let ([new-nrow (length kept)]) + (%make-df cols + (map (lambda (j) + (let ([src (vector-ref data j)] + [dst (make-vector new-nrow #f)]) + (let fill ([rows kept] [k 0]) + (unless (null? rows) + (vector-set! dst k (vector-ref src (car rows))) + (fill (cdr rows) (+ k 1)))) + dst)) + (iota ncol))))) + (let ([row-alist (dataframe-row df i)]) + (loop (+ i 1) + (if (pred row-alist) + (cons i acc) + acc))))))) + + ;; (dataframe-map df proc) — proc maps row alist -> row alist + (define (dataframe-map df proc) + (let ([nrow (dataframe-nrow df)]) + (dataframe-from-alists + (map (lambda (i) (proc (dataframe-row df i))) + (iota nrow))))) + + ;; (dataframe-mutate df col-name expr-proc) + ;; expr-proc takes a row alist and returns the new value for col-name + (define (dataframe-mutate df col-name expr-proc) + (let* ([cols (dataframe-columns df)] + [data (df-record-data df)] + [nrow (dataframe-nrow df)] + [ncol (length cols)] + [new-col (make-vector nrow #f)]) + ;; Fill new column + (let loop ([i 0]) + (when (< i nrow) + (vector-set! new-col i (expr-proc (dataframe-row df i))) + (loop (+ i 1)))) + ;; Check if col-name already exists + (let ([existing-idx + (let loop ([j 0]) + (cond + [(= j ncol) #f] + [(eq? (vector-ref (df-record-columns df) j) col-name) j] + [else (loop (+ j 1))]))]) + (if existing-idx + ;; Replace existing column + (let ([new-data (vector-copy data)]) + (vector-set! new-data existing-idx new-col) + (make-df-record (df-record-columns df) new-data)) + ;; Append new column + (%make-df (append cols (list col-name)) + (append (map (lambda (j) (vector-copy (vector-ref data j))) + (iota ncol)) + (list new-col))))))) + + ;; (dataframe-rename df old-name new-name) + (define (dataframe-rename df old-name new-name) + (let ([cols (df-record-columns df)]) + (make-df-record + (vector-map (lambda (c) (if (eq? c old-name) new-name c)) cols) + (df-record-data df)))) + + ;; ====================================================================== + ;; Sort + ;; ====================================================================== + + ;; (dataframe-sort df col [less?]) + (define (dataframe-sort df col . rest) + (let* ([less? (if (null? rest) < (car rest))] + [nrow (dataframe-nrow df)] + [col-vec (vector-ref (df-record-data df) (%col-index df col))] + [indices (list->vector (iota nrow))]) + ;; Sort indices by col values + (vector-sort! + (lambda (a b) (less? (vector-ref col-vec a) (vector-ref col-vec b))) + indices) + ;; Reorder all columns + (let* ([cols (dataframe-columns df)] + [data (df-record-data df)] + [ncol (length cols)]) + (%make-df cols + (map (lambda (j) + (let ([src (vector-ref data j)] + [dst (make-vector nrow #f)]) + (let loop ([k 0]) + (when (< k nrow) + (vector-set! dst k (vector-ref src (vector-ref indices k))) + (loop (+ k 1)))) + dst)) + (iota ncol)))))) + + ;; ====================================================================== + ;; Joins + ;; ====================================================================== + + ;; Inner join: keep rows where key exists in both df1 and df2. + (define (dataframe-join df1 df2 key) + (%join-impl df1 df2 key 'inner)) + + ;; Left join: keep all rows from df1, fill df2 cols with #f if no match. + (define (dataframe-left-join df1 df2 key) + (%join-impl df1 df2 key 'left)) + + (define (%join-impl df1 df2 key join-type) + (let* ([cols1 (dataframe-columns df1)] + [cols2 (filter (lambda (c) (not (eq? c key))) + (dataframe-columns df2))] + [all-cols (append cols1 cols2)] + [ncols2 (length cols2)] + [nrow1 (dataframe-nrow df1)] + [nrow2 (dataframe-nrow df2)] + ;; Build lookup table: key-val -> first row index in df2 + [lookup (make-hashtable equal-hash equal?)]) + (let loop ([i 0]) + (when (< i nrow2) + (let ([kv (dataframe-ref df2 i key)]) + (unless (hashtable-ref lookup kv #f) + (hashtable-set! lookup kv i))) + (loop (+ i 1)))) + ;; Build result rows + (let ([result-rows '()]) + (let loop ([i (- nrow1 1)]) + (when (>= i 0) + (let* ([kv (dataframe-ref df1 i key)] + [j (hashtable-ref lookup kv #f)]) + (when (or j (eq? join-type 'left)) + (let ([row1 (dataframe-row df1 i)] + [row2 (if j + (filter (lambda (p) (not (eq? (car p) key))) + (dataframe-row df2 j)) + (map (lambda (c) (cons c #f)) cols2))]) + (set! result-rows (cons (append row1 row2) result-rows))))) + (loop (- i 1)))) + (dataframe-from-alists result-rows)))) + + ;; ====================================================================== + ;; Append (concatenate rows) + ;; ====================================================================== + + (define (dataframe-append df1 df2) + (let* ([cols (dataframe-columns df1)] + [data1 (df-record-data df1)] + [data2 (df-record-data df2)] + [nrow1 (dataframe-nrow df1)] + [nrow2 (dataframe-nrow df2)] + [ncol (length cols)]) + (%make-df cols + (map (lambda (j) + (let* ([v1 (vector-ref data1 j)] + [v2 (vector-ref data2 j)]