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