diff --git a/docs/reference/data-sources/iceberg.md b/docs/reference/data-sources/iceberg.md
index 6402a7ec419..3760977bd15 100644
--- a/docs/reference/data-sources/iceberg.md
+++ b/docs/reference/data-sources/iceberg.md
@@ -120,6 +120,62 @@ driver_stats_fv = FeatureView(
 )
 ```
 
+## Local Materialization Sink
+
+Install the Iceberg integration before using an Iceberg table as a sink:
+
+```bash
+pip install "feast[iceberg]"
+```
+
+The local compute engine can materialize a derived Feature View into an
+`IcebergSource`. The catalog namespace must exist, but Feast creates the table
+when it is missing.
+
+```python
+from feast import FeatureView, Field
+from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
+from feast.types import Float64
+
+driver_stats_transformed = FeatureView(
+    name="driver_stats_transformed",
+    entities=[driver],
+    schema=[Field(name="adjusted_conv_rate", dtype=Float64)],
+    source=driver_stats_fv,
+    sink_source=IcebergSource(
+        catalog_type="sql",
+        catalog_name="local",
+        catalog_properties={"uri": "sqlite:////tmp/iceberg_catalog.db"},
+        warehouse="file:///tmp/iceberg_warehouse",
+        namespace="features",
+        table="driver_stats_transformed",
+        timestamp_field="event_timestamp",
+    ),
+    online=False,
+    offline=False,
+)
+```
+
+With `compute_engine.type: local` in `feature_store.yaml`, materialize the view
+normally:
+
+```bash
+feast materialize 2026-08-15T00:00:00 2026-08-16T00:00:00
+```
+
+PyIceberg upserts each computed Arrow batch. The upsert key is the mapped entity
+join-key columns plus the mapped event-timestamp column. Entityless views use
+the event timestamp alone. Incoming null or duplicate keys are rejected.
+
+Repeated materialization of identical data is idempotent. An existing target
+table must have exactly the same column names and compatible types; Feast does
+not evolve Iceberg schemas automatically.
+
+This sink is currently supported only by the local compute engine. Online-store,
+offline-store, and Iceberg writes are sequential and independent, not a single
+transaction. Spark support will use a distributed Iceberg `MERGE INTO` path in
+a separate change.
+
 ## Configuration Reference
 
 ### IcebergSource
diff --git a/docs/superpowers/plans/2026-08-16-iceberg-local-materialization-sink.md b/docs/superpowers/plans/2026-08-16-iceberg-local-materialization-sink.md
new file mode 100644
index 00000000000..3bbbee5a823
--- /dev/null
+++ b/docs/superpowers/plans/2026-08-16-iceberg-local-materialization-sink.md
@@ -0,0 +1,474 @@
+# Iceberg Local Materialization Sink Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Let the local compute engine materialize a derived FeatureView into an `IcebergSource` sink with deterministic PyIceberg upsert semantics.
+
+**Architecture:** Keep sink selection in `LocalOutputNode`, but put all catalog, table-creation, schema-validation, duplicate-key, and upsert behavior behind a focused `IcebergSource.write_materialized_table()` method. `LocalFeatureBuilder` passes resolved `ColumnInfo` to the output node so join keys and the event timestamp use the same mapped names as the computed Arrow table. Existing online/offline writes remain independent and unchanged.
+
+**Tech Stack:** Python, PyArrow, PyIceberg 0.10+, Feast local compute DAG, pytest, ruff, mypy.
+
+**Spec:** `docs/superpowers/specs/2026-08-16-iceberg-local-materialization-sink-design.md`
+
+## Global Constraints
+
+- Limit this PR to the local compute engine. Spark `MERGE INTO` is a follow-up.
+- Reuse the existing `sink_source` field with an `IcebergSource`; do not add a new public sink protocol or protobuf fields.
+- Use mapped entity join keys plus the mapped event timestamp as the upsert key.
+- Reject duplicate keys in each incoming batch before any catalog mutation.
+- Create a missing table, but require its namespace to exist.
+- Require exact column names and compatible Arrow/Iceberg types for an existing table; do not evolve its schema.
+- Preserve the lightweight REST client for existing read/validation paths. Use PyIceberg for writes for every catalog type, including REST.
+- Do not retry commit conflicts and do not imply transactional consistency with online/offline writes.
+
+## File Map
+
+- Modify `pyproject.toml`: raise the optional Iceberg dependency floor to PyIceberg 0.10.
+- Modify `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py`: construct a PyIceberg catalog for writes and own validation/create/upsert behavior.
+- Modify `sdk/python/feast/infra/compute_engines/local/nodes.py`: detect an Iceberg sink and invoke it with resolved keys.
+- Modify `sdk/python/feast/infra/compute_engines/local/feature_builder.py`: pass `ColumnInfo` into `LocalOutputNode`.
+- Modify `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py`: unit-test PyIceberg configuration and the writer contract.
+- Modify `sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py`: unit-test local routing, mapped keys, empty input, and independent writes.
+- Create `sdk/python/tests/component/iceberg/__init__.py`: component-test package marker.
+- Create `sdk/python/tests/component/iceberg/test_local_materialization_sink.py`: exercise a real SQL catalog and filesystem warehouse.
+- Modify `docs/reference/data-sources/iceberg.md`: document local sink configuration, guarantees, and limitations.
+
+---
+
+### Task 1: Establish the PyIceberg 0.10 writer contract
+
+**Files:**
+- Modify: `pyproject.toml`
+- Test: `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py`
+- Modify: `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py`
+
+- [ ] **Step 1: Raise the optional dependency floor**
+
+Change the extra to:
+
+```toml
+iceberg = ["pyiceberg>=0.10.0"]
+```
+
+Do not regenerate the default, minimal, or CI requirements files: `lock-python-dependencies-all` compiles extras `ci`, `minimal`, and `minimal-sdist-build`, none of which select `iceberg`.
+
+- [ ] **Step 2: Write failing tests for a dedicated PyIceberg catalog loader**
+
+Add parameterized tests proving `get_pyiceberg_catalog()` calls `pyiceberg.catalog.load_catalog()` for both REST and non-REST sources. Assert that it passes `type`, `uri`, `warehouse`, token from `token_env_var`, and `catalog_properties` without changing `get_catalog_client()` behavior.
+
+```python
+@patch("pyiceberg.catalog.load_catalog")
+def test_get_pyiceberg_catalog_for_rest(mock_load_catalog):
+    source = IcebergSource(
+        catalog_type="rest",
+        endpoint="http://catalog.test",
+        warehouse="warehouse",
+        namespace="features",
+        table="driver_stats",
+        token_env_var="ICEBERG_TOKEN",
+        catalog_properties={"prefix": "tenant"},
+    )
+    with patch.dict("os.environ", {"ICEBERG_TOKEN": "secret"}):
+        source.get_pyiceberg_catalog()
+    mock_load_catalog.assert_called_once_with(
+        "feast_iceberg",
+        type="rest",
+        prefix="tenant",
+        uri="http://catalog.test",
+        warehouse="warehouse",
+        token="secret",
+    )
+```
+
+- [ ] **Step 3: Run the focused tests and confirm the expected failure**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py -k pyiceberg_catalog
+```
+
+Expected: fail because `IcebergSource.get_pyiceberg_catalog` does not exist.
+
+- [ ] **Step 4: Implement `get_pyiceberg_catalog()` and centralize configuration**
+
+Add a private `_pyiceberg_catalog_config() -> Dict[str, str]` helper and:
+
+```python
+def get_pyiceberg_catalog(self) -> Any:
+    """Load a PyIceberg catalog for mutation-capable operations."""
+    try:
+        from pyiceberg.catalog import load_catalog
+    except ImportError as exc:
+        raise ImportError(
+            "Iceberg materialization requires PyIceberg; install feast[iceberg]."
+        ) from exc
+    return load_catalog(self.catalog_name, **self._pyiceberg_catalog_config())
+```
+
+Have the non-REST branch of `get_catalog_client()` reuse this method. Leave its REST branch on `IcebergRestClient`.
+
+- [ ] **Step 5: Run focused tests and static checks**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+uv run ruff check sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+uv run ruff format --check sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+```
+
+Expected: all pass.
+
+- [ ] **Step 6: Commit the dependency and catalog boundary**
+
+```bash
+git add pyproject.toml sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+git commit -s -m "feat: Add PyIceberg write catalog support"
+```
+
+---
+
+### Task 2: Implement strict create-or-upsert semantics on `IcebergSource`
+
+**Files:**
+- Modify: `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py`
+- Test: `sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py`
+
+- [ ] **Step 1: Write failing unit tests for pre-mutation validation**
+
+Cover these cases with mocked catalogs/tables:
+
+- a missing join-key column raises `ValueError` naming the missing key;
+- a null value in any key column raises `ValueError`;
+- duplicate composite keys raise `ValueError` before `load_table`, `create_table`, or `upsert`;
+- an existing table with missing, unexpected, or incompatible columns raises a single actionable `ValueError` describing every mismatch;
+- an exact existing schema calls `table.upsert(incoming, join_cols=join_cols)`;
+- a missing table calls `create_table(identifier, schema=incoming.schema)` and then upserts;
+- a missing namespace propagates as an error and never calls `create_namespace`;
+- entityless input succeeds when the timestamp is the sole key.
+
+Use the public contract:
+
+```python
+source.write_materialized_table(
+    incoming,
+    join_cols=["driver_id", "event_timestamp"],
+)
+```
+
+- [ ] **Step 2: Run the writer tests and confirm the expected failure**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py -k materialized_table
+```
+
+Expected: fail because `write_materialized_table()` does not exist.
+
+- [ ] **Step 3: Implement key validation without pandas conversion**
+
+Add private helpers that operate on `pyarrow.Table`:
+
+```python
+def _validate_upsert_keys(table: pa.Table, join_cols: list[str]) -> None:
+    missing = sorted(set(join_cols) - set(table.column_names))
+    if missing:
+        raise ValueError(f"Iceberg upsert key columns are missing: {missing}")
+    # Reject null keys and duplicate StructArray rows before catalog access.
+```
+
+Use Arrow compute or a struct of the key arrays to find duplicates. Include the key names and duplicate count in the error; do not log row values because entity keys may be sensitive.
+
+- [ ] **Step 4: Implement strict existing-table schema validation**
+
+Convert the target PyIceberg schema to Arrow via PyIceberg's Arrow schema conversion utilities. Compare sets of column names, then compare each shared field with nullability and type normalization limited to representations PyIceberg accepts losslessly. The error must separately list `missing`, `unexpected`, and `incompatible` entries.
+
+Do not add, rename, widen, or reorder target columns. Reordering the incoming table to the target schema order before `upsert` is allowed after validation.
+
+- [ ] **Step 5: Implement create-or-load and upsert**
+
+Implement:
+
+```python
+def write_materialized_table(
+    self,
+    table: pa.Table,
+    join_cols: list[str],
+) -> None:
+```
+
+Behavior:
+
+1. Validate non-empty `join_cols`, key presence, nulls, and duplicates.
+2. Load `f"{namespace}.{iceberg_table}"` using `get_pyiceberg_catalog()`.
+3. On PyIceberg `NoSuchTableError`, call `catalog.create_table(identifier, schema=table.schema)`; do not catch `NoSuchNamespaceError` and do not create a namespace.
+4. For an existing table, validate the incoming schema strictly and reorder columns to target order.
+5. Call `iceberg_table.upsert(table, join_cols=join_cols)`, the PyIceberg 0.10 API. Do not overload catalog properties as snapshot properties; snapshot metadata is optional in the design and is omitted until PyIceberg exposes a stable per-upsert metadata API.
+6. Let commit-conflict and catalog errors propagate with the table identifier in a wrapping message; do not retry.
+
+- [ ] **Step 6: Run tests and type/lint checks**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py -k 'materialized_table or pyiceberg_catalog'
+uv run ruff check sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+uv run bash -c "cd sdk/python && mypy feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py"
+```
+
+Expected: all pass.
+
+- [ ] **Step 7: Commit the Iceberg writer**
+
+```bash
+git add sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+git commit -s -m "feat: Add Iceberg materialization upserts"
+```
+
+---
+
+### Task 3: Route local derived-view output to Iceberg
+
+**Files:**
+- Modify: `sdk/python/feast/infra/compute_engines/local/nodes.py`
+- Modify: `sdk/python/feast/infra/compute_engines/local/feature_builder.py`
+- Test: `sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py`
+
+- [ ] **Step 1: Write failing output-node tests**
+
+Create real `ColumnInfo` values and mocked feature views. Cover:
+
+- `LocalOutputNode` calls an Iceberg sink once with mapped `join_keys_columns` plus `timestamp_column`;
+- entityless views pass only `timestamp_column`;
+- absent timestamp columns fail before invoking the sink;
+- non-Iceberg sinks are ignored by this PR and existing online/offline behavior is unchanged;
+- zero-row input performs no online, offline, or Iceberg write;
+- when online and offline are also enabled, all three independent writes occur;
+- Iceberg failure propagates rather than being swallowed.
+
+Avoid unconstrained `MagicMock` attribute behavior by explicitly setting `feature_view.source_views` and `feature_view.sink_source`.
+
+- [ ] **Step 2: Run the focused node tests and confirm failure**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py -k iceberg
+```
+
+Expected: fail because `LocalOutputNode` does not accept `ColumnInfo` or write an Iceberg sink.
+
+- [ ] **Step 3: Pass resolved `ColumnInfo` from the builder**
+
+Change the constructor to require `column_info: ColumnInfo` and build the node with:
+
+```python
+column_info = self.get_column_info(view)
+node = LocalOutputNode(
+    "output",
+    self.dag_root.view,
+    column_info,
+    inputs=[input_node],
+)
+```
+
+Update every direct unit-test construction of `LocalOutputNode` to supply a realistic `ColumnInfo`; do not make it optional merely to preserve old tests.
+
+- [ ] **Step 4: Add narrowly scoped Iceberg sink routing**
+
+After the empty-table early return and independently of online/offline flags:
+
+```python
+sink_source = getattr(self.feature_view, "sink_source", None)
+if self.feature_view.source_views and isinstance(sink_source, IcebergSource):
+    join_cols = [
+        *self.column_info.join_keys_columns,
+        self.column_info.timestamp_column,
+    ]
+    sink_source.write_materialized_table(
+        input_table,
+        join_cols=join_cols,
+    )
+```
+
+Deduplicate `join_cols` while preserving order. Validate that `timestamp_column` is non-empty and present. Keep existing online and offline writes intact; the chosen order must be documented in code and tests as sequential but non-transactional.
+
+- [ ] **Step 5: Run local node and builder regression tests**
+
+Run:
+
+```bash
+uv run pytest -q sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
+uv run pytest -q sdk/python/tests/unit/infra/compute_engines/test_local_compute_engine.py sdk/python/tests/unit/infra/compute_engines/test_local_job.py
+uv run ruff check sdk/python/feast/infra/compute_engines/local/nodes.py sdk/python/feast/infra/compute_engines/local/feature_builder.py sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
+uv run bash -c "cd sdk/python && mypy feast/infra/compute_engines/local/nodes.py feast/infra/compute_engines/local/feature_builder.py"
+```
+
+Expected: all pass.
+
+- [ ] **Step 6: Commit local routing**
+
+```bash
+git add sdk/python/feast/infra/compute_engines/local/nodes.py sdk/python/feast/infra/compute_engines/local/feature_builder.py sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
+git commit -s -m "feat: Write local materializations to Iceberg"
+```
+
+---
+
+### Task 4: Prove idempotent upserts with a real local catalog
+
+**Files:**
+- Create: `sdk/python/tests/component/iceberg/__init__.py`
+- Create: `sdk/python/tests/component/iceberg/test_local_materialization_sink.py`
+
+- [ ] **Step 1: Write the component test**
+
+Use `tmp_path` for both a SQLite catalog database and a filesystem warehouse. Load a PyIceberg SQL catalog, create the namespace explicitly, and configure `IcebergSource` with `catalog_type="sql"`, a SQLite `uri` in `catalog_properties`, the filesystem warehouse URI, namespace `features`, and table `driver_stats`.
+
+The test must:
+
+1. write two distinct composite keys;
+2. write the identical batch again and assert the table still has two rows;
+3. write one existing key with a changed feature value and assert that row is updated, not appended;
+4. assert the table schema and key columns remain unchanged;
+5. query through `catalog.load_table("features.driver_stats").scan().to_arrow()` rather than inspecting implementation mocks.
+
+- [ ] **Step 2: Run the component test and address only environment-real issues**
+
+Run:
+
+```bash
+uv run --extra iceberg pytest -q sdk/python/tests/component/iceberg/test_local_materialization_sink.py
+```
+
+Expected: pass with no external services. If SQL catalog support requires a declared PyIceberg extra in 0.10, add that extra to Feast's `iceberg` dependency and document why in `pyproject.toml`; do not replace the test with mocks.
+
+- [ ] **Step 3: Run Iceberg regression tests**
+
+Run:
+
+```bash
+uv run --extra iceberg pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests sdk/python/tests/component/iceberg
+```
+
+Expected: all pass.
+
+- [ ] **Step 4: Commit the component coverage**
+
+```bash
+git add sdk/python/tests/component/iceberg
+git commit -s -m "test: Cover local Iceberg materialization upserts"
+```
+
+---
+
+### Task 5: Document the user workflow and limitations
+
+**Files:**
+- Modify: `docs/reference/data-sources/iceberg.md`
+
+- [ ] **Step 1: Add a local materialization-sink example**
+
+Show a derived FeatureView that uses an existing source view and:
+
+```python
+sink_source=IcebergSource(
+    catalog_type="sql",
+    catalog_properties={"uri": "sqlite:////tmp/iceberg_catalog.db"},
+    warehouse="file:///tmp/iceberg_warehouse",
+    namespace="features",
+    table="driver_stats_transformed",
+    timestamp_field="event_timestamp",
+)
+```
+
+Show the installation command `pip install "feast[iceberg]"` and a local-engine materialization command.
+
+- [ ] **Step 2: Document exact semantics**
+
+State that:
+
+- only the local compute engine supports this sink in this release;
+- PyIceberg performs the write;
+- the namespace must already exist, while the table may be created;
+- keys are mapped entity join keys plus mapped event timestamp;
+- duplicate or null incoming keys are rejected;
+- existing schemas must match and are never evolved automatically;
+- repeated identical materializations are idempotent;
+- online, offline, and Iceberg writes are independent and not transactional;
+- Spark support will use distributed Iceberg `MERGE INTO` separately.
+
+- [ ] **Step 3: Check documentation formatting and links**
+
+Run:
+
+```bash
+uv run pre-commit run --files docs/reference/data-sources/iceberg.md
+```
+
+Expected: all configured documentation hooks pass.
+
+- [ ] **Step 4: Commit documentation**
+
+```bash
+git add docs/reference/data-sources/iceberg.md
+git commit -s -m "docs: Document local Iceberg materialization sinks"
+```
+
+---
+
+### Task 6: Final verification and PR readiness
+
+**Files:**
+- Verify all files listed above.
+
+- [ ] **Step 1: Run the complete focused test matrix**
+
+Run:
+
+```bash
+uv run --extra iceberg pytest -q sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests sdk/python/tests/unit/infra/compute_engines/local sdk/python/tests/unit/infra/compute_engines/test_local_compute_engine.py sdk/python/tests/unit/infra/compute_engines/test_local_job.py sdk/python/tests/component/iceberg
+```
+
+Expected: all pass.
+
+- [ ] **Step 2: Run formatting, lint, and targeted typing**
+
+Run:
+
+```bash
+uv run ruff format --check sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/compute_engines/local/nodes.py sdk/python/feast/infra/compute_engines/local/feature_builder.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py sdk/python/tests/component/iceberg/test_local_materialization_sink.py
+uv run ruff check sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/compute_engines/local/nodes.py sdk/python/feast/infra/compute_engines/local/feature_builder.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py sdk/python/tests/component/iceberg/test_local_materialization_sink.py
+uv run bash -c "cd sdk/python && mypy feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py feast/infra/compute_engines/local/nodes.py feast/infra/compute_engines/local/feature_builder.py"
+```
+
+Expected: all pass.
+
+- [ ] **Step 3: Run repository hooks on the changed files**
+
+Run:
+
+```bash
+uv run pre-commit run --files pyproject.toml docs/reference/data-sources/iceberg.md sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py sdk/python/feast/infra/compute_engines/local/nodes.py sdk/python/feast/infra/compute_engines/local/feature_builder.py sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py sdk/python/tests/component/iceberg/__init__.py sdk/python/tests/component/iceberg/test_local_materialization_sink.py
+```
+
+Expected: all hooks pass. If a hook rewrites a file, inspect the diff and rerun until clean.
+
+- [ ] **Step 4: Audit scope and behavior**
+
+Run:
+
+```bash
+git diff --check
+git status --short
+git diff --stat HEAD~5
+git log --oneline --decorate -6
+```
+
+Confirm there are no Spark changes, protobuf changes, generic sink abstractions, generated dependency lock churn, namespace creation, schema evolution, retry loops, or unrelated edits.
+
+- [ ] **Step 5: Request code review before publishing**
+
+Use `superpowers:requesting-code-review` to compare the implementation with the design spec and this plan. Resolve correctness findings, rerun the affected checks, and only then prepare the branch for publication.
diff --git a/docs/superpowers/specs/2026-08-16-iceberg-local-materialization-sink-design.md b/docs/superpowers/specs/2026-08-16-iceberg-local-materialization-sink-design.md
new file mode 100644
index 00000000000..49cdb751cc5
--- /dev/null
+++ b/docs/superpowers/specs/2026-08-16-iceberg-local-materialization-sink-design.md
@@ -0,0 +1,244 @@
+# Local Iceberg Materialization Sink Design
+
+## Summary
+
+Extend Feast's existing `sink_source` model so the local compute engine can
+persist derived FeatureView materialization results to Apache Iceberg. The
+first phase uses PyIceberg to create a missing table and idempotently upsert a
+PyArrow result into an existing table.
+
+This design deliberately reuses `sink_source=IcebergSource(...)`. It does not
+add another FeatureView field, make Iceberg the repository's global offline
+store, or add Spark writes. A follow-up can implement the same behavior in the
+Spark compute engine with Iceberg `MERGE INTO`.
+
+## Goals
+
+- Support `IcebergSource` as the `sink_source` of a derived FeatureView during
+  local materialization.
+- Preserve Feast materialization idempotency by upserting on entity join keys
+  plus the event timestamp.
+- Support the catalog configurations already represented by `IcebergSource`,
+  including REST, Glue, Hive, SQL, and DynamoDB catalogs.
+- Create the destination Iceberg table when it does not exist.
+- Keep catalog, authentication, table, schema, and write behavior isolated
+  behind the Iceberg source implementation.
+- Produce actionable errors for invalid configuration, incompatible schemas,
+  missing keys, duplicate input keys, authentication failures, and commit
+  failures.
+
+## Non-goals
+
+- Spark, Ray, or other compute-engine writes.
+- A generic writable-data-source interface.
+- Automatic Iceberg namespace creation.
+- Automatic schema, partition-spec, or sort-order evolution.
+- Cross-store transactional atomicity among Iceberg, online stores, and the
+  configured offline store.
+- Writing a base FeatureView to an additional arbitrary sink. In this phase,
+  the existing `sink_source` semantics remain limited to derived views.
+- Snapshot expiration, compaction, branching, tagging, or other Iceberg table
+  maintenance.
+
+## User API
+
+The public API is the existing derived-view API:
+
+```python
+from feast import BatchFeatureView
+from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
+
+driver_stats_iceberg = IcebergSource(
+    catalog_type="rest",
+    endpoint="https://catalog.example.com/iceberg",
+    warehouse="production",
+    namespace="features",
+    table="daily_driver_stats",
+    token_env_var="ICEBERG_TOKEN",
+    timestamp_field="event_timestamp",
+)
+
+daily_driver_stats = BatchFeatureView(
+    name="daily_driver_stats",
+    source=hourly_driver_stats,
+    sink_source=driver_stats_iceberg,
+    entities=[driver],
+    schema=[...],
+    udf=build_daily_features,
+    online=True,
+)
+```
+
+The derived view continues to serialize its sink as its effective batch source,
+so no FeatureView or DataSource protobuf change is required.
+
+## Architecture
+
+### Iceberg writer boundary
+
+`IcebergSource` will own a focused materialization method that accepts:
+
+- the PyArrow table produced by the local DAG;
+- resolved upsert key column names;
+- optional snapshot metadata such as Feast project and FeatureView name.
+
+The method is responsible for constructing a PyIceberg catalog, resolving the
+table identifier, creating or loading the table, validating the write, and
+performing the upsert. Keeping this behavior with `IcebergSource` prevents
+catalog-specific details from leaking into compute-engine nodes.
+
+The current lightweight Feast REST client remains unchanged for reads and
+governance operations. Sink writes construct a PyIceberg catalog for every
+supported catalog type, including REST. Catalog construction reuses
+`catalog_name`, `endpoint`, `warehouse`, `catalog_properties`, and the token
+resolved from `token_env_var`.
+
+### Local output integration
+
+`LocalFeatureBuilder` will pass its resolved `ColumnInfo` to `LocalOutputNode`.
+This avoids reconstructing field mappings in the output node and guarantees
+that upsert keys refer to the post-mapping Arrow column names.
+
+`LocalOutputNode.execute()` retains its existing empty-table short circuit and
+online/offline writes. For a derived view whose effective batch source is an
+`IcebergSource`, it additionally invokes the Iceberg materialization method.
+
+The Iceberg write is part of the materialization job. An exception propagates
+through the existing local compute engine and results in an errored
+materialization job.
+
+### Dependency
+
+The `iceberg` optional dependency changes from `pyiceberg>=0.7.0` to
+`pyiceberg>=0.10.0`. PyIceberg 0.10 introduced the native Arrow-based
+`Table.upsert()` API required for the idempotency guarantee.
+
+Users without the `iceberg` extra remain unaffected. Attempting to use an
+Iceberg sink without PyIceberg installed raises an installation-oriented error
+that names `feast[iceberg]`.
+
+## Data Flow
+
+1. The local compute engine resolves and executes the derived FeatureView DAG.
+2. The final node produces a PyArrow table with source field mappings already
+   applied.
+3. `LocalOutputNode` performs the configured online and offline store writes.
+4. If the derived view has an Iceberg sink, the node resolves upsert keys from
+   `ColumnInfo`: all mapped entity join keys followed by the mapped event
+   timestamp column.
+5. `IcebergSource` constructs a PyIceberg catalog and resolves
+   `<namespace>.<table>`.
+6. If the table does not exist, `IcebergSource` creates it from the Arrow
+   schema. The namespace must already exist.
+7. The source validates schema compatibility and key presence.
+8. `Table.upsert(arrow_table, join_cols=keys)` atomically commits the Iceberg
+   snapshot.
+9. The node returns its existing `ArrowTableValue` output.
+
+## Upsert Semantics
+
+The identity of a materialized record is:
+
+```text
+mapped entity join keys + mapped event timestamp
+```
+
+Including the event timestamp preserves multiple historical values for an
+entity while making a repeated materialization of the same interval
+idempotent. The created-timestamp column is not part of the key; a later write
+for the same entity/event timestamp replaces the prior row.
+
+Entityless views use the event timestamp as their key. If no usable event
+timestamp column is present, materialization fails before any Iceberg commit.
+
+The input batch must contain at most one row for each composite key. Duplicate
+keys are rejected rather than relying on unspecified merge ordering.
+
+## Table Creation and Schema Rules
+
+When the destination table is absent:
+
+- the containing namespace must already exist;
+- the table is created from the materialized Arrow schema;
+- Feast passes the explicit join columns to each upsert, so the initial phase
+  does not require changing Iceberg identifier-field metadata;
+- table properties from `IcebergSource.catalog_properties` remain catalog
+  configuration and are not silently copied into table properties.
+
+When the table exists, the materialized columns and types must be compatible
+with its schema. The first phase does not add, drop, rename, or widen columns.
+The error identifies missing, unexpected, and incompatible columns.
+
+## Failure Behavior
+
+- Missing `pyiceberg`: raise an error instructing the user to install
+  `feast[iceberg]`.
+- Missing namespace: fail without creating governance boundaries implicitly.
+- Authentication or credential-vending failure: preserve the underlying cause
+  and identify the catalog endpoint and target table without exposing secrets.
+- Missing upsert columns or duplicate input keys: fail before catalog mutation.
+- Incompatible existing schema: fail before the upsert.
+- Concurrent Iceberg commit conflict: surface the PyIceberg commit error; the
+  first phase does not implement implicit retries.
+- Empty Arrow table: perform no catalog call or write.
+
+Feast cannot atomically commit Iceberg together with its online and offline
+stores. A failure after another store has accepted data is reported as a failed
+materialization job, and rerunning is safe for the Iceberg sink because the
+write is an upsert.
+
+## Testing
+
+### Unit tests
+
+- Catalog configuration for REST, Glue, Hive, SQL, and DynamoDB sources.
+- Helpful missing-dependency error.
+- Table creation with the materialized Arrow schema.
+- Existing-table load and `upsert()` invocation.
+- Composite join keys and entityless event-timestamp keys.
+- Field-mapped entity and timestamp columns.
+- Empty input as a no-op.
+- Missing key, duplicate key, namespace, authentication, schema, and commit
+  failures.
+- Verification that secrets resolved from environment variables do not appear
+  in serialized source configuration or error messages.
+- Existing IcebergSource protobuf round-trip when used as a derived view sink.
+
+### Component test
+
+Use a temporary filesystem warehouse and PyIceberg SQL catalog with the local
+compute engine:
+
+1. Materialize a derived FeatureView into a new Iceberg table.
+2. Verify the expected rows and schema.
+3. Materialize the same interval again.
+4. Verify the row count is unchanged.
+5. Change a non-key feature value, materialize again, and verify the row is
+   updated rather than duplicated.
+
+### Regression tests
+
+- Local online-only and offline-only materialization without an Iceberg sink is
+  unchanged.
+- A derived view with a non-Iceberg sink retains its current behavior.
+
+## Documentation
+
+Extend the Iceberg data-source documentation with:
+
+- a derived FeatureView `sink_source` example;
+- the `feast[iceberg]` installation requirement;
+- supported catalog types;
+- composite upsert-key behavior;
+- strict schema behavior;
+- the phase-one local-engine limitation;
+- a note that Spark support is planned separately.
+
+## Follow-up: Spark
+
+A separate PR will reuse the same `sink_source=IcebergSource(...)` contract in
+`SparkWriteNode`. It will configure the Spark Iceberg catalog, create a missing
+table, register the materialized DataFrame as a temporary view, and execute a
+distributed Iceberg `MERGE INTO` on the same entity-plus-event-timestamp key.
+It will not collect Spark data into the driver or route writes through
+PyIceberg.
diff --git a/pyproject.toml b/pyproject.toml
index 4c668a89847..7239afb4208 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -114,7 +114,7 @@ mysql = ["pymysql", "types-PyMySQL"]
 openlineage = ["openlineage-python>=1.40.0"]
 opentelemetry = ["prometheus_client", "psutil"]
 spark = ["pyspark>=4.0.0"]
-iceberg = ["pyiceberg>=0.7.0"]
+iceberg = ["pyiceberg[hive,glue,dynamodb,sql-sqlite]>=0.10.0"]
 trino = ["trino>=0.305.0,<0.400.0", "regex"]
 postgres = ["psycopg[binary,pool]>=3.2.5"]
 # psycopg[c] install requires a system with a C compiler, python dev headers, & postgresql client dev headers
diff --git a/sdk/python/feast/infra/compute_engines/local/feature_builder.py b/sdk/python/feast/infra/compute_engines/local/feature_builder.py
index 754a00db76f..6f5120a215a 100644
--- a/sdk/python/feast/infra/compute_engines/local/feature_builder.py
+++ b/sdk/python/feast/infra/compute_engines/local/feature_builder.py
@@ -129,6 +129,9 @@ def build_validation_node(self, view, input_node):
         return node
 
     def build_output_nodes(self, view, input_node):
-        node = LocalOutputNode("output", self.dag_root.view, inputs=[input_node])
+        column_info = self.get_column_info(view)
+        node = LocalOutputNode(
+            "output", self.dag_root.view, column_info, inputs=[input_node]
+        )
         self.nodes.append(node)
         return node
diff --git a/sdk/python/feast/infra/compute_engines/local/nodes.py b/sdk/python/feast/infra/compute_engines/local/nodes.py
index 9d3e1a48881..2d798587a52 100644
--- a/sdk/python/feast/infra/compute_engines/local/nodes.py
+++ b/sdk/python/feast/infra/compute_engines/local/nodes.py
@@ -18,6 +18,7 @@
     create_offline_store_retrieval_job,
     infer_entity_timestamp_column,
 )
+from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
 from feast.utils import _convert_arrow_to_proto
 
 logger = logging.getLogger(__name__)
@@ -357,10 +358,12 @@ def __init__(
         self,
         name: str,
         feature_view: Union[BatchFeatureView, StreamFeatureView],
+        column_info: ColumnInfo,
         inputs=None,
     ):
         super().__init__(name, inputs=inputs)
         self.feature_view = feature_view
+        self.column_info = column_info
 
     def execute(self, context: ExecutionContext) -> ArrowTableValue:
         input_table = self.get_single_table(context).data
@@ -407,4 +410,20 @@ def execute(self, context: ExecutionContext) -> ArrowTableValue:
                 progress=lambda x: None,
             )
 
+        sink_source = getattr(self.feature_view, "batch_source", None)
+        if self.feature_view.source_views and isinstance(sink_source, IcebergSource):
+            timestamp_column = self.column_info.timestamp_column
+            if timestamp_column not in input_table.column_names:
+                raise ValueError(
+                    "Iceberg materialization timestamp column is missing: "
+                    f"{timestamp_column}"
+                )
+            join_cols = list(
+                dict.fromkeys([*self.column_info.join_keys_columns, timestamp_column])
+            )
+            sink_source.write_materialized_table(
+                input_table,
+                join_cols=join_cols,
+            )
+
         return output
diff --git a/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py b/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py
index 757033865f9..8ceed582e9c 100644
--- a/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py
+++ b/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/iceberg_source.py
@@ -15,6 +15,9 @@
 import json
 import os
 from typing import Any, Callable, Dict, Iterable, Optional, Tuple
+from urllib.parse import urlsplit, urlunsplit
+
+import pyarrow as pa
 
 from feast.data_source import DataSource
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
@@ -109,8 +112,9 @@ def get_catalog_client(self):
                 credential_vending=self.credential_vending,
             )
 
-        from pyiceberg.catalog import load_catalog
+        return self.get_pyiceberg_catalog()
 
+    def _pyiceberg_catalog_config(self) -> Dict[str, str]:
         config = {
             "type": self.catalog_type,
             **self.catalog_properties,
@@ -123,7 +127,67 @@ def get_catalog_client(self):
             token = os.environ.get(self.token_env_var, "")
             if token:
                 config.setdefault("token", token)
-        return load_catalog(self.catalog_name, **config)
+        return config
+
+    def get_pyiceberg_catalog(self) -> Any:
+        """Load a PyIceberg catalog for mutation-capable operations."""
+        try:
+            from pyiceberg.catalog import load_catalog
+        except ImportError as exc:
+            raise ImportError(
+                "Iceberg materialization requires PyIceberg; install feast[iceberg]."
+            ) from exc
+        return load_catalog(self.catalog_name, **self._pyiceberg_catalog_config())
+
+    def write_materialized_table(
+        self,
+        table: pa.Table,
+        join_cols: list[str],
+    ) -> None:
+        """Create or upsert a materialized Arrow table through PyIceberg."""
+        _validate_upsert_keys(table, join_cols)
+
+        identifier = f"{self.namespace}.{self.iceberg_table}"
+        location = _sanitize_catalog_location(self.endpoint)
+        try:
+            catalog = self.get_pyiceberg_catalog()
+        except ImportError:
+            raise
+        except Exception as exc:
+            raise RuntimeError(
+                f"Could not access Iceberg catalog at '{location}' for table "
+                f"'{identifier}': {_sanitize_catalog_error(exc, self)}"
+            ) from exc
+
+        from pyiceberg.exceptions import NoSuchNamespaceError, NoSuchTableError
+
+        try:
+            iceberg_table = catalog.load_table(identifier)
+        except NoSuchTableError:
+            try:
+                iceberg_table = catalog.create_table(identifier, schema=table.schema)
+            except NoSuchNamespaceError:
+                raise
+            except Exception as exc:
+                raise RuntimeError(
+                    f"Could not create Iceberg table '{identifier}' through catalog "
+                    f"at '{location}': {_sanitize_catalog_error(exc, self)}"
+                ) from exc
+        except Exception as exc:
+            raise RuntimeError(
+                f"Could not load Iceberg table '{identifier}' through catalog at "
+                f"'{location}': {_sanitize_catalog_error(exc, self)}"
+            ) from exc
+        else:
+            table = _validate_and_reorder_schema(table, iceberg_table.schema())
+
+        try:
+            iceberg_table.upsert(table, join_cols=join_cols)
+        except Exception as exc:
+            raise RuntimeError(
+                f"Could not upsert Iceberg table '{identifier}' through catalog at "
+                f"'{location}': {_sanitize_catalog_error(exc, self)}"
+            ) from exc
 
     def source_type(self) -> DataSourceProto.SourceType.ValueType:
         return DataSourceProto.BATCH_ICEBERG
@@ -285,6 +349,92 @@ def __hash__(self):
         )
 
 
+def _validate_upsert_keys(table: pa.Table, join_cols: list[str]) -> None:
+    if not join_cols:
+        raise ValueError("Iceberg upsert requires at least one key column.")
+
+    missing = sorted(set(join_cols) - set(table.column_names))
+    if missing:
+        raise ValueError(f"Iceberg upsert key columns are missing: {missing}")
+
+    null_columns = [name for name in join_cols if table[name].null_count]
+    if null_columns:
+        raise ValueError(f"Iceberg upsert keys contain null values: {null_columns}")
+
+    keys = zip(*(table[name].to_pylist() for name in join_cols))
+    seen = set()
+    duplicate_count = 0
+    for key in keys:
+        if key in seen:
+            duplicate_count += 1
+        else:
+            seen.add(key)
+    if duplicate_count:
+        raise ValueError(
+            f"Iceberg upsert contains {duplicate_count} duplicate key row(s) "
+            f"for columns {join_cols}."
+        )
+
+
+def _sanitize_catalog_location(endpoint: str) -> str:
+    if not endpoint:
+        return "configured catalog"
+    parsed = urlsplit(endpoint)
+    if not parsed.scheme or not parsed.hostname:
+        return endpoint.split("?", 1)[0]
+    netloc = parsed.hostname
+    if parsed.port:
+        netloc = f"{netloc}:{parsed.port}"
+    return urlunsplit((parsed.scheme, netloc, parsed.path, "", ""))
+
+
+def _sanitize_catalog_error(exc: Exception, source: IcebergSource) -> str:
+    message = str(exc)
+    parsed = urlsplit(source.endpoint)
+    secrets = [parsed.password]
+    if source.token_env_var:
+        secrets.append(os.environ.get(source.token_env_var))
+    for secret in secrets:
+        if secret:
+            message = message.replace(secret, "***")
+    return f"{type(exc).__name__}: {message}"
+
+
+def _validate_and_reorder_schema(table: pa.Table, iceberg_schema: Any) -> pa.Table:
+    from pyiceberg.io.pyarrow import schema_to_pyarrow
+
+    target_schema = schema_to_pyarrow(iceberg_schema)
+    incoming_by_name = {field.name: field for field in table.schema}
+    target_by_name = {field.name: field for field in target_schema}
+
+    missing = sorted(set(target_by_name) - set(incoming_by_name))
+    unexpected = sorted(set(incoming_by_name) - set(target_by_name))
+    incompatible = sorted(
+        name
+        for name in set(incoming_by_name) & set(target_by_name)
+        if not _arrow_types_compatible(
+            incoming_by_name[name].type, target_by_name[name].type
+        )
+        or incoming_by_name[name].nullable != target_by_name[name].nullable
+    )
+    if missing or unexpected or incompatible:
+        raise ValueError(
+            "Iceberg materialization schema mismatch: "
+            f"missing={missing}, unexpected={unexpected}, "
+            f"incompatible={incompatible}."
+        )
+
+    return table.select(target_schema.names)
+
+
+def _arrow_types_compatible(incoming: pa.DataType, target: pa.DataType) -> bool:
+    if incoming == target:
+        return True
+    return (pa.types.is_string(incoming) and pa.types.is_large_string(target)) or (
+        pa.types.is_binary(incoming) and pa.types.is_large_binary(target)
+    )
+
+
 def _iceberg_type_to_feast_value_type(iceberg_type: str) -> ValueType:
     """Maps Iceberg data types to Feast ValueTypes."""
     type_map = {
diff --git a/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py b/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
index d728bda4545..dbbbef0ec8a 100644
--- a/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
+++ b/sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/tests/test_iceberg_source.py
@@ -1,10 +1,16 @@
 """Unit tests for IcebergSource, UnityCatalogSource, and IcebergRestClient."""
 
 import json
+import sys
+from datetime import datetime
 from unittest.mock import MagicMock, Mock, patch
 
+import pyarrow as pa
 import pytest
 import requests
+from pyiceberg.exceptions import NoSuchNamespaceError, NoSuchTableError
+from pyiceberg.schema import Schema
+from pyiceberg.types import DoubleType, LongType, NestedField, StringType, TimestampType
 
 from feast.infra.data_sources.contrib.iceberg_catalog.iceberg_rest_client import (
     CatalogAuthError,
@@ -24,6 +30,37 @@
 
 
 class TestIcebergSource:
+    @staticmethod
+    def _materialization_source() -> IcebergSource:
+        return IcebergSource(
+            catalog_type="sql",
+            endpoint="sqlite:////tmp/catalog.db",
+            warehouse="file:///tmp/warehouse",
+            namespace="features",
+            table="driver_stats",
+        )
+
+    @staticmethod
+    def _materialization_table() -> pa.Table:
+        return pa.table(
+            {
+                "driver_id": [1, 2],
+                "event_timestamp": pa.array(
+                    [datetime(2026, 8, 16, 10), datetime(2026, 8, 16, 11)],
+                    type=pa.timestamp("us"),
+                ),
+                "value": [1.0, 2.0],
+            }
+        )
+
+    @staticmethod
+    def _materialization_schema() -> Schema:
+        return Schema(
+            NestedField(1, "driver_id", LongType(), required=False),
+            NestedField(2, "event_timestamp", TimestampType(), required=False),
+            NestedField(3, "value", DoubleType(), required=False),
+        )
+
     def test_basic_creation(self):
         source = IcebergSource(
             endpoint="http://localhost:8080/api/2.1/unity-catalog/iceberg",
@@ -82,6 +119,256 @@ def test_catalog_type_explicit(self):
         assert source.catalog_type == "glue"
         assert source.catalog_properties == {"region_name": "us-east-1"}
 
+    @patch.dict("os.environ", {"ICEBERG_TOKEN": "secret"})
+    def test_get_pyiceberg_catalog_for_rest(self):
+        mock_load_catalog = MagicMock()
+        mock_catalog_module = MagicMock(load_catalog=mock_load_catalog)
+        source = IcebergSource(
+            catalog_type="rest",
+            endpoint="http://catalog.test",
+            warehouse="warehouse",
+            namespace="features",
+            table="driver_stats",
+            token_env_var="ICEBERG_TOKEN",
+            catalog_properties={"prefix": "tenant"},
+        )
+
+        with patch.dict(sys.modules, {"pyiceberg.catalog": mock_catalog_module}):
+            source.get_pyiceberg_catalog()
+
+        mock_load_catalog.assert_called_once_with(
+            "feast_iceberg",
+            type="rest",
+            prefix="tenant",
+            uri="http://catalog.test",
+            warehouse="warehouse",
+            token="secret",
+        )
+
+    def test_get_pyiceberg_catalog_for_non_rest(self):
+        mock_load_catalog = MagicMock()
+        mock_catalog_module = MagicMock(load_catalog=mock_load_catalog)
+        source = IcebergSource(
+            catalog_type="sql",
+            endpoint="sqlite:////tmp/catalog.db",
+            warehouse="file:///tmp/warehouse",
+            namespace="features",
+            table="driver_stats",
+            catalog_name="local",
+            catalog_properties={"echo": "false"},
+        )
+
+        with patch.dict(sys.modules, {"pyiceberg.catalog": mock_catalog_module}):
+            source.get_pyiceberg_catalog()
+
+        mock_load_catalog.assert_called_once_with(
+            "local",
+            type="sql",
+            echo="false",
+            uri="sqlite:////tmp/catalog.db",
+            warehouse="file:///tmp/warehouse",
+        )
+
+    def test_get_pyiceberg_catalog_missing_dependency_has_install_guidance(self):
+        source = self._materialization_source()
+
+        with patch.dict(sys.modules, {"pyiceberg": None, "pyiceberg.catalog": None}):
+            with pytest.raises(ImportError, match=r"install feast\[iceberg\]"):
+                source.get_pyiceberg_catalog()
+
+    def test_write_materialized_table_rejects_missing_key_before_catalog_access(self):
+        source = self._materialization_source()
+        source.get_pyiceberg_catalog = MagicMock()
+
+        with pytest.raises(ValueError, match="missing.*unknown_key"):
+            source.write_materialized_table(
+                self._materialization_table(), join_cols=["unknown_key"]
+            )
+
+        source.get_pyiceberg_catalog.assert_not_called()
+
+    def test_write_materialized_table_rejects_null_key_before_catalog_access(self):
+        source = self._materialization_source()
+        source.get_pyiceberg_catalog = MagicMock()
+        incoming = self._materialization_table().set_column(
+            0, "driver_id", pa.array([1, None], type=pa.int64())
+        )
+
+        with pytest.raises(ValueError, match="null.*driver_id"):
+            source.write_materialized_table(incoming, join_cols=["driver_id"])
+
+        source.get_pyiceberg_catalog.assert_not_called()
+
+    def test_write_materialized_table_rejects_duplicate_keys_before_catalog_access(
+        self,
+    ):
+        source = self._materialization_source()
+        source.get_pyiceberg_catalog = MagicMock()
+        incoming = pa.table(
+            {
+                "driver_id": [1, 1],
+                "event_timestamp": pa.array(
+                    [datetime(2026, 8, 16, 10), datetime(2026, 8, 16, 10)],
+                    type=pa.timestamp("us"),
+                ),
+                "value": [1.0, 2.0],
+            }
+        )
+
+        with pytest.raises(ValueError, match="1 duplicate.*driver_id"):
+            source.write_materialized_table(
+                incoming, join_cols=["driver_id", "event_timestamp"]
+            )
+
+        source.get_pyiceberg_catalog.assert_not_called()
+
+    def test_write_materialized_table_upserts_existing_table(self):
+        source = self._materialization_source()
+        incoming = self._materialization_table()
+        iceberg_table = MagicMock()
+        iceberg_table.schema.return_value = self._materialization_schema()
+        catalog = MagicMock()
+        catalog.load_table.return_value = iceberg_table
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        source.write_materialized_table(
+            incoming,
+            join_cols=["driver_id", "event_timestamp"],
+        )
+
+        catalog.load_table.assert_called_once_with("features.driver_stats")
+        catalog.create_table.assert_not_called()
+        iceberg_table.upsert.assert_called_once_with(
+            incoming,
+            join_cols=["driver_id", "event_timestamp"],
+        )
+
+    def test_write_materialized_table_creates_missing_table(self):
+        source = self._materialization_source()
+        incoming = self._materialization_table()
+        iceberg_table = MagicMock()
+        catalog = MagicMock()
+        catalog.load_table.side_effect = NoSuchTableError("missing")
+        catalog.create_table.return_value = iceberg_table
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        source.write_materialized_table(
+            incoming, join_cols=["driver_id", "event_timestamp"]
+        )
+
+        catalog.create_table.assert_called_once_with(
+            "features.driver_stats", schema=incoming.schema
+        )
+        iceberg_table.upsert.assert_called_once_with(
+            incoming,
+            join_cols=["driver_id", "event_timestamp"],
+        )
+
+    def test_write_materialized_table_does_not_create_missing_namespace(self):
+        source = self._materialization_source()
+        catalog = MagicMock()
+        catalog.load_table.side_effect = NoSuchTableError("missing")
+        catalog.create_table.side_effect = NoSuchNamespaceError("missing namespace")
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        with pytest.raises(NoSuchNamespaceError, match="missing namespace"):
+            source.write_materialized_table(
+                self._materialization_table(),
+                join_cols=["driver_id", "event_timestamp"],
+            )
+
+        catalog.create_namespace.assert_not_called()
+
+    def test_write_materialized_table_rejects_schema_mismatch(self):
+        source = self._materialization_source()
+        incoming = self._materialization_table().append_column(
+            "unexpected", pa.array([1, 2])
+        )
+        target_schema = Schema(
+            NestedField(1, "driver_id", LongType(), required=False),
+            NestedField(2, "event_timestamp", TimestampType(), required=False),
+            NestedField(3, "value", StringType(), required=False),
+            NestedField(4, "missing", LongType(), required=False),
+        )
+        iceberg_table = MagicMock()
+        iceberg_table.schema.return_value = target_schema
+        catalog = MagicMock()
+        catalog.load_table.return_value = iceberg_table
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        with pytest.raises(ValueError) as exc_info:
+            source.write_materialized_table(
+                incoming, join_cols=["driver_id", "event_timestamp"]
+            )
+
+        message = str(exc_info.value)
+        assert "missing" in message
+        assert "unexpected" in message
+        assert "incompatible" in message
+        assert "value" in message
+        iceberg_table.upsert.assert_not_called()
+
+    def test_write_materialized_table_supports_timestamp_only_key(self):
+        source = self._materialization_source()
+        incoming = self._materialization_table().drop(["driver_id"])
+        iceberg_table = MagicMock()
+        iceberg_table.schema.return_value = Schema(
+            NestedField(1, "event_timestamp", TimestampType(), required=False),
+            NestedField(2, "value", DoubleType(), required=False),
+        )
+        catalog = MagicMock()
+        catalog.load_table.return_value = iceberg_table
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        source.write_materialized_table(incoming, join_cols=["event_timestamp"])
+
+        iceberg_table.upsert.assert_called_once()
+
+    def test_write_materialized_table_wraps_catalog_error_with_sanitized_context(self):
+        token_value = "sensitive-value"
+        source = IcebergSource(
+            catalog_type="rest",
+            endpoint="https://catalog.test/api",
+            warehouse="warehouse",
+            namespace="features",
+            table="driver_stats",
+            token_env_var="ICEBERG_TEST_TOKEN",
+        )
+        source.get_pyiceberg_catalog = MagicMock(
+            side_effect=RuntimeError(f"unauthorized: {token_value}")
+        )
+
+        with patch.dict("os.environ", {"ICEBERG_TEST_TOKEN": token_value}):
+            with pytest.raises(RuntimeError) as exc_info:
+                source.write_materialized_table(
+                    self._materialization_table(),
+                    join_cols=["driver_id", "event_timestamp"],
+                )
+
+        message = str(exc_info.value)
+        assert "features.driver_stats" in message
+        assert "https://catalog.test/api" in message
+        assert token_value not in message
+
+    def test_write_materialized_table_wraps_commit_error_with_table_context(self):
+        source = self._materialization_source()
+        incoming = self._materialization_table()
+        iceberg_table = MagicMock()
+        iceberg_table.schema.return_value = self._materialization_schema()
+        iceberg_table.upsert.side_effect = RuntimeError("commit conflict")
+        catalog = MagicMock()
+        catalog.load_table.return_value = iceberg_table
+        source.get_pyiceberg_catalog = MagicMock(return_value=catalog)
+
+        with pytest.raises(RuntimeError) as exc_info:
+            source.write_materialized_table(
+                incoming, join_cols=["driver_id", "event_timestamp"]
+            )
+
+        message = str(exc_info.value)
+        assert "features.driver_stats" in message
+        assert "commit conflict" in message
+
     def test_proto_roundtrip(self):
         source = IcebergSource(
             endpoint="http://localhost:8080/iceberg",
diff --git a/sdk/python/tests/component/iceberg/__init__.py b/sdk/python/tests/component/iceberg/__init__.py
new file mode 100644
index 00000000000..2c90b270c37
--- /dev/null
+++ b/sdk/python/tests/component/iceberg/__init__.py
@@ -0,0 +1 @@
+"""Component tests for Apache Iceberg integrations."""
diff --git a/sdk/python/tests/component/iceberg/test_local_materialization_sink.py b/sdk/python/tests/component/iceberg/test_local_materialization_sink.py
new file mode 100644
index 00000000000..132ae5eb53a
--- /dev/null
+++ b/sdk/python/tests/component/iceberg/test_local_materialization_sink.py
@@ -0,0 +1,55 @@
+from datetime import datetime
+
+import pyarrow as pa
+from pyiceberg.catalog import load_catalog
+
+from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
+
+
+def test_local_iceberg_materialization_is_idempotent(tmp_path):
+    catalog_uri = f"sqlite:///{tmp_path / 'catalog.db'}"
+    warehouse_uri = (tmp_path / "warehouse").as_uri()
+    catalog = load_catalog(
+        "test",
+        type="sql",
+        uri=catalog_uri,
+        warehouse=warehouse_uri,
+    )
+    catalog.create_namespace("features")
+    source = IcebergSource(
+        catalog_type="sql",
+        catalog_name="test",
+        catalog_properties={"uri": catalog_uri},
+        warehouse=warehouse_uri,
+        namespace="features",
+        table="driver_stats",
+        timestamp_field="event_timestamp",
+    )
+    first = pa.table(
+        {
+            "driver_id": [1, 2],
+            "event_timestamp": pa.array(
+                [datetime(2026, 8, 16, 10), datetime(2026, 8, 16, 11)],
+                type=pa.timestamp("us"),
+            ),
+            "value": [1.0, 2.0],
+            "label": pa.array(["one", "two"], type=pa.string()),
+        }
+    )
+    join_cols = ["driver_id", "event_timestamp"]
+
+    source.write_materialized_table(first, join_cols=join_cols)
+    source.write_materialized_table(first, join_cols=join_cols)
+
+    after_repeat = catalog.load_table("features.driver_stats").scan().to_arrow()
+    assert after_repeat.num_rows == 2
+
+    updated = first.set_column(2, "value", pa.array([10.0, 2.0]))
+    source.write_materialized_table(updated, join_cols=join_cols)
+
+    iceberg_table = catalog.load_table("features.driver_stats")
+    result = iceberg_table.scan().to_arrow().sort_by("driver_id")
+    assert result.num_rows == 2
+    assert result.column_names == first.column_names
+    assert result["driver_id"].to_pylist() == [1, 2]
+    assert result["value"].to_pylist() == [10.0, 2.0]
diff --git a/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py b/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
index 872dd978c9e..13df0424c43 100644
--- a/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
+++ b/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py
@@ -1,9 +1,12 @@
 from datetime import timedelta
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pandas as pd
 import pyarrow as pa
+import pytest
 
+from feast import FeatureView, Field, FileSource
 from feast.infra.compute_engines.backends.pandas_backend import PandasBackend
 from feast.infra.compute_engines.dag.context import ColumnInfo, ExecutionContext
 from feast.infra.compute_engines.local.arrow_table_value import ArrowTableValue
@@ -15,10 +18,18 @@
     LocalOutputNode,
     LocalTransformationNode,
 )
+from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
 from feast.repo_config import MaterializationConfig
+from feast.types import Float64
 
 backend = PandasBackend()
 now = pd.Timestamp.utcnow()
+default_column_info = ColumnInfo(
+    join_keys=["entity_id"],
+    feature_cols=["value"],
+    ts_col="event_timestamp",
+    created_ts_col=None,
+)
 
 sample_df = pd.DataFrame(
     {
@@ -328,7 +339,7 @@ def test_local_output_node():
     context = create_context(
         node_outputs={"source": ArrowTableValue(pa.Table.from_pandas(sample_df))}
     )
-    node = LocalOutputNode("output", MagicMock())
+    node = LocalOutputNode("output", MagicMock(), default_column_info)
     node.add_input(MagicMock())
     node.inputs[0].name = "source"
     result = node.execute(context)
@@ -349,7 +360,7 @@ def test_local_output_node_online_write_default_batch():
         node_outputs={"source": ArrowTableValue(pa.Table.from_pandas(sample_df))}
     )
 
-    node = LocalOutputNode("output", feature_view)
+    node = LocalOutputNode("output", feature_view, default_column_info)
     node.add_input(MagicMock())
     node.inputs[0].name = "source"
 
@@ -375,7 +386,7 @@ def test_local_output_node_online_write_batched():
         online_write_batch_size=2
     )
 
-    node = LocalOutputNode("output", feature_view)
+    node = LocalOutputNode("output", feature_view, default_column_info)
     node.add_input(MagicMock())
     node.inputs[0].name = "source"
 
@@ -383,3 +394,189 @@ def test_local_output_node_online_write_batched():
 
     # Verify online_write_batch was called twice (4 rows / batch_size 2 = 2 batches)
     assert context.online_store.online_write_batch.call_count == 2
+
+
+def _iceberg_sink_feature_view() -> tuple[SimpleNamespace, IcebergSource]:
+    sink = IcebergSource(
+        catalog_type="sql",
+        warehouse="file:///tmp/warehouse",
+        namespace="features",
+        table="driver_stats",
+        timestamp_field="event_timestamp",
+    )
+    sink.write_materialized_table = MagicMock()
+    feature_view = SimpleNamespace(
+        name="driver_stats",
+        online=False,
+        offline=False,
+        entity_columns=[],
+        features=[],
+        source_views=[MagicMock()],
+        batch_source=sink,
+    )
+    return feature_view, sink
+
+
+def test_local_output_node_writes_mapped_keys_to_iceberg():
+    feature_view, sink = _iceberg_sink_feature_view()
+    table = pa.Table.from_pandas(sample_df.rename(columns={"entity_id": "driver_id"}))
+    context = create_context(node_outputs={"source": ArrowTableValue(table)})
+    column_info = ColumnInfo(
+        join_keys=["ENTITY_ID"],
+        feature_cols=["value"],
+        ts_col="EVENT_TIMESTAMP",
+        created_ts_col=None,
+        field_mapping={
+            "ENTITY_ID": "driver_id",
+            "EVENT_TIMESTAMP": "event_timestamp",
+        },
+    )
+    node = LocalOutputNode("output", feature_view, column_info=column_info)
+    node.add_input(MagicMock(name="source"))
+    node.inputs[0].name = "source"
+
+    node.execute(context)
+
+    sink.write_materialized_table.assert_called_once_with(
+        table,
+        join_cols=["driver_id", "event_timestamp"],
+    )
+
+
+def test_local_output_node_uses_timestamp_only_key_for_entityless_view():
+    feature_view, sink = _iceberg_sink_feature_view()
+    table = pa.Table.from_pandas(sample_df.drop(columns=["entity_id"]))
+    context = create_context(node_outputs={"source": ArrowTableValue(table)})
+    column_info = ColumnInfo(
+        join_keys=[],
+        feature_cols=["value"],
+        ts_col="event_timestamp",
+        created_ts_col=None,
+    )
+    node = LocalOutputNode("output", feature_view, column_info=column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    node.execute(context)
+
+    assert sink.write_materialized_table.call_args.kwargs["join_cols"] == [
+        "event_timestamp"
+    ]
+
+
+def test_local_output_node_skips_all_writes_for_empty_input():
+    feature_view, sink = _iceberg_sink_feature_view()
+    feature_view.online = True
+    feature_view.offline = True
+    empty = pa.Table.from_pandas(sample_df.iloc[:0])
+    context = create_context(node_outputs={"source": ArrowTableValue(empty)})
+    column_info = ColumnInfo(
+        join_keys=["entity_id"],
+        feature_cols=["value"],
+        ts_col="event_timestamp",
+        created_ts_col=None,
+    )
+    node = LocalOutputNode("output", feature_view, column_info=column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    node.execute(context)
+
+    sink.write_materialized_table.assert_not_called()
+    context.online_store.online_write_batch.assert_not_called()
+    context.offline_store.offline_write_batch.assert_not_called()
+
+
+def test_local_output_node_propagates_iceberg_failure():
+    feature_view, sink = _iceberg_sink_feature_view()
+    sink.write_materialized_table.side_effect = RuntimeError("commit conflict")
+    context = create_context(
+        node_outputs={"source": ArrowTableValue(pa.Table.from_pandas(sample_df))}
+    )
+    column_info = ColumnInfo(
+        join_keys=["entity_id"],
+        feature_cols=["value"],
+        ts_col="event_timestamp",
+        created_ts_col=None,
+    )
+    node = LocalOutputNode("output", feature_view, column_info=column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    with pytest.raises(RuntimeError, match="commit conflict"):
+        node.execute(context)
+
+
+def test_local_output_node_rejects_missing_iceberg_timestamp():
+    feature_view, sink = _iceberg_sink_feature_view()
+    table = pa.Table.from_pandas(sample_df.drop(columns=["event_timestamp"]))
+    context = create_context(node_outputs={"source": ArrowTableValue(table)})
+    column_info = ColumnInfo(
+        join_keys=["entity_id"],
+        feature_cols=["value"],
+        ts_col="event_timestamp",
+        created_ts_col=None,
+    )
+    node = LocalOutputNode("output", feature_view, column_info=column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    with pytest.raises(ValueError, match="timestamp column is missing"):
+        node.execute(context)
+
+    sink.write_materialized_table.assert_not_called()
+
+
+def test_local_output_node_runs_online_offline_and_iceberg_writes():
+    feature_view, sink = _iceberg_sink_feature_view()
+    feature_view.online = True
+    feature_view.offline = True
+    context = create_context(
+        node_outputs={"source": ArrowTableValue(pa.Table.from_pandas(sample_df))}
+    )
+    node = LocalOutputNode("output", feature_view, default_column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    node.execute(context)
+
+    context.online_store.online_write_batch.assert_called_once()
+    context.offline_store.offline_write_batch.assert_called_once()
+    sink.write_materialized_table.assert_called_once()
+
+
+def test_local_output_node_routes_real_derived_feature_view_batch_source():
+    parent = FeatureView(
+        name="parent",
+        entities=[],
+        schema=[Field(name="value", dtype=Float64)],
+        source=FileSource(path="parent.parquet", timestamp_field="event_timestamp"),
+    )
+    sink = IcebergSource(
+        catalog_type="sql",
+        warehouse="file:///tmp/warehouse",
+        namespace="features",
+        table="derived",
+        timestamp_field="event_timestamp",
+    )
+    sink.write_materialized_table = MagicMock()
+    derived = FeatureView(
+        name="derived",
+        entities=[],
+        schema=[Field(name="value", dtype=Float64)],
+        source=parent,
+        sink_source=sink,
+        online=False,
+        offline=False,
+    )
+    context = create_context(
+        node_outputs={"source": ArrowTableValue(pa.Table.from_pandas(sample_df))}
+    )
+    node = LocalOutputNode("output", derived, default_column_info)
+    node.add_input(MagicMock())
+    node.inputs[0].name = "source"
+
+    node.execute(context)
+
+    assert derived.batch_source is sink
+    sink.write_materialized_table.assert_called_once()