(
(let $1 (Configure! world (DataSource '"config") '"DqEngine" '"force"))
(let $2 (DataSource '"dq" '"$all"))
(let $3 (Configure! $1 $2 '"Attr" '"maxtasksperstage" '2))
(let $4 (Configure! $3 $2 '"Attr" '"watermarksmode" '"default"))
(let $5 (Configure! $4 $2 '"Attr" '"computeactortype" '"async"))
(let $6 (Configure! $5 (DataSource '"pq" '"$all") '"Attr" '"consumer" '"test_client"))
(let $7 (DataSource '"pq" '"pq"))
(let $8 '('('"PartitionsCount" '"1")))
(let $9 (DataType 'Uint64))
(let $10 (OptionalType $9))
(let $11 '('"t" $10))
(let $12 (StructType '('"k" $10) $11 '('"v" $10)))
(let $13 (PqTopic '"pq" '"local" '"test_topic_input" $8 '() $12))
(let $14 '('"k" '"t" '"v"))
(let $15 '('"Endpoint" '"<pq_pq_endpoint>"))
(let $16 '('"SharedReading" '"1"))
(let $17 '('"UseSsl" '"1"))
(let $18 '('('"Consumer" '"test_client") $15 $16 '('"ReconnectPeriod" '"") '('"Format" '"json_each_row") '('"ReadGroup" '"fqrun") $17 '('"WatermarksEnable" '"1") '('"WatermarksGranularityUs" '"1000000") '('"WatermarksLateArrivalDelayUs" '"5000000")))
(let $19 (SecureParam '"cluster:default_pq"))
(let $20 (DqPqTopicSource $6 $13 $14 $18 $19 '"" $12 '""))
(let $21 (DqStage '((DqSource $7 $20)) (lambda '($25) (block '(
  (let $26 '('('"format" '"json_each_row") '('"formatSettings" '('('"data.datetime.formatname" '"POSIX") '('"data.timestamp.formatname" '"POSIX"))) '('"settings" '($16))))
  (let $27 (DqSourceWideWrap $25 $7 $12 $26))
  (return (FlatMap (NarrowMap $27 (lambda '($28 $29 $30) (AsStruct '('"k" $28) '('"t" $29) '('"v" $30)))) (lambda '($31) (Just (AsStruct '('group0 (Min (Member $31 '"k") (Just (Uint64 '2)))) '('"t" (Member $31 '"t")) '('"v" (Member $31 '"v")))))))
))) '('('"_logical_id" '0))))
(let $22 (DataSink '"pq" '"pq"))
(let $23 (PqTopic '"pq" '"local" '"test_topic_output" $8 '() (StructType '('"Data" (DataType 'String)))))
(let $24 (DqPqTopicSink $23 '($15 $17) $19))
(return (Commit! (DqQuery! $6 '((DqStage '((DqCnHashShuffle (TDqOutput $21 '0) '('group0))) (lambda '($32) (block '(
  (let $33 '('"strict"))
  (let $34 (lambda '($45) $45))
  (let $35 (MultiHoppingCore (FromFlow $32) (lambda '($36) (Member $36 'group0)) (lambda '($37) (FlatMap (Member (SafeCast $37 (StructType $11)) '"t") (lambda '($38) (block '(
    (let $39 '($9 '"" '"1"))
    (let $40 (CallableType '() '((OptionalType (DataType 'Timestamp))) $39))
    (let $41 (Udf '"DateTime2.FromMilliseconds" (Void) (VoidType) '"" $40 (VoidType) '"" '('('"blocks") $33)))
    (return (Apply $41 $38))
  ))))) (Interval '"5000") (Interval '"10000") (Interval '"5000000") 'true (lambda '($42) (AsStruct '('Sum0 (Member $42 '"v")))) (lambda '($43 $44) (AsStruct '('Sum0 (AggrAdd (Member $43 '"v") (Member $44 'Sum0))))) $34 $34 (lambda '($46 $47) (AsStruct '('Sum0 (AggrAdd (Member $46 'Sum0) (Member $47 'Sum0))))) (lambda '($48 $49 $50) (AsStruct '('Sum0 (Member $49 'Sum0)) '('group0 $48) '('"group1" $50))) '"1"))
  (return (FlatMap (ExtractMembers $35 '('Sum0)) (lambda '($51) (block '(
    (let $52 (ResourceType '"Yson2.Node"))
    (let $53 '($52 '"" '"1"))
    (let $54 (CallableType '() '((DataType 'Yson)) $53))
    (let $55 '($33))
    (let $56 (Udf '"Yson2.SerializeText" (Void) (VoidType) '"" $54 (VoidType) '"" $55))
    (let $57 (StructType '('"sum" $10)))
    (let $58 (TupleType (TupleType $57) (StructType) (TupleType)))
    (let $59 (CallableType '() '($52) '($57)))
    (let $60 (Udf '"Yson2.From" (Void) $58 '"" $59 (VoidType) '"" $55))
    (return (Just (AsStruct '('"column0" (Apply $56 (Apply $60 (AsStruct '('"sum" (Member $51 'Sum0)))))))))
  )))))
))) '('('"_logical_id" '0)) '((DqSink '0 $22 $24))))) $22))
)
