(
(let $1 (Configure! world (DataSource '"config") '"DqEngine" '"force"))
(let $2 (Configure! $1 (DataSource '"dq" '"$all") '"Attr" '"maxtasksperstage" '"2"))
(let $3 (Configure! $2 (DataSource '"pq" '"$all") '"Attr" '"consumer" '"test_client"))
(let $4 (DataSource '"pq" '"pq"))
(let $5 '('('"PartitionsCount" '"1")))
(let $6 (DataType 'String))
(let $7 (StructType '('"Data" $6)))
(let $8 (PqTopic '"pq" '"local" '"test_topic_input" $5 '() $7))
(let $9 '('"Endpoint" '"<pq_pq_endpoint>"))
(let $10 '('"SharedReading" '"1"))
(let $11 '('"UseSsl" '"1"))
(let $12 '('('"Consumer" '"test_client") $9 $10 '('"ReconnectPeriod" '"") '('"Format" '"raw") '('"ReadGroup" '"fqrun") $11))
(let $13 (SecureParam '"cluster:default_pq"))
(let $14 (DqPqTopicSource $3 $8 '('"Data") $12 $13 '"" $7 '""))
(let $15 (ResourceType '"Yson2.Node"))
(let $16 '('"strict"))
(let $17 '($16))
(let $18 (DataType 'Yson))
(let $19 '($15 '"" '"1"))
(let $20 (CallableType '() '($18) $19))
(let $21 (OptionalType $6))
(let $22 '($15))
(let $23 (DataType 'Uint64))
(let $24 (OptionalType $23))
(let $25 (OptionalType (DataType 'Int64)))
(let $26 (DqStage '((DqSource $4 $14)) (lambda '($30) (block '(
  (let $31 '('('"format" '"raw") '('"formatSettings" '('('"data.datetime.formatname" '"POSIX") '('"data.timestamp.formatname" '"POSIX"))) '('"settings" '($10))))
  (let $32 (DqSourceWideWrap $30 $4 $7 $31))
  (return (FlatMap (NarrowMap $32 (lambda '($33) (AsStruct '('"Data" $33)))) (lambda '($34) (block '(
    (let $35 (ResourceType '"Yson2.Options"))
    (let $36 (TupleType (TupleType $6 $35) (StructType) (TupleType)))
    (let $37 '('"1"))
    (let $38 '($6 '"" '"1"))
    (let $39 '((OptionalType $35)))
    (let $40 (CallableType $37 '((OptionalType $15)) $38 $39))
    (let $41 (Udf '"Yson2.Parse" (Void) $36 '"" $40 (VoidType) '"" '()))
    (let $42 (OptionalType (DataType 'Bool)))
    (let $43 (CallableType '('"2") '($35) '($42 '"AutoConvert") '($42 '"Strict")))
    (let $44 (Udf '"Yson2.Options" (Void) (VoidType) '"" $43 (VoidType) '"" $17))
    (let $45 (Just (NamedApply $44 '() (AsStruct '('"AutoConvert" (Just (Bool '"false"))) '('"Strict" (Just (Bool 'true)))))))
    (let $46 (Apply $41 (Member $34 '"Data") $45))
    (let $47 (FlatMap $46 (lambda '($53) (block '(
      (let $54 (Udf '"Yson2.Serialize" (Void) (VoidType) '"" $20 (VoidType) '"" $17))
      (return (Just (Apply $54 $53)))
    )))))
    (let $48 '($6))
    (let $49 (TupleType (TupleType $18) (StructType) (TupleType)))
    (let $50 '($18 '"" '"1"))
    (let $51 (CallableType $37 $22 $50 $39))
    (let $52 (Udf '"Yson2.Parse" (Void) $49 '"" $51 (VoidType) '"" $17))
    (return (Just (AsStruct '('"k" (FlatMap $47 (lambda '($55) (block '(
      (let $56 (CallableType $37 '($21) $19 $48 $39))
      (let $57 (Udf '"Yson2.LookupString" (Void) (VoidType) '"" $56 (VoidType) '"" '()))
      (return (Apply $57 (Apply $52 $55) (String '"key") $45))
    ))))) '('"t" (FlatMap $47 (lambda '($58) (block '(
      (let $59 (CallableType $37 '($24) $19 $48 $39))
      (let $60 (Udf '"Yson2.LookupUint64" (Void) (VoidType) '"" $59 (VoidType) '"" '()))
      (return (Apply $60 (Apply $52 $58) (String '"time") $45))
    ))))) '('"v" (FlatMap $47 (lambda '($61) (block '(
      (let $62 (CallableType $37 '($25) $19 $48 $39))
      (let $63 (Udf '"Yson2.LookupInt64" (Void) (VoidType) '"" $62 (VoidType) '"" '()))
      (return (Apply $63 (Apply $52 $61) (String '"val") $45))
    ))))))))
  )))))
))) '('('"_logical_id" '0))))
(let $27 (DataSink '"pq" '"pq"))
(let $28 (PqTopic '"pq" '"local" '"test_topic_output" $5 '() $7))
(let $29 (DqPqTopicSink $28 '($9 $11) $13))
(return (Commit! (DqQuery! $3 '((DqStage '((DqCnHashShuffle (TDqOutput $26 '0) '('"k"))) (lambda '($64) (block '(
  (let $65 (Interval '10000))
  (let $66 (lambda '($77) $77))
  (let $67 (MultiHoppingCore (FromFlow $64) (lambda '($68) (Member $68 '"k")) (lambda '($69) (FlatMap (SafeCast (Unwrap (Member (SafeCast $69 (StructType '('"t" $24))) '"t")) (OptionalType (DataType 'Uint32))) (lambda '($70) (block '(
    (let $71 '($23 '"" '"1"))
    (let $72 (CallableType '() '((OptionalType (DataType 'Timestamp))) $71))
    (let $73 (Udf '"DateTime2.FromMilliseconds" (Void) (VoidType) '"" $72 (VoidType) '"" '('('"blocks") $16)))
    (return (Apply $73 (Convert $70 $23)))
  ))))) (Interval '5000) $65 $65 'true (lambda '($74) (AsStruct '('Sum0 (Member $74 '"v")))) (lambda '($75 $76) (AsStruct '('Sum0 (AggrAdd (Member $75 '"v") (Member $76 'Sum0))))) $66 $66 (lambda '($78 $79) (AsStruct '('Sum0 (AggrAdd (Member $78 'Sum0) (Member $79 'Sum0))))) (lambda '($80 $81 $82) (AsStruct '('Sum0 (Member $81 'Sum0)) '('"_yql_time" $82) '('"k" $80))) '0 '"_yql_time"))
  (return (FlatMap (ExtractMembers $67 '('Sum0 '"k")) (lambda '($83) (block '(
    (let $84 (Udf '"Yson2.SerializeText" (Void) (VoidType) '"" $20 (VoidType) '"" $17))
    (let $85 (StructType '('"k" $21) '('"sum" $25)))
    (let $86 (TupleType (TupleType $85) (StructType) (TupleType)))
    (let $87 (CallableType '() $22 '($85)))
    (let $88 (Udf '"Yson2.From" (Void) $86 '"" $87 (VoidType) '"" $17))
    (return (Just (AsStruct '('"column0" (Apply $84 (Apply $88 (AsStruct '('"k" (Member $83 '"k")) '('"sum" (Member $83 'Sum0)))))))))
  )))))
))) '('('"_logical_id" '0)) '((DqSink '0 $27 $29))))) $27))
)
