index: add lazy ordered cursors + k-way merge (streaming foundation)
ober
7abd5f03d2a0e1f2c97c02b9121ebb7813cf6005
--- a/lib/jerboa-db/index/leveldb.ss +++ b/lib/jerboa-db/index/leveldb.ss @@ -167,7 +167,12 @@ (cons (fasl-bytevector->datom val) acc)) '()))) - (make-dbi name add! remove! range-query seek count-range snapshot all-datoms)) + ;; Cursor fallback: materialise the range, then stream it. Correct but not + ;; lazy; a future version can wrap LevelDB's native iterator for true + ;; early-exit. + (def (cursor start end) (list->stream (range-query start end))) + + (make-dbi name add! remove! range-query seek count-range snapshot all-datoms cursor)) ;; ---- Create the four covering indices ---- --- a/lib/jerboa-db/index/memory.ss +++ b/lib/jerboa-db/index/memory.ss @@ -237,35 +237,39 @@ (def (bt->list tree) (reverse (bt-fold (lambda (k acc) (cons k acc)) '() tree))) - ;; Fold over keys k with lo <= k <= hi (inclusive), ascending. - (def (bt-range-fold proc init tree lo hi cmp) - (cond - [(not tree) init] - [(leaf? tree) - (let* ([keys (leaf-keys tree)] - [n (vector-length keys)] - [start (vec-bsearch keys lo cmp #f)]) - (let loop ([i start] [acc init]) - (if (or (= i n) (> (cmp (vector-ref keys i) hi) 0)) - acc - (loop (+ i 1) (proc (vector-ref keys i) acc)))))] - [else - (let* ([seps (branch-seps tree)] - [kids (branch-kids tree)] - [nk (vector-length kids)]) - (let loop ([i 0] [acc init]) - (if (= i nk) - acc - (let ([left-sep (if (= i 0) #f (vector-ref seps (- i 1)))] - [right-sep (if (< i (- nk 1)) (vector-ref seps i) #f)]) - (cond - ;; child's min > hi -> all remaining children are larger too - [(and left-sep (> (cmp left-sep hi) 0)) acc] - ;; child's upper bound <= lo -> entirely below the range - [(and right-sep (<= (cmp right-sep lo) 0)) (loop (+ i 1) acc)] - [else (loop (+ i 1) - (bt-range-fold proc acc (vector-ref kids i) - lo hi cmp))])))))])) + ;; Lazy in-order stream of keys k with lo <= k <= hi (inclusive), ascending. + ;; Returns '() or (cons datom (delay rest)); only the forced prefix is walked, + ;; so a consumer that stops early (first-match, top-k) pays O(log n + taken). + ;; k is the continuation: the stream of in-range keys that follow `node`. + (def (bt-range-stream tree lo hi cmp) + (define (node-stream node k) + (cond + [(leaf? node) + (let* ([keys (leaf-keys node)] + [n (vector-length keys)] + [start (vec-bsearch keys lo cmp #f)]) + (let loop ([i start]) + (cond + [(>= i n) (k)] ;; leaf done -> siblings + [(> (cmp (vector-ref keys i) hi) 0) '()] ;; past hi -> stream ends + [else (cons (vector-ref keys i) (delay (loop (+ i 1))))])))] + [else + (let* ([seps (branch-seps node)] + [kids (branch-kids node)] + [nk (vector-length kids)]) + (let build ([i 0]) + (if (= i nk) + (k) + (let ([left-sep (if (= i 0) #f (vector-ref seps (- i 1)))] + [right-sep (if (< i (- nk 1)) (vector-ref seps i) #f)]) + (cond + ;; child's min > hi -> this child and all that follow are > hi + [(and left-sep (> (cmp left-sep hi) 0)) '()] + ;; child's upper bound <= lo -> entirely below the range + [(and right-sep (<= (cmp right-sep lo) 0)) (build (+ i 1))] + [else (node-stream (vector-ref kids i) + (lambda () (build (+ i 1))))])))))])) + (if (not tree) '() (node-stream tree (lambda () '())))) ;; Order-statistic rank: count of keys < key (le?=#f) or <= key (le?=#t). (def (bt-rank tree key cmp le?) @@ -330,47 +334,16 @@ (flush!) (set-tree! (bt-delete (get-tree) datom comparator))) - (define (in-range? d lo hi) - (and (>= (comparator d lo) 0) - (<= (comparator d hi) 0))) - - (define (buffer-in-range lo hi) - ;; Filter buffer for datoms in [lo, hi]. - (let loop ([xs (get-buf)] [acc '()]) - (cond - [(null? xs) acc] - [(in-range? (car xs) lo hi) - (loop (cdr xs) (cons (car xs) acc))] - [else (loop (cdr xs) acc)]))) - - ;; Merge two sorted-by-comparator lists, dropping duplicates. - (define (merge-sorted a b) - (let loop ([a a] [b b] [acc '()]) - (cond - [(null? a) (append (reverse acc) b)] - [(null? b) (append (reverse acc) a)] - [else - (let ([c (comparator (car a) (car b))]) - (cond - [(< c 0) (loop (cdr a) b (cons (car a) acc))] - [(> c 0) (loop a (cdr b) (cons (car b) acc))] - ;; equal — drop one - [else (loop (cdr a) (cdr b) (cons (car a) acc))]))]))) + ;; Lazy ordered cursor: flush pending writes into the B+-tree, then stream + ;; the range on demand. The flush keeps the buffer out of the read path + ;; (matching snapshot/count), so the cursor is a pure ordered tree walk. + (define (cursor start end) + (flush!) + (bt-range-stream (get-tree) start end comparator)) + ;; Eager range as a list, implemented on top of the cursor (dogfoods it). (define (range-query start end) - (let* ([base-list (reverse - (bt-range-fold - (lambda (k acc) (cons k acc)) - '() (get-tree) start end comparator))] - [buf-list (buffer-in-range start end)]) - (cond - [(null? buf-list) base-list] - [(null? base-list) - (list-sort (lambda (a b) (< (comparator a b) 0)) buf-list)] - [else - (merge-sorted - base-list - (list-sort (lambda (a b) (< (comparator a b) 0)) buf-list))]))) + (stream->list (cursor start end))) (define (seek . components) (let-values ([(e a v tx) (apply values components)]) @@ -404,7 +377,7 @@ (flush!) (bt->list (get-tree))) - (make-dbi name add! remove! range-query seek count-range snapshot all-datoms))) + (make-dbi name add! remove! range-query seek count-range snapshot all-datoms cursor))) ;; ---- Create the four covering indices ---- --- a/lib/jerboa-db/index/protocol.ss +++ b/lib/jerboa-db/index/protocol.ss @@ -9,7 +9,11 @@ (export make-dbi dbi? dbi-add! dbi-remove! dbi-range dbi-seek dbi-count dbi-snapshot - dbi-datoms dbi-name + dbi-datoms dbi-name dbi-cursor + + ;; Lazy ordered cursors (streams) for early-exit / streaming joins + stream-null? stream-car stream-cdr stream->list stream-take + list->stream merge-streams ;; Index set (all four indices) make-index-set index-set? @@ -38,7 +42,8 @@ seek-fn ;; (components) -> list of datoms count-fn ;; (start-datom end-datom) -> integer snapshot-fn ;; () -> opaque snapshot - datoms-fn)) ;; () -> list of all datoms + datoms-fn ;; () -> list of all datoms + cursor-fn)) ;; (start-datom end-datom) -> lazy stream of datoms ;; Dispatch wrappers @@ -63,6 +68,51 @@ (def (dbi-datoms idx) ((dbi-datoms-fn idx))) + ;; Lazy ordered cursor over [start, end]: returns a stream, evaluated on + ;; demand. Lets callers stop early (existence, first-match, top-k, or a + ;; merge-join that exhausts one side) without materialising the whole range. + (def (dbi-cursor idx start end) + ((dbi-cursor-fn idx) start end)) + + ;; ---- Streams ---- + ;; A stream is '() or (cons datom promise); (stream-cdr s) forces the tail. + + (def (stream-null? s) (null? s)) + (def (stream-car s) (car s)) + (def (stream-cdr s) (force (cdr s))) + + (def (stream->list s) + (let loop ([s s] [acc '()]) + (if (null? s) (reverse acc) (loop (force (cdr s)) (cons (car s) acc))))) + + ;; First n elements of a stream as a list (early-exit: only forces n tails). + (def (stream-take s n) + (let loop ([s s] [n n] [acc '()]) + (if (or (= n 0) (null? s)) + (reverse acc) + (loop (force (cdr s)) (- n 1) (cons (car s) acc))))) + + (def (list->stream lst) + (if (null? lst) + '() + (cons (car lst) (delay (list->stream (cdr lst)))))) + + ;; K-way merge of ascending streams, dropping duplicates (datoms equal by the + ;; comparator). cmp is three-way (-1/0/1). Lazy: one step per element consumed. + (def (merge-streams cmp streams) + (let ([live (filter (lambda (s) (not (null? s))) streams)]) + (if (null? live) + '() + (let* ([heads (map car live)] + [m (fold-left (lambda (a b) (if (<= (cmp a b) 0) a b)) + (car heads) (cdr heads))]) + (cons m + (delay + (merge-streams cmp + (map (lambda (s) + (if (= (cmp (car s) m) 0) (force (cdr s)) s)) + live)))))))) + ;; ---- Index set: the four covering indices ---- (define-record-type index-set --- a/tests/test-core.ss +++ b/tests/test-core.ss @@ -5,7 +5,9 @@ (jerboa-db history) (jerboa-db entity) (jerboa-db spec) - (jerboa-db value-store)) + (jerboa-db value-store) + (jerboa-db index protocol) + (jerboa-db index memory)) ;; ---- Test harness ---- @@ -673,6 +675,33 @@ (assert-equal (value-store-get vs hash1) "hello world") (assert-equal (value-store-get vs hash3) "different value"))) +(test "index cursor streams in order with early-exit" + (let* ([is (make-mem-index-set)] + [eavt (index-set-eavt is)] + [lo (make-datom 0 0 +min-val+ 0 #t)] + [hi (make-datom 100 100 +max-val+ 100 #t)]) + (for-each (lambda (e) (dbi-add! eavt (make-datom e 1 e 1 #t))) '(5 3 1 4 2)) + ;; early-exit: stream-take 1 yields the smallest element only + (assert-equal (datom-e (car (stream-take (dbi-cursor eavt lo hi) 1))) 1) + ;; full stream is ascending and matches the eager range + (assert-equal (map datom-e (stream->list (dbi-cursor eavt lo hi))) '(1 2 3 4 5)) + (assert-equal (map datom-e (dbi-range eavt lo hi)) '(1 2 3 4 5)))) + +(test "merge-streams unions ascending and dedups" + (let* ([is1 (make-mem-index-set)] [is2 (make-mem-index-set)] + [a (index-set-eavt is1)] [b (index-set-eavt is2)] + [lo (make-datom 0 0 +min-val+ 0 #t)] + [hi (make-datom 100 100 +max-val+ 100 #t)]) + (for-each (lambda (e) (dbi-add! a (make-datom e 1 e 1 #t))) '(1 3 5)) + (for-each (lambda (e) (dbi-add! b (make-datom e 1 e 1 #t))) '(3 4 5 6)) + ;; shared datoms (e=3, e=5) collapse to one + (assert-equal + (map datom-e + (stream->list + (merge-streams compare-datoms-eavt + (list (dbi-cursor a lo hi) (dbi-cursor b lo hi))))) + '(1 3 4 5 6)))) + ;; ============================================================ ;; Report ;; ============================================================