What this PR does / why we need it

Fixes materialization failures when using BatchFeatureView + SparkComputeEngine with vector stores and Spark 3.5+.

1. Replace mapInArrow with foreachPartition (main fix)

Spark 3.5 inserts WindowGroupLimitExec upstream of MapInArrowExec when UDFs use Window operations, routing the Python worker through the wrong serialiser:

AttributeError: 'list' object has no attribute 'dtype'

foreachPartition uses pickle — no Arrow UDF bridge — so the mismatch cannot occur.

2. Re-apply session configs on reuse

SparkSession.getOrCreate() silently drops new spark.sql.* / spark.hadoop.* overrides when a session already exists, causing S3 access failures. Now explicitly re-applied after getOrCreate().

3. Fix map_in_pandas dummy yield

Changed pd.DataFrame([pd.Series(range(1, 2))]) → pd.DataFrame({"status": [0]}) to match the declared return schema.

4. Remove redundant _apply_bfv_transformations

Superseded by SparkFeatureBuilder DAG pipeline (#6357) which handles BFV transformations for both materialization and historical retrieval.

5. Add spark_embed utility

Helper for @batch_feature_view UDFs that generate embeddings via SentenceTransformer. Uses localCheckpoint(eager=True) to sever Python lineage before downstream writes.

Which issue(s) this PR fixes

Fixes BatchFeatureView materialization with vector stores (Milvus/Redis), Spark 3.5+ Window operations, and K8s-mode session config drift.

Testing Strategy

• Unit tests: session config forwarding, map_in_pandas schema, empty partition handling, S3 event log dir
• Manual: Materialized BatchFeatureView with PySpark + RAPIDS GPU on K8s (4 executors), 26M+ keys to Redis

Checks

• Tests passing
• Commits signed off
• PR title follows conventional commits