# Incremental code examples

This page provides code examples for incremental Iceberg transforms. For more background on the Foundry APIs used below, see [Incremental Python transforms with Iceberg](../incremental-transforms/index.md) and [Iceberg changelogs and CDC pipelines](../changelogs/index.md). For a conceptual introduction to working with Iceberg tables incrementally, see [Incremental processing with Iceberg tables](../incremental-overview/index.md).

## Append-only incremental using Foundry APIs

In an append-only incremental transform, both the read and the write Foundry APIs default to their incremental behavior: reads return the rows appended since the last build, and writes append to the output rather than replacing it.

The `table_read_mode="append_only"` option is included here for clarity. You can omit it and get the same behavior because append-only is the default [read mode](../incremental-transforms/index.md#incremental-decorator-and-read-mode).

```python tab="Polars"
import polars as pl
from transforms.api import transform, incremental
from transforms.tables import IncrementalIcebergInput, IncrementalIcebergOutput, TableInput, TableOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.polars()
    active = added_rows.filter(pl.col("status") == "active")
    output.write_table(active)
```

```python tab="DuckDB"
from transforms.api import LightweightContext, incremental, transform
from transforms.tables import IncrementalIcebergInput, IncrementalIcebergOutput, TableInput, TableOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: LightweightContext, source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    conn = ctx.duckdb().conn
    conn.register("added_rows", source.arrow())
    query = conn.sql("SELECT * FROM added_rows WHERE status = 'active'")
    output.write_table(query.to_arrow_table())
```

```python tab="pandas"
from transforms.api import incremental, transform
from transforms.tables import IncrementalIcebergInput, IncrementalIcebergOutput, TableInput, TableOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    added_rows = source.pandas()
    active = added_rows[added_rows["status"] == "active"].astype({"status": "string"})
    output.write_table(active)
```

```python tab="PySpark"
from transforms.api import incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(source: IncrementalTableTransformInput, output: TableTransformOutput):
    added_rows = source.dataframe()
    active = added_rows.filter(added_rows.status == "active")
    output.write_dataframe(active)
```

## Append-only incremental using native Iceberg write APIs

These examples use Foundry APIs to read appended rows and native [open source Iceberg APIs](../advanced-apis/index.md) to append updates.

```python tab="PyIceberg"
from transforms.api import incremental, transform
from transforms.tables import (
    IncrementalIcebergInput,
    IncrementalIcebergOutput,
    TableInput,
    TableOutput,
)
from pyiceberg.expressions import EqualTo


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(source: IncrementalIcebergInput, output: IncrementalIcebergOutput):
    
    # Use Foundry APIs to retrieve the added rows & output
    active = source.arrow(row_filter=EqualTo("status", "active"))
    output_table = output.table()

    # Handle first run case
    if output_table is None:
        output.write_table(active)
        return

    # Use PyIceberg append to write incrementally
    with output_table.transaction() as txn:
        txn.append(active)
```

```python tab="Iceberg Spark"
from pyspark.sql.functions import col, days
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="append_only")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output")
)
def compute(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):

    # Use Foundry APIs to retrieve the added rows    
    added_rows = source.dataframe()
    active = added_rows.filter(col("status") == "active")

    # Write with Iceberg's DataFrameWriterV2 API, partitioned by a transform of a column
    writer = active.writeTo(output.identifier).partitionedBy(days(col("ts")))

    # Use ctx.is_incremental to differentiate between incremental and non-incremental resolution cases
    if ctx.is_incremental:
        writer.append()
    else:
        writer.createOrReplace()
```

## Changelog incremental using Foundry APIs

This example demonstrates how to read an Iceberg changelog and apply it to an output table based on a unique identifier using Foundry's changelog APIs.

The example assumes `id` is unique in the source and keeps only active rows in the output. When a row becomes inactive, it sends a `DELETE` to filter the inactive row from the output.

```python tab="PySpark"
from transforms.api import (
    IncrementalTableTransformInput,
    IncrementalTransformContext,
    incremental,
    transform
)
from transforms.tables import TableInput, TableOutput, TableTransformOutput
from pyspark.sql import functions as F


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def cdc_transform(
    ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput
):
    if not ctx.is_incremental:
        output.write_dataframe(source.dataframe().filter(F.col("status") == "active"))
        return

    changelog_df = source.changelog(["id"])

    reconciled = changelog_df.withColumn(
        "_change_type",
        F.when(
            (F.col("_change_type") != "DELETE")
            & (F.col("status") == "active"),
            F.lit("INSERT"),
        ).otherwise(F.lit("DELETE")),
    )

    output.apply_changelog(reconciled, ["id"])
```

## Changelog incremental using Iceberg APIs

Call Iceberg's [`create_changelog_view` procedure ↗](https://iceberg.apache.org/docs/latest/spark-procedures/#create_changelog_view) directly when you need options that Foundry's API does not expose. Without a snapshot range, it reads from the table's first snapshot.

The native procedure includes `UPDATE_BEFORE` rows and intermediate changes across snapshots. This example stores the last processed snapshot in an output table property and selects the latest action before merging. It assumes `id` is unique in the source, with one action per key per snapshot after removing `UPDATE_BEFORE`.

This example shows the basic structure of the merge. Consider the appropriate `MERGE INTO` logic carefully in the context of your pipeline.

```python tab="Iceberg Spark"
from transforms.api import IncrementalTransformContext, incremental, transform, IncrementalTableTransformInput
from transforms.tables import TableInput, TableOutput, TableTransformOutput

_WATERMARK_PROP = "watermark.source-snapshot-id"  # Last source snapshot ID processed


@incremental(v2_semantics=True).with_table_incremental_options(table_read_mode="changelog")
@transform.spark.using(
    source=TableInput("/path/input"),
    output=TableOutput("/path/output"),
)
def compute(ctx: IncrementalTransformContext, source: IncrementalTableTransformInput, output: TableTransformOutput):

    spark = ctx.spark_session
    source_tbl = source.identifier
    output_tbl = output.identifier

    # The snapshot this build should read up to: the source's current snapshot.
    end = spark.sql(f"""
        SELECT snapshot_id FROM {source_tbl}.snapshots
        ORDER BY committed_at DESC LIMIT 1
    """).collect()
    end = end[0]["snapshot_id"] if end else None

    # First build, or a build that could not resolve incrementally (using Foundry APIs)
    if not ctx.is_incremental:
        source_df = source.dataframe()
        output.write_dataframe(source_df.filter(source_df.status == "active"))
        # Set watermark so the first incremental build has a start point
        spark.sql(f"ALTER TABLE {output.identifier} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")
        return

    # The last snapshot previously processed, read back from the output table property.
    start = spark.sql(f"""
        SHOW TBLPROPERTIES {output_tbl} ('{_WATERMARK_PROP}')
    """).collect()[0]["value"]

    if start == str(end):
        # Nothing new to process.
        return

    # Build the changelog over the range (start, end]
    spark.sql(f"""
        CALL {source.catalog}.system.create_changelog_view(
            table => '{source_tbl}',
            changelog_view => 'source_changelog',
            identifier_columns => array('id'),
            options => map('start-snapshot-id', '{start}', 'end-snapshot-id', '{end}')
        )
    """)

    # Apply the deduplicated changelog
    spark.sql(f"""
        MERGE INTO {output_tbl} AS tgt
        USING (
            SELECT * FROM (
                SELECT *,
                    ROW_NUMBER() OVER (
                        PARTITION BY id ORDER BY _change_ordinal DESC
                    ) AS _rn
                FROM source_changelog
                WHERE _change_type != 'UPDATE_BEFORE'
            )
            WHERE _rn = 1
        ) AS src
        ON src.id = tgt.id
        WHEN MATCHED AND (src._change_type = 'DELETE' OR src.status != 'active') THEN DELETE
        WHEN MATCHED AND src.status = 'active' THEN UPDATE SET *
        WHEN NOT MATCHED AND src._change_type != 'DELETE' AND src.status = 'active' THEN INSERT *
    """)

    # Update the watermark to track latest read snapshot
    spark.sql(f"ALTER TABLE {output_tbl} SET TBLPROPERTIES ('{_WATERMARK_PROP}' = '{end}')")
```