ydb/yql/essentials/tests/s-expressions/suites/InMem/Fold1MapOverPreservedStream.yqls

(
(let config (DataSource 'config))

(let res_sink (DataSink 'result))
(let list (AsList
    '((Uint32 '1) (String 'r))
    '((Uint32 '1) (String 'a))
    '((Uint32 '2) (String 'b))
    '((Uint32 '3) (String 'b))
    '((Uint32 '5) (String 'i))
    '((Uint32 '8) (String 't))
))
(let base_stream (Iterator list))

(let queue (QueueCreate (ListItemType (TypeOf list)) (Uint64 '5) (Uint64 '0)))
#(let queue (QueueCreate (TupleType (DataType 'Uint32) (DataType 'String)) (Uint64 '5) (Uint64 '0)))
(let stream (PreserveStream base_stream queue (Uint64 '2)))

(let init (lambda '(item) (block '(
    (let key (Nth item '0))
    (let val (Nth item '1))
    (let next (QueuePeek queue (Uint64 '1) (DependsOn item)))
    (let over (QueuePeek queue (Uint64 '2) (DependsOn item)))
    (let nkey (Nth next '0))
    (let nval (Nth next '1))
    (let okey (Nth over '0))
    (let oval (Nth over '1))
    (let skey key)
    (let sval val)
    (let validate (== (+ key (Coalesce nkey (Uint32 '0))) (Coalesce okey (Uint32 '0))))
    (let words (Concat (Concat val (Coalesce nval (String '""))) (Coalesce oval (String '""))))
    (let new_item '(key val skey sval validate words))
    (let new_state '(skey sval))
    (return '(new_item new_state)
)))))

(let update (lambda '(item state) (block '(
    (let key (Nth item '0))
    (let val (Nth item '1))
    (let next (QueuePeek queue (Uint64 '1) (DependsOn item)))
    (let over (QueuePeek queue (Uint64 '2) (DependsOn item)))
    (let nkey (Nth next '0))
    (let nval (Nth next '1))
    (let okey (Nth over '0))
    (let oval (Nth over '1))
    (let skey (Nth state '0))
    (let sval (Nth state '1))
    (let skey (+ skey key))
    (let sval (Concat sval val))
    (let validate (== (+ key (Coalesce nkey (Uint32 '0))) (Coalesce okey (Uint32 '0))))
    (let words (Concat (Concat val (Coalesce nval (String '""))) (Coalesce oval (String '""))))
    (let new_item '(key val skey sval validate words))
    (let new_state '(skey sval))
    (return '(new_item new_state)
)))))

(let result (Fold1Map stream init update))

(let world (Write! world res_sink (Key) (Collect result) '('('type))))
(let world (Commit! world res_sink))
(return world)
)