From ffecc83e2858dc1a49549e248488cde18116d79c Mon Sep 17 00:00:00 2001 From: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> Date: Mon, 27 May 2024 15:38:09 +0545 Subject: [PATCH 1/3] py: apply backpressure by default for http egress Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> --- python/feldera/rest/client.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/python/feldera/rest/client.py b/python/feldera/rest/client.py index 50b7d7bc481..6bab5f28a79 100644 --- a/python/feldera/rest/client.py +++ b/python/feldera/rest/client.py @@ -11,6 +11,10 @@ from feldera.rest._httprequests import HttpRequests +def _prepare_boolean_input(value: bool) -> str: + return "true" if value else "false" + + class Client: """ A client for the Feldera HTTP API @@ -399,8 +403,8 @@ def push_to_pipeline( raise ValueError("update_format must be one of 'insert_delete', 'weighted', 'debezium', 'snowflake', 'raw'") # python sends `True` which isn't accepted by the backend - array = "true" if array else "false" - force = "true" if force else "false" + array = _prepare_boolean_input(array) + force = _prepare_boolean_input(force) params = { "force": force, @@ -430,6 +434,7 @@ def listen_to_pipeline( table_name: str, format: str, mode: str = "watch", + backpressure: bool = True, query: Optional[str] = None, quantiles: Optional[int] = None, array: bool = False, @@ -442,6 +447,8 @@ def listen_to_pipeline( :param table_name: The name of the table to listen to :param format: The format of the data, either "json" or "csv" :param mode: The mode to listen in, either "watch" or "snapshot" + :param backpressure: Set True to apply backpressure to the HTTP output connector to make it more reliable, + True by default :param quantiles: For 'quantiles' queries: the number of quantiles to output. The default value is 100 :param query: Query to execute on the table, either "table", "neighborhood" or "quantiles" :param array: Set True to group updates in this stream into JSON arrays, used in conjunction with the @@ -459,13 +466,14 @@ def listen_to_pipeline( params = { "mode": mode, "format": format, + "backpressure": _prepare_boolean_input(backpressure), } if quantiles: params["quantiles"] = quantiles if format == "json": - params["array"] = "true" if array else "false" + params["array"] = _prepare_boolean_input(array) resp = self.http.post( path=f"/pipelines/{pipeline_name}/egress/{table_name}", From 0a6a71812872cd37de07f01e7df405d3a4d9ffa0 Mon Sep 17 00:00:00 2001 From: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> Date: Mon, 27 May 2024 22:08:56 +0545 Subject: [PATCH 2/3] Update docs about backpressure Co-authored-by: Leonid Ryzhyk Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> --- python/feldera/rest/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/feldera/rest/client.py b/python/feldera/rest/client.py index 6bab5f28a79..421a45654dc 100644 --- a/python/feldera/rest/client.py +++ b/python/feldera/rest/client.py @@ -447,7 +447,7 @@ def listen_to_pipeline( :param table_name: The name of the table to listen to :param format: The format of the data, either "json" or "csv" :param mode: The mode to listen in, either "watch" or "snapshot" - :param backpressure: Set True to apply backpressure to the HTTP output connector to make it more reliable, + :param backpressure: When the flag is True (the default), this method waits for the consumer to receive each chunk and blocks the pipeline if the consumer cannot keep up. When this flag is False, the pipeline drops data chunks if the consumer is not keeping up with its output. This prevents a slow consumer from slowing down the entire pipeline. True by default :param quantiles: For 'quantiles' queries: the number of quantiles to output. The default value is 100 :param query: Query to execute on the table, either "table", "neighborhood" or "quantiles" From 53fb27b433be7d6a472e7fd352bd121025238e17 Mon Sep 17 00:00:00 2001 From: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> Date: Mon, 27 May 2024 22:13:58 +0545 Subject: [PATCH 3/3] py: reformat the docs Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com> --- python/feldera/rest/client.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/python/feldera/rest/client.py b/python/feldera/rest/client.py index 421a45654dc..dc0dbdd262f 100644 --- a/python/feldera/rest/client.py +++ b/python/feldera/rest/client.py @@ -447,8 +447,10 @@ def listen_to_pipeline( :param table_name: The name of the table to listen to :param format: The format of the data, either "json" or "csv" :param mode: The mode to listen in, either "watch" or "snapshot" - :param backpressure: When the flag is True (the default), this method waits for the consumer to receive each chunk and blocks the pipeline if the consumer cannot keep up. When this flag is False, the pipeline drops data chunks if the consumer is not keeping up with its output. This prevents a slow consumer from slowing down the entire pipeline. - True by default + :param backpressure: When the flag is True (the default), this method waits for the consumer to receive each + chunk and blocks the pipeline if the consumer cannot keep up. When this flag is False, the pipeline drops + data chunks if the consumer is not keeping up with its output. This prevents a slow consumer from slowing + down the entire pipeline. :param quantiles: For 'quantiles' queries: the number of quantiles to output. The default value is 100 :param query: Query to execute on the table, either "table", "neighborhood" or "quantiles" :param array: Set True to group updates in this stream into JSON arrays, used in conjunction with the