Blocker: entity_df is silently ignored for historical retrieval.

This class inherits the base _build() which only creates a JoinNode when there are upstream input_nodes (multi-FeatureView DAGs). For the common single-FeatureView case, the DAG is:

ReadNode -> FilterNode -> Agg/Dedup -> WriteNode — no JoinNode is ever created.

This means get_historical_features(entity_df=..., features=[...]) returns the full feature table instead of the PIT-correct subset matching the entity_df.

How Flink solves this: FlinkFeatureBuilder overrides _build() and adds:

if self._should_join_entity_df():
    last_node = self.build_join_node(view, [last_node])

How Spark solves this: SparkReadNode delegates to create_offline_store_retrieval_job() which handles entity_df at the offline store level.

Trino does neither. Please override _build() here (like Flink) to inject the entity_df join step for HistoricalRetrievalTask.

Note: the component test test_trino_compute_engine_get_historical_features hides this because mock_client.execute_query returns canned data — it never validates the generated SQL includes entity_df columns.

Regression: _feast_rn column leaks into output.

This outer SELECT * includes the _feast_rn ROW_NUMBER column in the CTE output. Downstream nodes and the final result schema will contain this spurious column.

How peer engines handle this:

• Spark SparkDedupNode: .drop("row_num") — explicitly removes the internal column
• Flink FlinkDedupNode: Uses SELECT {explicit_column_list} (excludes DEDUP_ROW_NUMBER) + _drop_internal_columns() in the output node

Fix: Either wrap in an outer CTE that lists all columns except _feast_rn, or track the upstream column list in TrinoQueryPlan.columns and use it here.

Data loss risk: non-atomic swap + destructive offline write.

Two issues here:

1. Non-atomic swap loses data if RENAME fails:
This line DROPs the target table before the RENAME. If the RENAME on line 617 fails (e.g., permission error, catalog issue), the original target is already gone. The except block then also drops the staging table — all data is permanently lost.

Fix: Reverse the order: RENAME target to backup first, then RENAME staging to target, then DROP backup.

2. Regression vs peers: offline write replaces entire table instead of appending:

• Spark uses write.mode("append").save(path) — preserves existing data
• Flink uses offline_store.offline_write_batch() — delegates to store's append logic
• Trino uses DROP + CREATE TABLE AS — destroys ALL existing offline data

Every materialize() call for a feature view with offline=True wipes all previously materialized data.

Fix: Use INSERT INTO target_table (SELECT ... FROM ...) for appending, or implement a time-partitioned merge. At minimum, document this as a destructive operation.

SQL injection surface: filter_condition is injected directly into SQL with f"({self.filter_condition})" — no sanitization or parameterization. While typically developer-provided, a malformed string will execute arbitrary Trino SQL.

At minimum, add a docstring noting that filter_condition must be trusted input.

self.client can be None here (if host/catalog/user aren't configured and offline store isn't TrinoOfflineStoreConfig). The # type: ignore suppresses the type error but None will cause a confusing runtime crash inside TrinoFeatureBuilder.

Please add an explicit guard:

if self.client is None:
    raise RuntimeError(
        "Trino client is not configured. Set host, catalog, and user "
        "in batch_engine config or use a TrinoOfflineStoreConfig."
    )

Missing: integration tests against a real Trino cluster.

The component and unit tests use mocked Trino clients throughout, which is good for unit-level coverage but misses:

1. SQL compilation correctness — generated CTEs are never executed against a Trino parser/planner
2. entity_df upload + PIT join — upload_pandas_dataframe_to_trino is mocked
3. Streaming batch correctness — stream_trino_arrow_batches uses cursor._query (private API), never tested against the real client
4. Offline staging swap atomicity — the DROP/RENAME sequence is never tested for failure modes
5. Type mapping round-trips — from_feast_to_trino_type and trino_to_pa_value_type end-to-end

Other engines have integration suites under tests/integration/compute_engines/. Consider adding a Trino integration test (e.g., using Testcontainers with trinodb/trino Docker image) that validates at least:

• Historical retrieval with entity_df produces PIT-correct results
• Materialization writes correct data to online + offline stores
• Streaming batch size respects the config

Also: test_trino_compute_engine_get_historical_features doesn't verify the generated SQL includes entity_df join conditions — it should assert entity columns appear in the compiled query.

stream_trino_arrow_batches uses private APIs: client._get_cursor() and cursor._query. These are internal to trino-python-client and can break on library upgrades.

Consider using cursor.description (public API) for column metadata instead of cursor._query.columns, and adding a version pin on the trino dependency.

Minor: SELECT _entity.*, {latest_cte}.* will produce duplicate columns when entity_df and the feature table share join key columns (e.g., driver_id). Downstream Arrow/Pandas conversion may fail or silently rename them.

Consider qualifying the feature-side SELECT to exclude join keys already present from _entity.*.