{"action":"opened","issue":{"url":"https://api.github.com/repos/pipelinedb/pipelinedb/issues/1317","labels_url":"https://api.github.com/repos/pipelinedb/pipelinedb/issues/1317/labels{/name}","comments_url":"https://api.github.com/repos/pipelinedb/pipelinedb/issues/1317/comments","events_url":"https://api.github.com/repos/pipelinedb/pipelinedb/issues/1317/events","html_url":"https://github.com/pipelinedb/pipelinedb/issues/1317","id":123000453,"number":1317,"title":"Continuous transforms","user":{"login":"derekjn","id":384666,"avatar_url":"https://avatars.githubusercontent.com/u/384666?v=3","gravatar_id":"","url":"https://api.github.com/users/derekjn","html_url":"https://github.com/derekjn","followers_url":"https://api.github.com/users/derekjn/followers","following_url":"https://api.github.com/users/derekjn/following{/other_user}","gists_url":"https://api.github.com/users/derekjn/gists{/gist_id}","starred_url":"https://api.github.com/users/derekjn/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/derekjn/subscriptions","organizations_url":"https://api.github.com/users/derekjn/orgs","repos_url":"https://api.github.com/users/derekjn/repos","events_url":"https://api.github.com/users/derekjn/events{/privacy}","received_events_url":"https://api.github.com/users/derekjn/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/pipelinedb/pipelinedb/labels/transforms","name":"transforms","color":"207de5"}],"state":"open","locked":false,"assignee":{"login":"derekjn","id":384666,"avatar_url":"https://avatars.githubusercontent.com/u/384666?v=3","gravatar_id":"","url":"https://api.github.com/users/derekjn","html_url":"https://github.com/derekjn","followers_url":"https://api.github.com/users/derekjn/followers","following_url":"https://api.github.com/users/derekjn/following{/other_user}","gists_url":"https://api.github.com/users/derekjn/gists{/gist_id}","starred_url":"https://api.github.com/users/derekjn/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/derekjn/subscriptions","organizations_url":"https://api.github.com/users/derekjn/orgs","repos_url":"https://api.github.com/users/derekjn/repos","events_url":"https://api.github.com/users/derekjn/events{/privacy}","received_events_url":"https://api.github.com/users/derekjn/received_events","type":"User","site_admin":false},"milestone":null,"comments":0,"created_at":"2015-12-18T18:15:10Z","updated_at":"2015-12-18T18:15:10Z","closed_at":null,"body":"Users want to be able to continuously transform incoming data without storing it, and then write the transformation result out to some external datastore. This can be accomplished by storing a non-aggregate query in a catalog table `pipeline_transform`, along with a function that receives the transformation's output row as an argument.\r\n\r\n```\r\nCREATE CONTINUOUS TRANSFORM t AS SELECT x, func(y), foo(z) FROM stream\r\n    THEN EXECUTE PROCEDURE external_write(output_row); \r\n```\r\n\r\nIf users want to do batching or something similar, we could use some kind of hearbeat parameter that would call the function periodically regardless of whether or not any new rows have arrived:\r\n\r\n```\r\nCREATE CONTINUOUS TRANSFORM batched_t WITH (hearbeat='5 seconds') AS\r\n    SELECT x, func(y), foo(z) FROM stream THEN EXECUTE PROCEDURE external_write(output_row); \r\n```"}}