@@ -31,10 +31,11 @@ This system builds and executes DAGs (Directed Acyclic Graphs) of typed operatio
 - Supports point-in-time joins and large-scale materialization
 - Integrates with `SparkOfflineStore` and `SparkMaterializationJob`
 
-### 🧪 LocalComputeEngine (WIP)
+### 🧪 LocalComputeEngine
 
-- Runs on Arrow + Pandas (or optionally DuckDB)
+- Runs on Arrow + Specified backend (e.g., Pandas, Polars)
 - Designed for local dev, testing, or lightweight feature generation
+- Supports `LocalMaterializationJob` and `LocalHistoricalRetrievalJob`
 
 ---
 
 
@@ -0,0 +1,59 @@
+import enum
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Callable, Optional, Union
+
+from tqdm import tqdm
+
+from feast import BatchFeatureView, FeatureView, StreamFeatureView
+
+
+@dataclass
+class MaterializationTask:
+    """
+    A MaterializationTask represents a unit of data that needs to be materialized from an
+    offline store to an online store.
+    """
+
+    project: str
+    feature_view: Union[BatchFeatureView, StreamFeatureView, FeatureView]
+    start_time: datetime
+    end_time: datetime
+    tqdm_builder: Callable[[int], tqdm]
+
+
+class MaterializationJobStatus(enum.Enum):
+    WAITING = 1
+    RUNNING = 2
+    AVAILABLE = 3
+    ERROR = 4
+    CANCELLING = 5
+    CANCELLED = 6
+    SUCCEEDED = 7
+    PAUSED = 8
+    RETRYING = 9
+
+
+class MaterializationJob(ABC):
+    """
+    A MaterializationJob represents an ongoing or executed process that materializes data as per the
+    definition of a materialization task.
+    """
+
+    task: MaterializationTask
+
+    @abstractmethod
+    def status(self) -> MaterializationJobStatus: ...
+
+    @abstractmethod
+    def error(self) -> Optional[BaseException]: ...
+
+    @abstractmethod
+    def should_be_retried(self) -> bool: ...
+
+    @abstractmethod
+    def job_id(self) -> str: ...
+
+    @abstractmethod
+    def url(self) -> Optional[str]: ...
@@ -4,12 +4,12 @@
 import pyarrow as pa
 
 from feast import RepoConfig
-from feast.infra.compute_engines.dag.context import ColumnInfo, ExecutionContext
-from feast.infra.compute_engines.tasks import HistoricalRetrievalTask
-from feast.infra.materialization.batch_materialization_engine import (
+from feast.infra.common.materialization_job import (
     MaterializationJob,
     MaterializationTask,
 )
+from feast.infra.common.retrieval_task import HistoricalRetrievalTask
+from feast.infra.compute_engines.dag.context import ColumnInfo, ExecutionContext
 from feast.infra.offline_stores.offline_store import OfflineStore
 from feast.infra.online_stores.online_store import OnlineStore
 from feast.infra.registry.registry import Registry
 
@@ -1,11 +1,10 @@
 from abc import ABC, abstractmethod
 from typing import Union
 
-from feast import BatchFeatureView, FeatureView, StreamFeatureView
+from feast.infra.common.materialization_job import MaterializationTask
+from feast.infra.common.retrieval_task import HistoricalRetrievalTask
 from feast.infra.compute_engines.dag.node import DAGNode
 from feast.infra.compute_engines.dag.plan import ExecutionPlan
-from feast.infra.compute_engines.tasks import HistoricalRetrievalTask
-from feast.infra.materialization.batch_materialization_engine import MaterializationTask
 
 
 class FeatureBuilder(ABC):
@@ -16,10 +15,9 @@ class FeatureBuilder(ABC):
 
     def __init__(
         self,
-        feature_view: Union[BatchFeatureView, StreamFeatureView, FeatureView],
         task: Union[MaterializationTask, HistoricalRetrievalTask],
     ):
-        self.feature_view = feature_view
+        self.feature_view = task.feature_view
         self.task = task
         self.nodes: list[DAGNode] = []
 
 
@@ -0,0 +1,12 @@
+import pyarrow as pa
+
+from feast.infra.compute_engines.dag.model import DAGFormat
+from feast.infra.compute_engines.dag.value import DAGValue
+
+
+class ArrowTableValue(DAGValue):
+    def __init__(self, data: pa.Table):
+        super().__init__(data, DAGFormat.ARROW)
+
+    def __repr__(self):
+        return f"ArrowTableValue(schema={self.data.schema}, rows={self.data.num_rows})"
@@ -0,0 +1,79 @@
+from abc import ABC, abstractmethod
+from datetime import timedelta
+
+
+class DataFrameBackend(ABC):
+    """
+    Abstract interface for DataFrame operations used by the LocalComputeEngine.
+
+    This interface defines the contract for implementing pluggable DataFrame backends
+    such as Pandas, Polars, or DuckDB. Each backend must support core table operations
+    such as joins, filtering, aggregation, conversion to/from Arrow, and deduplication.
+
+    The purpose of this abstraction is to allow seamless swapping of execution backends
+    without changing DAGNode or ComputeEngine logic. All nodes operate on pyarrow.Table
+    as the standard input/output format, while the backend defines how the computation
+    is actually performed.
+
+    Expected implementations include:
+    - PandasBackend
+    - PolarsBackend
+    - DuckDBBackend (future)
+
+    Methods
+    -------
+    from_arrow(table: pa.Table) -> Any
+        Convert a pyarrow.Table to the backend-native DataFrame format.
+
+    to_arrow(df: Any) -> pa.Table
+        Convert a backend-native DataFrame to pyarrow.Table.
+
+    join(left: Any, right: Any, on: List[str], how: str) -> Any
+        Join two dataframes on specified keys with given join type.
+
+    groupby_agg(df: Any, group_keys: List[str], agg_ops: Dict[str, Tuple[str, str]]) -> Any
+        Group and aggregate the dataframe. `agg_ops` maps output column names
+        to (aggregation function, source column name) pairs.
+
+    filter(df: Any, expr: str) -> Any
+        Apply a filter expression (string-based) to the DataFrame.
+
+    to_timedelta_value(delta: timedelta) -> Any
+        Convert a Python timedelta object to a backend-compatible value
+        that can be subtracted from a timestamp column.
+
+    drop_duplicates(df: Any, keys: List[str], sort_by: List[str], ascending: bool = False) -> Any
+        Deduplicate the DataFrame by key columns, keeping the first row
+        by descending or ascending sort order.
+
+    rename_columns(df: Any, columns: Dict[str, str]) -> Any
+        Rename columns in the DataFrame according to the provided mapping.
+    """
+
+    @abstractmethod
+    def columns(self, df): ...
+
+    @abstractmethod
+    def from_arrow(self, table): ...
+
+    @abstractmethod
+    def join(self, left, right, on, how): ...
+
+    @abstractmethod
+    def groupby_agg(self, df, group_keys, agg_ops): ...
+
+    @abstractmethod
+    def filter(self, df, expr): ...
+
+    @abstractmethod
+    def to_arrow(self, df): ...
+
+    @abstractmethod
+    def to_timedelta_value(self, delta: timedelta): ...
+
+    @abstractmethod
+    def drop_duplicates(self, df, keys, sort_by, ascending: bool = False):
+        pass
+
+    @abstractmethod
+    def rename_columns(self, df, columns: dict[str, str]): ...