(
(let $1 (Configure! world (DataSource '"config") '"DqEngine" '"force"))
(let $2 (DataSource '"dq" '"$all"))
(let $3 (Configure! $1 $2 '"Attr" '"watermarksmode" '"default"))
(let $4 (Configure! $3 $2 '"Attr" '"watermarksgranularityms" '"4"))
(let $5 (Configure! $4 $2 '"Attr" '"watermarksidletimeoutms" '"5"))
(let $6 (Configure! $5 $2 '"Attr" '"watermarkslatearrivaldelayms" '"6"))
(let $7 (Configure! $6 $2 '"Attr" '"watermarksenableidlepartitions" '"true"))
(let $8 (Configure! $7 (DataSource '"pq" '"$all") '"Attr" '"consumer" '"test_client"))
(let $9 (DataSink 'result))
(let $10 (DataSource '"pq" '"pq"))
(let $11 (StructType '('"ts" (DataType 'Timestamp))))
(let $12 (PqTopic '"pq" '"local" '"test_topic_input" '('('"PartitionsCount" '"1")) '() $11))
(let $13 '('"SharedReading" '"1"))
(let $14 '('('"Consumer" '"test_client") '('"Endpoint" '"<pq_pq_endpoint>") $13 '('"ReconnectPeriod" '"") '('"Format" '"json_each_row") '('"ReadGroup" '"fqrun") '('"UseSsl" '"1") '('"WatermarksEnable" '"1") '('"WatermarksGranularityUs" '"2000000") '('"WatermarksLateArrivalDelayUs" '"6000") '('"WatermarksLateEventsPolicy" '"adjust") '('"WatermarksIdlePartitions" '"1") '('"WatermarksIdleTimeoutUs" '"3000000")))
(let $15 (DqPqTopicSource world $12 '('"ts") $14 (SecureParam '"cluster:default_pq") '"" $11 '"\x1A\x1B\b\x03\x12\x04\x12\x02ts\x1A\x11\n\x0F\n\x02\b3\x12\t!@KL\x00\x00\x00\x00\x00"))
(let $16 (DqStage '((DqSource $10 $15)) (lambda '($19) (block '(
  (let $20 '('('"data.datetime.formatname" '"POSIX") '('"data.timestamp.formatname" '"POSIX") '('"watermarkgranularity" '"PT2S") '('"watermarkidletimeout" '"PT3S")))
  (let $21 '('('"format" '"json_each_row") '('"formatSettings" $20) '('"settings" '($13))))
  (let $22 (DqSourceWideWrap $19 $10 $11 $21))
  (return (NarrowMap $22 (lambda '($23) (AsStruct '('"ts" $23)))))
))) '('('"_logical_id" '0))))
(let $17 (DqStage '((DqCnUnionAll (TDqOutput $16 '"0"))) (lambda '($24) $24) '('('"_logical_id" '0))))
(let $18 (ResPull! $8 $9 (Key) (DqCnResult (TDqOutput $17 '"0") '()) '('('type) '('autoref)) '"dq"))
(return (Commit! (Commit! $18 $9) (DataSink '"pq" '"pq")))
)
