(
(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) (block '(
    (let $32 (Member $31 '"k"))
    (return (Just (AsStruct '('group0 (StablePickle (AsList (Just (Uint64 '1)) $32))) '('"k" $32) '('"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 '"k"))) (lambda '($33) (block '(
  (let $34 '('"strict"))
  (let $35 (lambda '($46) $46))
  (let $36 (MultiHoppingCore (FromFlow $33) (lambda '($37) '((Member $37 'group0) (Member $37 '"k"))) (lambda '($38) (FlatMap (Member (SafeCast $38 (StructType $11)) '"t") (lambda '($39) (block '(
    (let $40 '($9 '"" '1))
    (let $41 (CallableType '() '((OptionalType (DataType 'Timestamp))) $40))
    (let $42 (Udf '"DateTime2.FromMilliseconds" (Void) (VoidType) '"" $41 (VoidType) '"" '('('"blocks") $34)))
    (return (Apply $42 $39))
  ))))) (Interval '"5000") (Interval '"10000") (Interval '"5000000") 'true (lambda '($43) (AsStruct '('Sum0 (Member $43 '"v")))) (lambda '($44 $45) (AsStruct '('Sum0 (AggrAdd (Member $44 '"v") (Member $45 'Sum0))))) $35 $35 (lambda '($47 $48) (AsStruct '('Sum0 (AggrAdd (Member $47 'Sum0) (Member $48 'Sum0))))) (lambda '($49 $50 $51) (AsStruct '('Sum0 (Member $50 'Sum0)) '('group0 (Nth $49 '0)) '('"group1" $51) '('"k" (Nth $49 '1)))) '1))
  (return (FlatMap (ExtractMembers $36 '('Sum0)) (lambda '($52) (block '(
    (let $53 (ResourceType '"Yson2.Node"))
    (let $54 '($53 '"" '1))
    (let $55 (CallableType '() '((DataType 'Yson)) $54))
    (let $56 '($34))
    (let $57 (Udf '"Yson2.SerializeText" (Void) (VoidType) '"" $55 (VoidType) '"" $56))
    (let $58 (StructType '('"sum" $10)))
    (let $59 (TupleType (TupleType $58) (StructType) (TupleType)))
    (let $60 (CallableType '() '($53) '($58)))
    (let $61 (Udf '"Yson2.From" (Void) $59 '"" $60 (VoidType) '"" $56))
    (return (Just (AsStruct '('"column0" (Apply $57 (Apply $61 (AsStruct '('"sum" (Member $52 'Sum0)))))))))
  )))))
))) '('('"_logical_id" '0)) '((DqSink '0 $22 $24))))) $22))
)
