(
(let $1 (Configure! world (DataSource '"config") '"DqEngine" '"force"))
(let $2 (Configure! $1 (DataSource '"dq" '"$all") '"Attr" '"watermarksmode" '"default"))
(let $3 (Configure! $2 (DataSource '"pq" '"$all") '"Attr" '"consumer" '"test_client"))
(let $4 (DataSink 'result))
(let $5 '('('"system" '"_yql_sys_seq_no") '('"system" '"_yql_sys_partition_id") '('"system" '"_yql_sys_create_time") '('"system" '"_yql_sys_message_group_id") '('"system" '"_yql_sys_tsp_write_time") '('"system" '"_yql_sys_offset")))
(let $6 (PqTopic '"pq" '"local" '"test_topic_input" '('('"PartitionsCount" '"1")) $5 (StructType '('"ts" (DataType 'Timestamp)))))
(let $7 (Right! (PqReadTopic! $3 (DataSource '"pq" '"pq") $6 '('"ts") '"json_each_row" '"" (Void) '('('"data.datetime.formatname" '"POSIX") '('"data.timestamp.formatname" '"POSIX")) (lambda '($9) (block '(
  (let $10 '((DataType 'Int32) '"" '"1"))
  (let $11 (CallableType '() '((OptionalType (DataType 'Interval))) $10))
  (let $12 (Udf '"DateTime2.IntervalFromSeconds" (Void) (VoidType) '"" $11 (VoidType) '"" '('('"blocks") '('"strict"))))
  (return (Unwrap (- (Member $9 '"ts") (Apply $12 (Int32 '"5")))))
))))))
(let $8 (ResWrite! $3 $4 (Key) (RemovePrefixMembers (RemoveSystemMembers (PersistableRepr (SqlProject $7 '((SqlProjectStarItem (TypeOf $7) '"" (lambda '($13) $13) '()))))) '('_yql_sys_)) '('('type) '('autoref))))
(return (Commit! (Commit! $8 $4) (DataSink '"pq" '"pq")))
)
