fix lint

Signed-off-by: HaoXuAI <sduxuhao@gmail.com>
feast-dev · franciscojavierarceo · Jun 3, 2025 · May 14, 2025 · May 14, 2025 · May 15, 2025
commit 24e346527e9246bda5b9086acad335949b0e1b4a
@@ -41,6 +41,11 @@ def execute(self, context: ExecutionContext) -> ArrowTableValue:
             end_time=self.end_time,
         )
         arrow_table = retrieval_job.to_arrow()
+        field_mapping = context.column_info.field_mapping
+        if field_mapping:
+            arrow_table = arrow_table.rename_columns(
+                [field_mapping.get(col, col) for col in arrow_table.column_names]
+            )
         return ArrowTableValue(data=arrow_table)
 
 

@@ -453,7 +453,6 @@ def evaluate_func():
             # TODO: Decides if we want to field mapping for pull_latest_from_table_or_query
             # This is default for other offline store.
             df = df[list(columns_to_extract)]
-            df = _run_dask_field_mapping(df, data_source.field_mapping or {})
             df.persist()
             return df
 

diff --git a/sdk/python/feast/infra/offline_stores/ibis.py b/sdk/python/feast/infra/offline_stores/ibis.py
@@ -296,12 +296,6 @@ def pull_all_from_table_or_query_ibis(
         )
     )
 
-    if data_source.field_mapping:
-        for old_field_name in data_source.field_mapping:
-            table = table.rename(
-                {data_source.field_mapping[old_field_name]: old_field_name}
-            )
-
     return IbisRetrievalJob(
         table=table,
         on_demand_feature_views=[],

@@ -168,10 +168,9 @@ def transform_feature(df: DataFrame) -> DataFrame:
             repo_config=spark_environment.config,
             offline_store=SparkOfflineStore(),
             online_store=MagicMock(),
-            registry=registry,
         )
 
-        spark_dag_retrieval_job = engine.get_historical_features(task)
+        spark_dag_retrieval_job = engine.get_historical_features(registry, task)
         spark_df = cast(SparkDAGRetrievalJob, spark_dag_retrieval_job).to_spark_df()
         df_out = spark_df.orderBy("driver_id").to_pandas_on_spark()