feat: Bring Your Own Spark - SparkApplication

[aniketpalu]

What this PR does / why we need it:

Core changes

Upstream (feature_store.py):

• Refactored `materialize()` and `materialize_incremental()` to collect all `MaterializationTask`s and pass them to the engine in a single `batch_engine.materialize(registry, tasks)` call. This enables batching engines to process all feature views in one job. Existing engines are unaffected — `ComputeEngine.materialize()` base class loops tasks internally via `_materialize_one()`.
• Added public `get_provider()` method on `FeatureStore` (wraps existing private `_get_provider()`).
  New engine (spark_application/):
• `config.py`: Pydantic config with Spark resource specs, K8s options, validation (`env` entries must have `name`, `staging_location` warning).
• `compute.py`: Creates a K8s Secret with `feature_store.yaml` + task list, builds SparkApplication CR, polls for completion, retrieves driver logs on failure.
• `job.py`: Maps all 14 SparkApplication states to `MaterializationJobStatus`, with exponential backoff retry on transient API errors.
• `main.py`: Driver script that runs inside the pod — loads config from Secret, materializes each feature view via `provider.materialize_single_feature_view()`. Supports concurrent FV processing via ThreadPoolExecutor + FAIR scheduler.
• `Dockerfile`: Based on `apache/spark:4.0.1`, installs `feast[redis]==0.64.0`, includes Hadoop AWS JARs for S3A support.

Design decisions

• Two config rewrites in driver pod: `batch_engine` → `spark.engine` (prevents recursive CRD creation, uses local SparkSession), `registry` → `remote` when `registry_address` is set (pod can't access server filesystem).
• Driver calls `provider.materialize_single_feature_view()`, not `store.materialize()` — avoids FeatureViewState transition conflicts (MATERIALIZING→MATERIALIZING is invalid) and duplicate OpenLineage/MLflow emissions.
• SQLite online store rejected at init — data written inside a pod is lost on termination.
• Kueue integration via `queue_name` config field → `kueue.x-k8s.io/queue-name` label.

Validated on

• OpenShift AI (ROSA) with Spark Operator v2.5.0, MinIO (S3A), Redis
• 5 feature views, 4800 rows each, 2 executors: 2.4x speedup with concurrent mode

Test plan

• 12 unit tests: config defaults, SQLite rejection, registry validation, config rewrites, CR structure, state mapping (14 states), cleanup 404 handling, timeout behavior, job naming
• Existing tests unaffected: 29 feature_view_state + 22 local_feature_store tests pass
• Integration test on cluster with real Spark Operator"

Which issue(s) this PR fixes:

Checks

• I've made sure the tests are passing.
• My commits are signed off (git commit -s)
• My PR title follows conventional commits format

Testing Strategy

• Unit tests
• Integration tests
• Manual tests
• Testing is not required for this change

Misc

[jyejare]

@aniketpalu lint checks are failing

[jyejare]

While the core architecture is solid, there are several critical issues around thread safety, error handling, and resource management that need addressing before merge.

[jyejare]

[Critical] Race condition in state management

The state transition and rollback logic has potential race conditions when multiple materialization jobs run concurrently. The previous_states dictionary and feature view state updates are not atomic, which could lead to inconsistent states or lost rollbacks.

Suggested:

Suggested change

	# TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths
	# Use locks or atomic operations for state management
	import threading
	state_lock = threading.Lock()
	with state_lock:
	for fv, job in zip(regular_fvs, jobs):
	fv_status = job.status()
	if fv_status == MaterializationJobStatus.ERROR:
	failed_fvs.append(fv)
	if first_error is None and job.error():
	first_error = job.error()
	else:
	succeeded_fvs.append(fv)
	if failed_fvs:
	self._rollback_fv_states(failed_fvs, previous_states)

[aniketpalu]

The threading race condition described isn't present because previous_states is a local variable (line 2551) and all loops (for fv, job in zip(...)) are sequential Python for loops, not threaded. Each materialize() call creates its own dict.

However, reviewing this code path revealed a shared-object bug in our engine's _build_per_fv_jobs. Previously, all succeeded FVs received the same SparkApplicationMaterializationJob reference. During _wait_for_completion, polling sets _error on that object when the SparkApp transitions to FAILED. Later, when feature_store.py calls job.status() for each FV, they all hit the if self._error is not None: return ERROR early-return, even for the FVs that actually succeeded and wrote AVAILABLE_ONLINE to the registry.

[jyejare]

[Suggestion] Improve configuration validation

The env validation only checks for 'name' field but doesn't validate that 'value' exists or that the overall structure is correct for Kubernetes environment variables.

Suggested:

Suggested change

	"staging_location is configured but only used for "
	"get_historical_features (not yet supported by this engine). "
	"It will be ignored for materialize operations.",
	stacklevel=2,
	)
	for i, entry in enumerate(self.env):
	if "name" not in entry:
	raise ValueError(
	f"env[{i}] is missing required 'name' key. "
	f"Each env entry must have at least 'name' and 'value'. Got: {entry}"
	)
	for i, entry in enumerate(self.env):
	if not isinstance(entry, dict):
	raise ValueError(
	f"env[{i}] must be a dict, got {type(entry).__name__}: {entry}"
	)
	if "name" not in entry:
	raise ValueError(
	f"env[{i}] is missing required 'name' key. "
	f"Each env entry must have at least 'name' and 'value'. Got: {entry}"
	)
	if "value" not in entry and "valueFrom" not in entry:
	raise ValueError(
	f"env[{i}] must have either 'value' or 'valueFrom'. Got: {entry}"
	)

[codecov-commenter]

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

❌ Patch coverage is 42.05776% with 321 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.00%. Comparing base (14e0a83) to head (cbd4051).

Files with missing lines	Patch %	Lines
...st/infra/compute_engines/spark_application/main.py	0.00%	108 Missing ⚠️
sdk/python/feast/feature_store.py	20.00%	98 Missing and 2 partials ⚠️
...infra/compute_engines/spark_application/compute.py	56.08%	70 Missing and 13 partials ⚠️
...ast/infra/compute_engines/spark_application/job.py	70.12%	18 Missing and 5 partials ⚠️
.../infra/compute_engines/spark_application/config.py	87.75%	3 Missing and 3 partials ⚠️
sdk/python/feast/infra/compute_engines/base.py	83.33%	1 Missing ⚠️
❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #6550      +/-   ##
==========================================
- Coverage   46.02%   46.00%   -0.03%     
==========================================
  Files         402      406       +4     
  Lines       47656    48164     +508     
  Branches     6741     6814      +73     
==========================================
+ Hits        21936    22157     +221     
- Misses      24196    24461     +265     
- Partials     1524     1546      +22     

Flag	Coverage Δ
go-feature-server	30.58% <ø> (ø)
python-unit	47.32% <42.05%> (-0.05%)	⬇️

Files with missing lines	Coverage Δ
sdk/python/feast/repo_config.py	77.38% <ø> (ø)
sdk/python/feast/infra/compute_engines/base.py	78.84% <83.33%> (+0.58%)	⬆️
.../infra/compute_engines/spark_application/config.py	87.75% <87.75%> (ø)
...ast/infra/compute_engines/spark_application/job.py	70.12% <70.12%> (ø)
...infra/compute_engines/spark_application/compute.py	56.08% <56.08%> (ø)
sdk/python/feast/feature_store.py	41.06% <20.00%> (-1.33%)	⬇️
...st/infra/compute_engines/spark_application/main.py	0.00% <0.00%> (ø)

---

Continue to review full report in Codecov by Harness.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 14e0a83...cbd4051. Read the comment docs.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[ntkathole]

apply_materialization is called twice for each succeeded FV - once by the driver pod, once by the server after the job completes.

[ntkathole]

@aniketpalu This still exists

[aniketpalu]

Agreed, the double write was duplicating materialization intervals. The driver pod remains the owner of apply_materialization (needed for per-FV AVAILABLE_ONLINE / partial success). The server batch path now skips the second call when the engine sets applies_materialization=True (SparkApplication only). Other engines still apply on the server as before.

[ntkathole]

This should be async.

This blocks the Feast server thread for the entire duration of the Spark job (up to job_timeout_seconds, default 1 hour). For a server handling multiple clients, this is a scalability concern.

[aniketpalu]

Agreed this is a real scalability concern for the sync path: _wait_for_completion holds the caller until the SparkApplication finishes.

Making the engine poll itself async would redesign the ComputeEngine.materialize(). The intended non-blocking path is remote materialization in #6590 (POST /materialize-async + registry polling). We'll address true submit-without-wait server behavior there so the async path does not hold a worker thread for the full Spark job.

Would it be okay to leaving this as-is in #6550 and tracked against #6590?

[ntkathole]

fragile

batch_engine = getattr(provider, "batch_engine", None)
if batch_engine and getattr(batch_engine, "supports_batch", False):

[aniketpalu]

Done

[ntkathole]

looks good