Skip to content

Commit 6c76791

Browse files
Leonid Ryzhykryzhyk
Leonid Ryzhyk
authored and
ryzhyk
committed
[py] Take advantage of pipeline field selector.
Python SDK called the GET /pipeline/<name> endpoint without using the selector argument. This meant that even when polling the manager waiting for pipeline compilation to complete or for a specific pipeline status, it retrieved the entire source code, schema, and the list of compiler errors and warnings every time. This could cause a lot of unnecessary load on the mananger. This commit changes the SDK to instead use the "status" selector where possible to only retrieve fields needed for runtime monitoring of the pipeline. This approach may still be a bit crude, and we may want an even more fine grained API, but it should improve things significantly. Signed-off-by: Leonid Ryzhyk <leonid@feldera.com>
1 parent 7136cad commit 6c76791

6 files changed

Lines changed: 59 additions & 45 deletions

File tree

‎python/feldera/_callback_runner.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import pandas as pd
77
from feldera import FelderaClient
88
from feldera._helpers import dataframe_from_response
9+
from feldera.enums import PipelineFieldSelector
910

1011

1112
class _CallbackRunnerInstruction(Enum):
@@ -38,7 +39,7 @@ def run(self):
3839
:meta private:
3940
"""
4041

41-
pipeline = self.client.get_pipeline(self.pipeline_name)
42+
pipeline = self.client.get_pipeline(self.pipeline_name, PipelineFieldSelector.ALL)
4243

4344
schemas = pipeline.tables + pipeline.views
4445
for schema in schemas:

‎python/feldera/enums.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
from enum import Enum
22
from typing import Optional
33

4-
54
class CompilationProfile(Enum):
65
"""
76
The compilation profile to use when compiling the program.
@@ -336,3 +335,10 @@ def from_str(value):
336335
raise ValueError(
337336
f"Unknown value '{value}' for enum {FaultToleranceModel.__name__}"
338337
)
338+
339+
class PipelineFieldSelector(Enum):
340+
ALL = "all"
341+
"""Select all fields of a pipeline."""
342+
343+
STATUS = "status"
344+
"""Select only the fields required to know the status of a pipeline."""

‎python/feldera/pipeline.py‎

Lines changed: 34 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212

1313
from feldera.rest.errors import FelderaAPIError
1414
from feldera.enums import (
15+
PipelineFieldSelector,
1516
PipelineStatus,
1617
ProgramStatus,
1718
CheckpointStatus,
@@ -60,22 +61,24 @@ def __setup_output_listeners(self):
6061
# block until the callback runner is ready
6162
queue.join()
6263

63-
def refresh(self):
64+
def refresh(self, field_selector: PipelineFieldSelector):
6465
"""
6566
Calls the backend to get the updated, latest version of the pipeline.
6667
68+
:param field_selector: Choose what pipeline information to refresh; see PipelineFieldSelector enum definition.
69+
6770
:raises FelderaConnectionError: If there is an issue connecting to the backend.
6871
"""
6972

70-
self._inner = self.client.get_pipeline(self.name)
73+
self._inner = self.client.get_pipeline(self.name, field_selector)
7174

7275
def status(self) -> PipelineStatus:
7376
"""
7477
Return the current status of the pipeline.
7578
"""
7679

7780
try:
78-
self.refresh()
81+
self.refresh(PipelineFieldSelector.STATUS)
7982
return PipelineStatus.from_str(self._inner.deployment_status)
8083

8184
except FelderaAPIError as err:
@@ -124,7 +127,7 @@ def input_pandas(self, table_name: str, df: pandas.DataFrame, force: bool = Fals
124127

125128
ensure_dataframe_has_columns(df)
126129

127-
pipeline = self.client.get_pipeline(self.name)
130+
pipeline = self.client.get_pipeline(self.name, PipelineFieldSelector.ALL)
128131
if table_name.lower() != "now" and table_name.lower() not in [
129132
tbl.name.lower() for tbl in pipeline.tables
130133
]:
@@ -672,7 +675,7 @@ def get(name: str, client: FelderaClient) -> "Pipeline":
672675
"""
673676

674677
try:
675-
inner = client.get_pipeline(name)
678+
inner = client.get_pipeline(name, PipelineFieldSelector.ALL)
676679
return Pipeline._from_inner(inner, client)
677680
except FelderaAPIError as err:
678681
if err.status_code == 404:
@@ -948,7 +951,7 @@ def program_code(self) -> str:
948951
Return the program SQL code of the pipeline.
949952
"""
950953

951-
self.refresh()
954+
self.refresh(PipelineFieldSelector.ALL)
952955
return self._inner.program_code
953956

954957
def modify(
@@ -988,7 +991,7 @@ def storage_status(self) -> StorageStatus:
988991
Return the storage status of the pipeline.
989992
"""
990993

991-
self.refresh()
994+
self.refresh(PipelineFieldSelector.STATUS)
992995
return StorageStatus.from_str(self._inner.storage_status)
993996

994997
def program_status(self) -> ProgramStatus:
@@ -1000,47 +1003,47 @@ def program_status(self) -> ProgramStatus:
10001003
Rust code to a binary.
10011004
"""
10021005

1003-
self.refresh()
1006+
self.refresh(PipelineFieldSelector.STATUS)
10041007
return ProgramStatus.from_value(self._inner.program_status)
10051008

10061009
def program_status_since(self) -> datetime:
10071010
"""
10081011
Return the timestamp when the current program status was set.
10091012
"""
10101013

1011-
self.refresh()
1014+
self.refresh(PipelineFieldSelector.STATUS)
10121015
return datetime.fromisoformat(self._inner.program_status_since)
10131016

10141017
def udf_rust(self) -> str:
10151018
"""
10161019
Return the Rust code for UDFs.
10171020
"""
10181021

1019-
self.refresh()
1022+
self.refresh(PipelineFieldSelector.ALL)
10201023
return self._inner.udf_rust
10211024

10221025
def udf_toml(self) -> str:
10231026
"""
10241027
Return the Rust dependencies required by UDFs (in the TOML format).
10251028
"""
10261029

1027-
self.refresh()
1030+
self.refresh(PipelineFieldSelector.ALL)
10281031
return self._inner.udf_toml
10291032

10301033
def program_config(self) -> Mapping[str, Any]:
10311034
"""
10321035
Return the program config of the pipeline.
10331036
"""
10341037

1035-
self.refresh()
1038+
self.refresh(PipelineFieldSelector.ALL)
10361039
return self._inner.program_config
10371040

10381041
def runtime_config(self) -> RuntimeConfig:
10391042
"""
10401043
Return the runtime config of the pipeline.
10411044
"""
10421045

1043-
self.refresh()
1046+
self.refresh(PipelineFieldSelector.ALL)
10441047
return RuntimeConfig.from_dict(self._inner.runtime_config)
10451048

10461049
def set_runtime_config(self, runtime_config: RuntimeConfig):
@@ -1065,55 +1068,55 @@ def id(self) -> str:
10651068
Return the ID of the pipeline.
10661069
"""
10671070

1068-
self.refresh()
1071+
self.refresh(PipelineFieldSelector.STATUS)
10691072
return self._inner.id
10701073

10711074
def description(self) -> str:
10721075
"""
10731076
Return the description of the pipeline.
10741077
"""
10751078

1076-
self.refresh()
1079+
self.refresh(PipelineFieldSelector.STATUS)
10771080
return self._inner.description
10781081

10791082
def tables(self) -> List[SQLTable]:
10801083
"""
10811084
Return the tables of the pipeline.
10821085
"""
10831086

1084-
self.refresh()
1087+
self.refresh(PipelineFieldSelector.ALL)
10851088
return self._inner.tables
10861089

10871090
def views(self) -> List[SQLView]:
10881091
"""
10891092
Return the views of the pipeline.
10901093
"""
10911094

1092-
self.refresh()
1095+
self.refresh(PipelineFieldSelector.ALL)
10931096
return self._inner.views
10941097

10951098
def created_at(self) -> datetime:
10961099
"""
10971100
Return the creation time of the pipeline.
10981101
"""
10991102

1100-
self.refresh()
1103+
self.refresh(PipelineFieldSelector.STATUS)
11011104
return datetime.fromisoformat(self._inner.created_at)
11021105

11031106
def version(self) -> int:
11041107
"""
11051108
Return the version of the pipeline.
11061109
"""
11071110

1108-
self.refresh()
1111+
self.refresh(PipelineFieldSelector.STATUS)
11091112
return self._inner.version
11101113

11111114
def program_version(self) -> int:
11121115
"""
11131116
Return the program version of the pipeline.
11141117
"""
11151118

1116-
self.refresh()
1119+
self.refresh(PipelineFieldSelector.STATUS)
11171120
return self._inner.program_version
11181121

11191122
def deployment_status_since(self) -> datetime:
@@ -1122,15 +1125,15 @@ def deployment_status_since(self) -> datetime:
11221125
was set.
11231126
"""
11241127

1125-
self.refresh()
1128+
self.refresh(PipelineFieldSelector.STATUS)
11261129
return datetime.fromisoformat(self._inner.deployment_status_since)
11271130

11281131
def deployment_config(self) -> Mapping[str, Any]:
11291132
"""
11301133
Return the deployment config of the pipeline.
11311134
"""
11321135

1133-
self.refresh()
1136+
self.refresh(PipelineFieldSelector.ALL)
11341137
return self._inner.deployment_config
11351138

11361139
def deployment_desired_status(self) -> DeploymentDesiredStatus:
@@ -1139,15 +1142,15 @@ def deployment_desired_status(self) -> DeploymentDesiredStatus:
11391142
This is the next state that the pipeline should transition to.
11401143
"""
11411144

1142-
self.refresh()
1145+
self.refresh(PipelineFieldSelector.STATUS)
11431146
return DeploymentDesiredStatus.from_str(self._inner.deployment_desired_status)
11441147

11451148
def deployment_resources_desired_status(self) -> DeploymentResourcesDesiredStatus:
11461149
"""
11471150
Return the desired status of the the deployment resources.
11481151
"""
11491152

1150-
self.refresh()
1153+
self.refresh(PipelineFieldSelector.STATUS)
11511154
return DeploymentResourcesDesiredStatus.from_str(
11521155
self._inner.deployment_resources_desired_status
11531156
)
@@ -1157,7 +1160,7 @@ def deployment_resources_status(self) -> DeploymentResourcesStatus:
11571160
Return the status of the deployment resources.
11581161
"""
11591162

1160-
self.refresh()
1163+
self.refresh(PipelineFieldSelector.STATUS)
11611164
return DeploymentResourcesStatus.from_str(
11621165
self._inner.deployment_resources_status
11631166
)
@@ -1167,7 +1170,7 @@ def deployment_runtime_desired_status(self) -> DeploymentRuntimeDesiredStatus:
11671170
Return the deployment runtime desired status.
11681171
"""
11691172

1170-
self.refresh()
1173+
self.refresh(PipelineFieldSelector.STATUS)
11711174
return DeploymentRuntimeDesiredStatus.from_str(
11721175
self._inner.deployment_runtime_desired_status
11731176
)
@@ -1177,7 +1180,7 @@ def deployment_runtime_status(self) -> DeploymentRuntimeStatus:
11771180
Return the deployment runtime status.
11781181
"""
11791182

1180-
self.refresh()
1183+
self.refresh(PipelineFieldSelector.STATUS)
11811184
return DeploymentRuntimeStatus.from_str(self._inner.deployment_runtime_status)
11821185

11831186
def deployment_error(self) -> Mapping[str, Any]:
@@ -1186,7 +1189,7 @@ def deployment_error(self) -> Mapping[str, Any]:
11861189
Returns an empty string if there is no error.
11871190
"""
11881191

1189-
self.refresh()
1192+
self.refresh(PipelineFieldSelector.STATUS)
11901193
return self._inner.deployment_error
11911194

11921195
def deployment_location(self) -> str:
@@ -1196,7 +1199,7 @@ def deployment_location(self) -> str:
11961199
at runtime (a TCP port number or a URI).
11971200
"""
11981201

1199-
self.refresh()
1202+
self.refresh(PipelineFieldSelector.STATUS)
12001203
return self._inner.deployment_location
12011204

12021205
def program_info(self) -> Mapping[str, Any]:
@@ -1207,7 +1210,7 @@ def program_info(self) -> Mapping[str, Any]:
12071210
and the SQL program schema.
12081211
"""
12091212

1210-
self.refresh()
1213+
self.refresh(PipelineFieldSelector.ALL)
12111214
return self._inner.program_info
12121215

12131216
def program_error(self) -> Mapping[str, Any]:
@@ -1217,7 +1220,7 @@ def program_error(self) -> Mapping[str, Any]:
12171220
`sql_compilation` and `rust_compilation` will be 0.
12181221
"""
12191222

1220-
self.refresh()
1223+
self.refresh(PipelineFieldSelector.ALL)
12211224
return self._inner.program_error
12221225

12231226
def errors(self) -> List[Mapping[str, Any]]:

‎python/feldera/pipeline_builder.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
from feldera.rest.feldera_client import FelderaClient
55
from feldera.rest.pipeline import Pipeline as InnerPipeline
66
from feldera.pipeline import Pipeline
7-
from feldera.enums import CompilationProfile
7+
from feldera.enums import CompilationProfile, PipelineFieldSelector
88
from feldera.runtime_config import RuntimeConfig
99
from feldera.rest.errors import FelderaAPIError
1010

@@ -60,7 +60,7 @@ def create(self) -> Pipeline:
6060
raise ValueError("Name and SQL are required to create a pipeline")
6161

6262
try:
63-
if self.client.get_pipeline(self.name) is not None:
63+
if self.client.get_pipeline(self.name, PipelineFieldSelector.STATUS) is not None:
6464
raise RuntimeError(f"Pipeline with name {self.name} already exists")
6565
except FelderaAPIError as err:
6666
if err.error_code != "UnknownPipelineName":

0 commit comments

Comments
 (0)