Skip to content

Commit 11a47b4

Browse files
Marcus-RostiMarcus-Rosti
authored
feat: Add Trino compute engine for batch retrieval and materialization (#6898)
* feat: Adds Trino as a compute engine Signed-off-by: Marcus Rosti <[email protected]> * fix: Join entity_df during backfill Signed-off-by: Marcus Rosti <[email protected]> * fix: Exclude internal _feast_rn column from Trino dedup output Signed-off-by: Marcus Rosti <[email protected]> * fix: Adds a trino write type (append/merge/overwrite) Signed-off-by: Marcus Rosti <[email protected]> * fix: Throw an error when semicolons are present in filters Signed-off-by: Marcus Rosti <[email protected]> * fix: Throw error when client is not configured Signed-off-by: Marcus Rosti <[email protected]> * fix: Implements a trino integration test Signed-off-by: Marcus Rosti <[email protected]> * fix: More efficient streaming for pyarrow Signed-off-by: Marcus Rosti <[email protected]> * fix: Qualified select statements Signed-off-by: Marcus Rosti <[email protected]> * fix: Validate Trino client configuration in historical retrieval Signed-off-by: Marcus Rosti <[email protected]> --------- Signed-off-by: Marcus Rosti <[email protected]>
1 parent cc07c82 commit 11a47b4

18 files changed

Lines changed: 5055 additions & 1 deletion

File tree

‎docs/getting-started/components/compute-engine.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ engines.
2222
|-------------------------|-------------------------------------------------------------------------------------------------|------------|------|
2323
| LocalComputeEngine | Runs on Arrow + Pandas/Polars/Dask etc., designed for light weight transformation. | ✅ | |
2424
| SparkComputeEngine | Runs on Apache Spark, designed for large-scale distributed feature generation. | ✅ | |
25+
| TrinoComputeEngine | Runs on Trino, designed for scalable feature generation using Trino SQL. | ✅ | [docs](../../reference/compute-engine/trino.md) |
2526
| SnowflakeComputeEngine | Runs on Snowflake, designed for scalable feature generation using Snowflake SQL. | ✅ | |
2627
| LambdaComputeEngine | Runs on AWS Lambda, designed for serverless feature generation. | ✅ | |
2728
| FlinkComputeEngine | Runs on Apache Flink, designed for distributed feature generation through PyFlink Table API. | ✅ | |

‎docs/reference/compute-engine/trino.md‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
# Trino
2+
3+
## Description
4+
5+
Trino Compute Engine provides a distributed execution engine for batch materialization operations (`materialize` and `materialize-incremental`) and historical retrieval operations (`get_historical_features`).
6+
7+
It is designed to handle large-scale data processing directly on Trino clusters without moving raw data to the client machine.
8+
9+
### Design
10+
11+
The Trino Compute engine is implemented as a subclass of `feast.infra.compute_engines.base.ComputeEngine`.
12+
13+
The engine supports the following features:
14+
- **Pushdown SQL execution**: Compiles feature pipeline operations (filtering, deduplication, time-windowed aggregations, point-in-time joins, and transformations) into Trino SQL Common Table Expressions (CTEs) executed directly on the Trino cluster.
15+
- **Streaming online materialization**: Streams query results in chunks as PyArrow record batches directly to the online store via a thread pool, avoiding loading entire datasets into client memory.
16+
- **Atomic offline table writes**: Materializes offline tables using a temporary staging table and rename swap (`CREATE TABLE ...__staging AS ...; DROP TABLE ...; ALTER TABLE ...__staging RENAME TO ...`).
17+
- **Configuration inheritance**: Inherits connection details from the Trino offline store when both are configured.
18+
19+
---
20+
21+
## Example
22+
23+
```yaml
24+
project: feast_trino_project
25+
registry: data/registry.db
26+
provider: local
27+
28+
offline_store:
29+
type: trino.offline
30+
host: localhost
31+
port: 8080
32+
catalog: iceberg
33+
dataset: feast_offline
34+
user: feast_user
35+
36+
batch_engine:
37+
type: trino.engine
38+
batch_size: 10000
39+
write_concurrency: 4
40+
41+
online_store:
42+
type: redis
43+
connection_string: localhost:6379
44+
```
45+
46+
---
47+
48+
## Example in Python
49+
50+
```python
51+
from datetime import timedelta
52+
from feast import (
53+
BatchFeatureView,
54+
Entity,
55+
Field,
56+
)
57+
from feast.aggregation import Aggregation
58+
from feast.infra.offline_stores.contrib.trino_offline_store.trino_source import TrinoSource
59+
from feast.transformation.mode import TransformationMode
60+
from feast.transformation.trino_transformation import TrinoTransformation
61+
from feast.types import Float32, Int32
62+
from feast.value_type import ValueType
63+
64+
# 1. Define Entity
65+
driver = Entity(
66+
name="driver_id",
67+
value_type=ValueType.INT32,
68+
join_keys=["driver_id"],
69+
description="Driver identifier",
70+
)
71+
72+
# 2. Define Trino Batch Source
73+
driver_stats_source = TrinoSource(
74+
name="driver_stats_source",
75+
table="iceberg.feast_offline.driver_hourly_stats",
76+
timestamp_field="event_timestamp",
77+
created_timestamp_column="created_timestamp",
78+
)
79+
80+
# 3. Define Pure Trino SQL Transformation
81+
# The '{}' placeholder will be substituted with the upstream CTE/table name
82+
transform = TrinoTransformation(
83+
mode=TransformationMode.TRINO_SQL,
84+
udf="""
85+
SELECT
86+
driver_id,
87+
event_timestamp,
88+
conv_rate * 2.0 AS conv_rate,
89+
acc_rate * 2.0 AS acc_rate
90+
FROM {}
91+
""",
92+
)
93+
94+
# 4. Define Feature View with SQL Transformation & Aggregations
95+
driver_hourly_stats_fv = BatchFeatureView(
96+
name="driver_hourly_stats",
97+
entities=[driver],
98+
ttl=timedelta(days=3),
99+
feature_transformation=transform,
100+
aggregations=[
101+
Aggregation(column="conv_rate", function="sum"),
102+
Aggregation(column="acc_rate", function="avg"),
103+
],
104+
schema=[
105+
Field(name="sum_conv_rate", dtype=Float32),
106+
Field(name="avg_acc_rate", dtype=Float32),
107+
Field(name="driver_id", dtype=Int32),
108+
],
109+
online=True,
110+
offline=True,
111+
source=driver_stats_source,
112+
)
113+
```
114+
115+
---
116+
117+
## Retrieving and Materializing Features
118+
119+
```python
120+
import pandas as pd
121+
from datetime import datetime, timezone
122+
from feast import FeatureStore
123+
124+
fs = FeatureStore(repo_path=".")
125+
126+
entity_df: pd.DataFrame
127+
128+
# Lazy Historical Retrieval:
129+
# Compiles into a single Trino ANSI SQL query with CTEs.
130+
job = fs.get_historical_features(
131+
entity_df=entity_df,
132+
features=[
133+
"driver_hourly_stats:sum_conv_rate",
134+
"driver_hourly_stats:avg_acc_rate",
135+
],
136+
)
137+
138+
# Inspect the compiled pure SQL query:
139+
print(job.to_sql())
140+
141+
# Execute on cluster and return pandas dataframe:
142+
training_df = job.to_df()
143+
144+
# Materialize to Online and Offline Stores:
145+
fs.materialize(
146+
start_date=datetime(2025, 1, 1, tzinfo=timezone.utc),
147+
end_date=datetime(2025, 1, 15, tzinfo=timezone.utc),
148+
)
149+
```

‎sdk/python/feast/infra/compute_engines/dag/model.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,3 +7,4 @@ class DAGFormat(str, Enum):
77
ARROW = "arrow"
88
RAY = "ray"
99
FLINK = "flink"
10+
TRINO = "trino"

‎sdk/python/feast/infra/compute_engines/trino/__init__.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
from feast.infra.compute_engines.trino.compute import (
2+
TrinoComputeEngine,
3+
TrinoComputeEngineConfig,
4+
)
5+
6+
__all__ = ["TrinoComputeEngine", "TrinoComputeEngineConfig"]

0 commit comments

Comments
 (0)