feat: Enable Arrow-based columnar data transfers when to pandas in sp…

…arksource retrieval job

Signed-off-by: tanlocnguyen <tanlocnguyen296@gmail.com>

sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py

@@ -338,6 +338,11 @@ def to_spark_df(self) -> pyspark.sql.DataFrame:
 
     def _to_df_internal(self, timeout: Optional[int] = None) -> pd.DataFrame:
         """Return dataset as Pandas DataFrame synchronously"""
+        spark_session = get_spark_session_or_start_new_with_repoconfig(
+            self._config.offline_store
+        )
+        spark_session.conf.set("spark.sql.execution.arrow.fallback.enabled", "true")
+        spark_session.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
         return self.to_spark_df().toPandas()
 
     def _to_arrow_internal(self, timeout: Optional[int] = None) -> pyarrow.Table:
@@ -496,7 +501,8 @@ def _get_entity_df_event_timestamp_range(
 
 
 def _get_entity_schema(
-    spark_session: SparkSession, entity_df: Union[pandas.DataFrame, str]
+    spark_session: SparkSession,
+    entity_df: Union[pandas.DataFrame, str],
 ) -> Dict[str, np.dtype]:
     if isinstance(entity_df, pd.DataFrame):
         return dict(zip(entity_df.columns, entity_df.dtypes))