ydb/yql/essentials/tests/s-expressions/suites/Aggregation/AggregateOptState.yql

(
(let world (block '(
  (let x (Read! world (DataSource '"yt" '"plato") (Key '('table (String '"Input"))) '('"value" '"key" '"subkey") '()))
  (let world (Left! x))
  (let table2 (Right! x))
  (let output (block '(
    (let select (block '(
      (let core table2)
      (let core (block '(
        (let Sum1_create (lambda '(row) (FromString (Member row '"subkey") '"Int64")))
        (let Sum1_update (lambda '(row state) (OptionalReduce state (FromString (Member row '"subkey") '"Int64") (lambda '(a b) (+ a b)))))
        (let Sum1_save (lambda '(state) state))
        (let Sum1_load (lambda '(item) item))
        (let Sum1_merge (lambda '(a b) (OptionalReduce a b (lambda '(a b) (+ a b)))))
        (let Sum1_finish (lambda '(state) state))
        (return (Aggregate core '('"value") '('('"Sum1" (AggregationTraits (ListItemType (TypeOf core)) Sum1_create Sum1_update Sum1_save Sum1_load Sum1_merge Sum1_finish (Null))))))
)
))
      (return core)
)
))
    (return select)
)
))
  (let world (block '(
    (let result_sink (DataSink 'result))
    (let sortedOutput (Sort output (Bool 'false) (lambda '(x) (Member x 'value) )))
    (let world (Write! world result_sink (Key) sortedOutput '('('type) '('autoref))))
    (let plato_sink (DataSink '"yt" '"plato"))
    (let world (Commit! world plato_sink))
    (return (Commit! world result_sink))
)
))
  (return world)
)
))
(return world)
)