(
(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 'String))
(let $10 '('"k" (OptionalType $9)))
(let $11 (DataType 'Uint64))
(let $12 (OptionalType $11))
(let $13 '('"t" $12))
(let $14 (StructType $10 $13 '('"v" $12)))
(let $15 (PqTopic '"pq" '"local" '"test_topic_input" $8 '() $14))
(let $16 '('"k" '"t" '"v"))
(let $17 '('"Endpoint" '"<pq_pq_endpoint>"))
(let $18 '('"SharedReading" '"1"))
(let $19 '('"UseSsl" '"1"))
(let $20 '('('"Consumer" '"test_client") $17 $18 '('"ReconnectPeriod" '"") '('"Format" '"json_each_row") '('"ReadGroup" '"fqrun") $19 '('"WatermarksEnable" '"1") '('"WatermarksGranularityUs" '"1000000") '('"WatermarksLateArrivalDelayUs" '"5000000") '('"WatermarksLateEventsPolicy" '"adjust")))
(let $21 (SecureParam '"cluster:default_pq"))
(let $22 (DqPqTopicSource $6 $15 $16 $20 $21 '"" $14 '""))
(let $23 (DqStage '((DqSource $7 $22)) (lambda '($27) (block '(
  (let $28 '('('"format" '"json_each_row") '('"formatSettings" '('('"data.datetime.formatname" '"POSIX") '('"data.timestamp.formatname" '"POSIX"))) '('"settings" '($18))))
  (let $29 (DqSourceWideWrap $27 $7 $14 $28))
  (return (NarrowMap $29 (lambda '($30 $31 $32) (AsStruct '('"k" $30) '('"t" $31) '('"v" $32)))))
))) '('('"_logical_id" '0))))
(let $24 (DataSink '"pq" '"pq"))
(let $25 (PqTopic '"pq" '"local" '"test_topic_output" $8 '() (StructType '('"Data" $9))))
(let $26 (DqPqTopicSink $25 '($17 $19) $21))
(return (Commit! (DqQuery! $6 '((DqStage '((DqCnHashShuffle (TDqOutput $23 '0) '('"k"))) (lambda '($33) (block '(
  (let $34 '('"strict"))
  (let $35 (lambda '($46) $46))
  (let $36 (MultiHoppingCore (FromFlow $33) (lambda '($37) (Member $37 '"k")) (lambda '($38) (FlatMap (Member (SafeCast $38 (StructType $13)) '"t") (lambda '($39) (block '(
    (let $40 '($11 '"" '"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" $51) '('"k" $49))) '"1"))
  (return (FlatMap (ExtractMembers $36 '('Sum0 '"k")) (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 $10 '('"sum" $12)))
    (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 '('"k" (Member $52 '"k")) '('"sum" (Member $52 'Sum0)))))))))
  )))))
))) '('('"_logical_id" '0)) '((DqSink '0 $24 $26))))) $24))
)
