(
(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 '($55) (block '(
      (let $56 (Udf '"Yson2.Serialize" (Void) (VoidType) '"" $20 (VoidType) '"" $17))
      (return (Just (Apply $56 $55)))
    )))))
    (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))
    (let $54 (FlatMap $46 (lambda '($57) (Apply $49 (Apply $53 $57) (String '"key") $44))))
    (return (Just (AsStruct '('group0 (StablePickle (AsList (Just (Int64 '1)) $54))) '('"k" $54) '('"t" (FlatMap $46 (lambda '($58) (block '(
      (let $59 (CallableType $36 '($24) $19 $47 $38))
      (let $60 (Udf '"Yson2.LookupUint64" (Void) (VoidType) '"" $59 (VoidType) '"" '()))
      (return (Apply $60 (Apply $53 $58) (String '"time") $44))
    ))))) '('"v" (FlatMap $46 (lambda '($61) (Apply $49 (Apply $53 $61) (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) '('group0 '"k"))) (lambda '($62) (block '(
  (let $63 (Interval '10000))
  (let $64 (lambda '($75) $75))
  (let $65 (MultiHoppingCore (FromFlow $62) (lambda '($66) '((Member $66 'group0) (Member $66 '"k"))) (lambda '($67) (FlatMap (SafeCast (Unwrap (Member (SafeCast $67 (StructType '('"t" $24))) '"t")) (OptionalType (DataType 'Uint32))) (lambda '($68) (block '(
    (let $69 '($23 '"" '1))
    (let $70 (CallableType '() '((OptionalType (DataType 'Timestamp))) $69))
    (let $71 (Udf '"DateTime2.FromMilliseconds" (Void) (VoidType) '"" $70 (VoidType) '"" '('('"blocks") $16)))
    (return (Apply $71 (Convert $68 $23)))
  ))))) (Interval '5000) $63 $63 'true (lambda '($72) (AsStruct '('Sum0 (Member $72 '"v")))) (lambda '($73 $74) (AsStruct '('Sum0 (AggrAdd (Member $73 '"v") (Member $74 'Sum0))))) $64 $64 (lambda '($76 $77) (AsStruct '('Sum0 (AggrAdd (Member $76 'Sum0) (Member $77 'Sum0))))) (lambda '($78 $79 $80) (AsStruct '('Sum0 (Member $79 'Sum0)) '('"_yql_time" $80) '('group0 (Nth $78 '0)) '('"k" (Nth $78 '1)))) '0 '"_yql_time"))
  (return (FlatMap (ExtractMembers $65 '('Sum0)) (lambda '($81) (block '(
    (let $82 (Udf '"Yson2.SerializeText" (Void) (VoidType) '"" $20 (VoidType) '"" $17))
    (let $83 (StructType '('"sum" $21)))
    (let $84 (TupleType (TupleType $83) (StructType) (TupleType)))
    (let $85 (CallableType '() $22 '($83)))
    (let $86 (Udf '"Yson2.From" (Void) $84 '"" $85 (VoidType) '"" $17))
    (return (Just (AsStruct '('"column0" (Apply $82 (Apply $86 (AsStruct '('"sum" (Member $81 'Sum0)))))))))
  )))))
))) '('('"_logical_id" '0)) '((DqSink '0 $26 $28))))) $26))
)
