Repository: mrjsj/msfabricutils

Issue:
add delta load methods
add different load methods 
- overwrite
- append
- upsert
- incremental
- replace

add some transformation step to do scd 2

Discussion:

Patch:
diff --git a/docs/etl/index.md b/docs/etl/index.md
index 923f424..632c2c5 100644
--- a/docs/etl/index.md
+++ b/docs/etl/index.md
@@ -1,3 +1,5 @@
 # ETL
 
-The `msfabricutils.etl` module provides a set of functions and classes to facilitate ETL (Extract, Transform, Load) operations in the Microsoft Fabric environment.
+The `msfabricutils.etl` module provides a set of functions and classes to facilitate ETL (Extract, Transform, Load) operations.
+
+The docs provide some examples on how to use the functions, however more usecases can be found in the tests section on github.
\ No newline at end of file
diff --git a/docs/etl/load.md b/docs/etl/load.md
new file mode 100644
index 0000000..3918680
--- /dev/null
+++ b/docs/etl/load.md
@@ -0,0 +1,3 @@
+# Load
+
+::: msfabricutils.etl.load.delta_table
diff --git a/docs/etl/read.md b/docs/etl/read.md
new file mode 100644
index 0000000..3e1c1f7
--- /dev/null
+++ b/docs/etl/read.md
@@ -0,0 +1,5 @@
+# Read
+
+::: msfabricutils.etl.read.delta_table
+::: msfabricutils.etl.read.parquet
+
diff --git a/docs/etl/sinks.md b/docs/etl/sinks.md
deleted file mode 100644
index 7e9a13c..0000000
--- a/docs/etl/sinks.md
+++ /dev/null
@@ -1,3 +0,0 @@
-# Sinks
-
-::: msfabricutils.etl.sinks
diff --git a/docs/etl/sources.md b/docs/etl/sources.md
deleted file mode 100644
index 81047c9..0000000
--- a/docs/etl/sources.md
+++ /dev/null
@@ -1,4 +0,0 @@
-# Sources
-
-::: msfabricutils.etl.sources
-
diff --git a/docs/etl/transform.md b/docs/etl/transform.md
new file mode 100644
index 0000000..bac7456
--- /dev/null
+++ b/docs/etl/transform.md
@@ -0,0 +1,3 @@
+# Transform
+
+::: msfabricutils.etl.transform.transforms
diff --git a/docs/etl/transforms.md b/docs/etl/transforms.md
deleted file mode 100644
index ddf8f66..0000000
--- a/docs/etl/transforms.md
+++ /dev/null
@@ -1,3 +0,0 @@
-# Transforms
-
-::: msfabricutils.etl.transforms
diff --git a/docs/usage/etl.md b/docs/usage/etl.md
index 200253b..ceb4022 100644
--- a/docs/usage/etl.md
+++ b/docs/usage/etl.md
@@ -8,7 +8,7 @@ The ETL module provides a set of helper functions for extracting, transforming,
 
 ```python
 from msfabricutils.etl import (
-    source_parquet,
+    read_parquet,
     upsert_scd_type_1
 )
 import polars as pl
@@ -21,11 +21,11 @@ df = pl.DataFrame({
 df.write_parquet("source.parquet")
 
 # Read the source parquet file
-source_df = source_parquet("source.parquet")
+source_df = read_parquet("source.parquet")
 
 # Upsert to a target table
-upsert_scd_type_1(
-    table_uri="target_table",
+upsert(
+    table_or_uri="target_table",
     df=source_df,
     primary_key_columns="id"
 )
@@ -48,12 +48,12 @@ The result will be a new row for AliceS and BobB in the target table.
 ```python
 from msfabricutils.etl import (
     get_default_config,
-    source_delta,
+    read_delta,
     get_incremental_column_value,
-    deduplicate_transform,
-    normalize_column_names_transform,
-    add_audit_columns_transform,
-    upsert_scd_type_1,
+    deduplicate,
+    normalize_column_names,
+    add_audit_columns,
+    upsert,
 )
 
 source_table_path = "source_table"
@@ -71,7 +71,7 @@ source_df.write_delta(source_table_path)
 config = get_default_config()
 
 # Read the source delta table
-source_df = source_delta(source_table_path)
+source_df = read_delta(source_table_path)
 
 # Get the incremental column value
 incremental_column_value = get_incremental_column_value(target_table_path, "batch_id")
@@ -82,14 +82,19 @@ filtered_df = source_df.filter(pl.col("batch_id") > incremental_column_value)
 # Deduplicate the source dataframe
 deduped_df = deduplicate_transform(filtered_df, primary_key_columns="ID", deduplication_order_columns="batch_id")
 
-# Normalize the column names
-normalized_df = normalize_column_names_transform(deduped_df, config)
+# Normalize the column names using a normalization strategy
+def normalize_column_names(text: str) -> str:
+    return text.lower().replace(" ", "_")
+
+normalized_df = normalize_column_names_transform(deduped_df, normalize_column_names)
 
 # Add audit columns
-audit_df = add_audit_columns_transform(normalized_df, config)
+audit_columns = config.get_audit_columns()
+audit_df = add_audit_columns_transform(normalized_df, audit_columns)
 
 # Upsert to a target table
-upsert_scd_type_1(target_table_path, audit_df, primary_key_columns="ID", config=config)
+static_audit_columns = config.get_static_audit_columns()
+upsert(target_table_path, audit_df, primary_key_columns="id", update_exclusion_columns=static_audit_columns)
 
 
 ```
diff --git a/mkdocs.yml b/mkdocs.yml
index d233443..1d7987c 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -63,17 +63,17 @@ nav:
     - usage/etl.md
     - usage/fabric-api.md
   - API Reference:
-    - Common:
-      - common/fabric-duckdb-connection.md
-      - common/utilities.md
     - Core:
       - core/authentication.md
       - core/fabric-api.md
     - ETL:
       - etl/index.md
-      - etl/sources.md
-      - etl/transforms.md
-      - etl/sinks.md
+      - etl/read.md
+      - etl/transform.md
+      - etl/load.md
+    - Common:
+      - common/fabric-duckdb-connection.md
+      - common/utilities.md
 
 # Formatting options
 markdown_extensions:
diff --git a/src/msfabricutils/core/auth.py b/src/msfabricutils/core/auth.py
index 8c8e349..165da45 100644
--- a/src/msfabricutils/core/auth.py
+++ b/src/msfabricutils/core/auth.py
@@ -58,16 +58,12 @@ def get_fabric_bearer_token() -> str:
 
 def get_azure_devops_access_token() -> str:
     """
-    Retrieves a bearer token for Fabric (Power BI) API.
+    Retrieves a bearer token for Azure DevOps.
 
-    This function attempts to obtain a bearer token for authenticating requests to the
-    Power BI API. It first checks if the code is running in a Microsoft Fabric
-    notebook environment and tries to use the `notebookutils` library to get the token.
-    If the library is not available, it falls back to using the `DefaultAzureCredential`
-    from the Azure SDK to fetch the token.
+    This function attempts to obtain a bearer token for authenticating requests to Azure DevOps.
 
     Returns:
-        The bearer token used for authenticating requests to the Azure Fabric (Power BI) API.
+        The bearer token used for authenticating requests to Azure DevOps.
     """
     audience = "499b84ac-1321-427f-aa17-267ca6975798"
     return get_access_token(audience)
diff --git a/src/msfabricutils/etl/__init__.py b/src/msfabricutils/etl/__init__.py
index 34257d4..f1ffa27 100644
--- a/src/msfabricutils/etl/__init__.py
+++ b/src/msfabricutils/etl/__init__.py
@@ -1,11 +1,11 @@
-from .config import AuditColumn, Config, IncrementalColumn, create_config, get_default_config
-from .helpers import get_incremental_column_value
-from .sinks import upsert_scd_type_1
-from .sources import source_delta, source_parquet
-from .transforms import (
-    add_audit_columns_transform,
-    deduplicate_transform,
-    normalize_column_names_transform,
+from .config import Column, Config, IncrementalColumn, create_config, get_default_config
+from .helpers import get_max_column_value
+from .load import upsert
+from .read import read_delta, read_parquet
+from .transform import (
+    add_audit_columns,
+    deduplicate,
+    normalize_column_names,
 )
 
 __all__ = (
@@ -13,12 +13,12 @@ __all__ = (
     "create_config",
     "Config",
     "IncrementalColumn",
-    "AuditColumn",
-    "upsert_scd_type_1",
-    "source_parquet",
-    "source_delta",
-    "deduplicate_transform",
-    "normalize_column_names_transform",
-    "add_audit_columns_transform",
-    "get_incremental_column_value",
+    "Column",
+    "upsert",
+    "read_parquet",
+    "read_delta",
+    "deduplicate",
+    "normalize_column_names",
+    "add_audit_columns",
+    "get_max_column_value",
 )
diff --git a/src/msfabricutils/etl/config.py b/src/msfabricutils/etl/config.py
index 59f9101..ba0f536 100644
--- a/src/msfabricutils/etl/config.py
+++ b/src/msfabricutils/etl/config.py
@@ -27,7 +27,32 @@ class IncrementalColumn:
 
 
 @dataclass
-class AuditColumn:
+class KeyColumn:
+    """
+    Represents a key column in the configuration.
+
+    Attributes:
+        suffix (str): The suffix of the key column.
+        data_type (pl.DataType, optional): The data type of the key column.
+
+    Example:
+        ```python
+        key_column = KeyColumn("_sk", pl.Int64)
+        ```
+    """
+
+    prefix: str | None = None
+    suffix: str | None = None
+    data_type: pl.DataType | None = None
+    default_value: pl.Expr | None = None
+
+    def __post_init__(self):
+        if self.prefix is None and self.suffix is None:
+            raise ValueError("Either prefix or suffix must be provided")
+
+
+@dataclass
+class Column:
     """
     Represents an audit column in the configuration.
 
@@ -63,32 +88,43 @@ class Config:
         character_translation_map (dict[str, str]): A mapping of special characters to their translations.
         normalization_strategy (Callable[[str], str]): A function that takes a column name and returns the normalized name.
     """
+    surrogate_key_column: KeyColumn
+    historical_surrogate_key_calendar_column_base_name: str
+    historical_surrogate_key_column: KeyColumn
+    business_key_column: KeyColumn
+    composite_key_column: KeyColumn
 
     incremental_column: IncrementalColumn
-    column_created_at: AuditColumn
-    column_modified_at: AuditColumn
-    column_deleted_at: AuditColumn
-    column_valid_from: AuditColumn
-    column_valid_to: AuditColumn
+    column_created_at: Column
+    column_modified_at: Column
+    column_deleted_at: Column
+    column_valid_from: Column
+    column_valid_to: Column
     character_translation_map: dict[str, str]
     normalization_strategy: Callable[[str], str]
 
     def __init__(self):
+        self.surrogate_key_column = KeyColumn("_sk", pl.Int64, pl.lit(-1))
+        self.historical_surrogate_key_calendar_column_base_name = "calendar"
+        self.historical_surrogate_key_column = KeyColumn("_hsk", pl.Int64, pl.lit(-1))
+        self.business_key_column = KeyColumn("_bk")
+        self.composite_key_column = KeyColumn("_ck")
+
         # TODO: Change to `__run_id`
         self.incremental_column = IncrementalColumn("batch_id", pl.Int64)
-        self.column_created_at = AuditColumn(
+        self.column_created_at = Column(
             "__created_at", pl.lit(datetime.now(timezone.utc)).cast(pl.Datetime("us", "UTC"))
         )
-        self.column_modified_at = AuditColumn(
+        self.column_modified_at = Column(
             "__modified_at", pl.lit(datetime.now(timezone.utc)).cast(pl.Datetime("us", "UTC"))
         )
-        self.column_deleted_at = AuditColumn(
+        self.column_deleted_at = Column(
             "__deleted_at", pl.lit(None).cast(pl.Datetime("us", "UTC"))
         )
-        self.column_valid_from = AuditColumn(
+        self.column_valid_from = Column(
             "__valid_from", pl.lit(datetime.now(timezone.utc)).cast(pl.Datetime("us", "UTC"))
         )
-        self.column_valid_to = AuditColumn(
+        self.column_valid_to = Column(
             "__valid_to", pl.lit(None).cast(pl.Datetime("us", "UTC"))
         )
         self.character_translation_map = {
@@ -126,7 +162,7 @@ class Config:
             character_translation(name, self.character_translation_map)
         )
 
-    def get_static_audit_columns(self) -> list[AuditColumn]:
+    def get_static_audit_columns(self) -> list[Column]:
         """
         Returns a list of static audit columns, namely the `created_at` and `valid_from` columns.
 
@@ -143,7 +179,7 @@ class Config:
             self.column_valid_from,
         ]
 
-    def get_dynamic_audit_columns(self) -> list[AuditColumn]:
+    def get_dynamic_audit_columns(self) -> list[Column]:
         """
         Returns a list of dynamic audit columns, namely the `modified_at` and `valid_to` columns.
 
@@ -161,7 +197,7 @@ class Config:
             self.column_deleted_at,
         ]
 
-    def get_audit_columns(self) -> list[AuditColumn]:
+    def get_audit_columns(self) -> list[Column]:
         """
         Returns a list of all audit columns, namely the `created_at`, `modified_at`, `valid_from`, and `valid_to` columns.
 
@@ -184,11 +220,11 @@ class Config:
 
 def create_config(
     incremental_column: IncrementalColumn,
-    created_at: AuditColumn,
-    modified_at: AuditColumn,
-    deleted_at: AuditColumn,
-    valid_from: AuditColumn,
-    valid_to: AuditColumn,
+    created_at: Column,
+    modified_at: Column,
+    deleted_at: Column,
+    valid_from: Column,
+    valid_to: Column,
 ) -> Config:
     """
     Creates a new Config instance with the provided audit and incremental columns.
@@ -242,3 +278,4 @@ def get_default_config() -> Config:
         ```
     """
     return Config()
+
diff --git a/src/msfabricutils/etl/helpers/__init__.py b/src/msfabricutils/etl/helpers/__init__.py
index 6b9bc33..91e8931 100644
--- a/src/msfabricutils/etl/helpers/__init__.py
+++ b/src/msfabricutils/etl/helpers/__init__.py
@@ -1,4 +1,4 @@
-from .get_incremental_column_value import get_incremental_column_value
+from .get_max_column_value import get_max_column_value
 from .merge_helpers import (
     build_merge_predicate,
     build_when_matched_update_columns,
@@ -9,5 +9,5 @@ __all__ = (
     "build_merge_predicate",
     "build_when_matched_update_predicate",
     "build_when_matched_update_columns",
-    "get_incremental_column_value",
+    "get_max_column_value",
 )
diff --git a/src/msfabricutils/etl/helpers/get_incremental_column_value.py b/src/msfabricutils/etl/helpers/get_incremental_column_value.py
deleted file mode 100644
index 603b6ae..0000000
--- a/src/msfabricutils/etl/helpers/get_incremental_column_value.py
+++ /dev/null
@@ -1,37 +0,0 @@
-import polars as pl
-from deltalake import DeltaTable
-
-from msfabricutils.core.auth import get_storage_options
-
-
-def get_incremental_column_value(table_uri: str, incremental_column: str) -> int:
-    """
-    Retrieves the maximum value of the specified incremental column from a Delta table.
-
-    Args:
-        table_uri (str): The URI of the Delta table.
-        incremental_column (str): The name of the incremental column.
-
-    Returns:
-        The maximum value of the incremental column, or 0 if the table does not exist.
-
-    Example:
-        ```python
-        from msfabricutils.etl import get_incremental_column_value
-
-        max_value = get_incremental_column_value("path/to/delta_table", "incremental_id")
-        ```
-    """
-
-    storage_options = get_storage_options() if table_uri.startswith("abfss://") else None
-
-    if not DeltaTable.is_deltatable(table_uri, storage_options=storage_options):
-        return 0
-
-    return (
-        pl.scan_delta(table_uri, storage_options=storage_options)
-        .select(pl.col(incremental_column))
-        .max()
-        .collect()
-        .item()
-    )
diff --git a/src/msfabricutils/etl/helpers/get_max_column_value.py b/src/msfabricutils/etl/helpers/get_max_column_value.py
new file mode 100644
index 0000000..31316de
--- /dev/null
+++ b/src/msfabricutils/etl/helpers/get_max_column_value.py
@@ -0,0 +1,38 @@
+import polars as pl
+from deltalake import DeltaTable
+
+from msfabricutils.core.auth import get_storage_options
+
+
+def get_max_column_value(table_or_uri: str | DeltaTable, column_name: str) -> int:
+    """
+    Retrieves the maximum value of the specified column from a Delta table.
+
+    Args:
+        table_or_uri (str | DeltaTable): A string URI to a Delta table or a DeltaTable instance.
+        column_name (str): The name of the column.
+
+    Returns:
+        The maximum value of the column, or 0 if the table does not exist.
+
+    Example:
+        ```python
+        from msfabricutils.etl import get_max_column_value
+
+        max_value = get_max_column_value("path/to/delta_table", "incremental_id")
+        ```
+    """
+
+    storage_options = get_storage_options() if table_or_uri.startswith("abfss://") else None
+    
+    if isinstance(table_or_uri, str):
+        if not DeltaTable.is_deltatable(table_or_uri, storage_options=storage_options):
+            return 0
+
+    return (
+        pl.scan_delta(table_or_uri, storage_options=storage_options)
+        .select(pl.col(column_name))
+        .max()
+        .collect()
+        .item()
+    )
diff --git a/src/msfabricutils/etl/helpers/get_min_column_value.py b/src/msfabricutils/etl/helpers/get_min_column_value.py
new file mode 100644
index 0000000..374c87a
--- /dev/null
+++ b/src/msfabricutils/etl/helpers/get_min_column_value.py
@@ -0,0 +1,38 @@
+import polars as pl
+from deltalake import DeltaTable
+
+from msfabricutils.core.auth import get_storage_options
+
+
+def get_min_column_value(table_or_uri: str | DeltaTable, column_name: str) -> int:
+    """
+    Retrieves the maximum value of the specified column from a Delta table.
+
+    Args:
+        table_or_uri (str | DeltaTable): A string URI to a Delta table or a DeltaTable instance.
+        column_name (str): The name of the column.
+
+    Returns:
+        The minimum value of the column, or 0 if the table does not exist.
+
+    Example:
+        ```python
+        from msfabricutils.etl import get_min_column_value
+
+        min_value = get_min_column_value("path/to/delta_table", "incremental_id")
+        ```
+    """
+
+    storage_options = get_storage_options() if table_or_uri.startswith("abfss://") else None
+    
+    if isinstance(table_or_uri, str):
+        if not DeltaTable.is_deltatable(table_or_uri, storage_options=storage_options):
+            return 0
+
+    return (
+        pl.scan_delta(table_or_uri, storage_options=storage_options)
+        .select(pl.col(column_name))
+        .min()
+        .collect()
+        .item()
+    )
diff --git a/src/msfabricutils/etl/helpers/get_or_create_delta_table.py b/src/msfabricutils/etl/helpers/get_or_create_delta_table.py
new file mode 100644
index 0000000..f935181
--- /dev/null
+++ b/src/msfabricutils/etl/helpers/get_or_create_delta_table.py
@@ -0,0 +1,27 @@
+import polars as pl
+from deltalake import DeltaTable
+
+from msfabricutils.core.auth import get_storage_options
+
+
+def get_or_create_delta_table(table_uri: str, schema: pl.Schema) -> DeltaTable:
+    """
+    Retrieves a Delta table or creates a new one if it does not exist.
+
+    Args:
+        table_uri (str): The URI of the Delta table.
+        schema (pl.Schema): The Polars schema to create the Delta table with.
+
+    Returns:
+        The Delta table.
+    """
+
+    storage_options = get_storage_options() if table_uri.startswith("abfss://") else None
+
+    if DeltaTable.is_deltatable(table_uri, storage_options=storage_options):
+        dt = DeltaTable(table_uri, storage_options=storage_options)
+    else:
+        dt = DeltaTable.create(table_uri, schema, storage_options=storage_options)
+
+    return dt
+
diff --git a/src/msfabricutils/etl/load/__init__.py b/src/msfabricutils/etl/load/__init__.py
new file mode 100644
index 0000000..18afdf9
--- /dev/null
+++ b/src/msfabricutils/etl/load/__init__.py
@@ -0,0 +1,3 @@
+from .delta_table import append, incremental, overwrite, replace_range, upsert
+
+__all__ = ("write", "upsert", "overwrite", "append", "incremental", "replace_range")
diff --git a/src/msfabricutils/etl/load/delta_table.py b/src/msfabricutils/etl/load/delta_table.py
new file mode 100644
index 0000000..85e8dba
--- /dev/null
+++ b/src/msfabricutils/etl/load/delta_table.py
@@ -0,0 +1,277 @@
+
+import polars as pl
+from deltalake import DeltaTable, write_deltalake
+
+from msfabricutils.etl.helpers import get_max_column_value
+from msfabricutils.etl.helpers.get_or_create_delta_table import get_or_create_delta_table
+from msfabricutils.etl.helpers.merge_helpers import (
+    build_merge_predicate,
+    build_when_matched_update_columns,
+    build_when_matched_update_predicate,
+)
+from msfabricutils.etl.types import PolarsFrame
+
+
+def upsert(
+    table_or_uri: str | DeltaTable,
+    df: PolarsFrame,
+    primary_key_columns: str | list[str],
+    update_exclusion_columns: str | list[str] | None = None,
+    predicate_exclusion_columns: str | list[str] | None = None,
+) -> dict[str, str]:
+    """
+    Upserts dataframe into a Delta table using the provided primary key columns.
+
+    Args:
+        table_or_uri (str): The URI of the target Delta table.
+        df (PolarsFrame): The dataframe to upsert.
+        primary_key_columns (str | list[str]): Primary key column(s) for the upsert.
+        update_exclusion_columns (str | list[str] | None): Columns which will not be updated in the merge.
+        predicate_exclusion_columns (str | list[str] | None): Columns to exclude from the upsert. Difference between source and target of these columns will not trigger an update, however if there is a difference in the other columns, the row will be *also* be updated with the `excluded_columns`.
+
+    Returns:
+        Result of the merge operation.
+
+    Example:
+        ```python
+        from msfabricutils.etl import Config, upsert_scd_type_1
+        import polars as pl
+
+
+        config = get_default_config()
+        data = pl.DataFrame({...})
+
+        upsert_scd_type_1(
+            "path/to/delta_table",
+            data,
+            primary_key_columns=["id"],
+            static_audit_columns=["created_at"],
+            dynamic_audit_columns=["modified_at", "valid_to"],
+        )
+        ```
+    """
+
+    if isinstance(primary_key_columns, str):
+        primary_key_columns = [primary_key_columns]
+
+    if update_exclusion_columns is None:
+        update_exclusion_columns = []
+
+    if isinstance(update_exclusion_columns, str):
+        update_exclusion_columns = [update_exclusion_columns]        
+
+    if predicate_exclusion_columns is None:
+        predicate_exclusion_columns = []
+
+    if isinstance(predicate_exclusion_columns, str):
+        predicate_exclusion_columns = [predicate_exclusion_columns]
+
+    if isinstance(df, pl.LazyFrame):
+        df = df.collect()
+
+    df = df.to_arrow()
+
+    if isinstance(table_or_uri, str):
+        dt = get_or_create_delta_table(table_or_uri, df.schema)
+    else:
+        dt = table_or_uri
+
+    merge_predicate = build_merge_predicate(primary_key_columns)
+
+    predicate_update_columns = [
+        column
+        for column in df.column_names
+        if column
+        not in primary_key_columns + predicate_exclusion_columns + update_exclusion_columns
+    ]
+    when_matched_update_predicates = build_when_matched_update_predicate(predicate_update_columns)
+
+    update_columns = [
+        column
+        for column in df.column_names
+        if column not in primary_key_columns + update_exclusion_columns
+    ]
+    when_matched_update_columns = build_when_matched_update_columns(update_columns)
+
+    table_merger = (
+        dt.merge(
+            df,
+            source_alias="source",
+            target_alias="target",
+            predicate=merge_predicate,
+        )
+        .when_matched_update(
+            predicate=when_matched_update_predicates, updates=when_matched_update_columns
+        )
+        .when_not_matched_insert_all()
+    )
+
+    return table_merger.execute()
+
+
+def overwrite(table_or_uri: str | DeltaTable, df: PolarsFrame) -> None:
+    """
+    Overwrites the entire Delta table with the provided dataframe.
+
+    Args:
+        table_or_uri (str): The URI of the target Delta table.
+        df (PolarsFrame): The dataframe to write to the Delta table.
+
+    Example:
+        ```python
+        from msfabricutils.etl import overwrite
+        import polars as pl
+
+        data = pl.DataFrame({...})
+
+        overwrite("path/to/delta_table", data)
+        ```
+    """
+
+    if isinstance(table_or_uri, str):
+        dt = get_or_create_delta_table(table_or_uri, df.schema)
+    else:
+        dt = table_or_uri
+
+    if isinstance(df, pl.LazyFrame):
+        df = df.collect()
+
+    df = df.to_arrow()
+
+    write_deltalake(
+        table_or_uri=dt,
+        data=df,
+        mode="overwrite",
+    )
+
+
+def replace_range(
+    table_or_uri: str | DeltaTable,
+    df: PolarsFrame,
+    range_column: str,
+) -> None:
+    """
+    Replaces a range of data in the Delta table based on a specified column.
+
+    Args:
+        table_or_uri (str): The URI of the target Delta table.
+        df (PolarsFrame): The dataframe to write to the Delta table.
+        range_column (str): The column used to determine the range of data to replace. This replaces the data in the range of the `range_column` in the delta table based off the min and max values of the `range_column` in the dataframe.
+
+    Example:
+        ```python
+        from msfabricutils.etl import replace_range
+        import polars as pl
+
+        data = pl.DataFrame({...})
+
+        replace_range("path/to/delta_table", data, range_column="date")
+        ```
+    """
+
+    if isinstance(table_or_uri, str):
+        dt = get_or_create_delta_table(table_or_uri, df.schema)
+    else:
+        dt = table_or_uri
+
+    min_value, max_value = (
+        df.select(
+            pl.col(range_column).min(),
+            pl.col(range_column).max(),
+        )
+        .collect()
+        .row(0)
+    )
+
+    predicate = f'{range_column} >= "{min_value}" AND {range_column} <= "{max_value}"'
+
+    if isinstance(df, pl.LazyFrame):
+        df = df.collect()
+
+    df = df.to_arrow()
+
+    write_deltalake(
+        table_or_uri=dt,
+        data=df,
+        mode="overwrite",
+        predicate=predicate,
+    )
+
+
+def append(table_or_uri: str | DeltaTable, df: PolarsFrame) -> None:
+    """
+    Appends the provided dataframe to the Delta table.
+
+    Args:
+        table_or_uri (str): The URI of the target Delta table.
+        df (PolarsFrame): The dataframe to append to the Delta table.
+
+    Example:
+        ```python
+        from msfabricutils.etl import append
+        import polars as pl
+
+        data = pl.DataFrame({...})
+
+        append("path/to/delta_table", data)
+        ```
+    """
+
+    if isinstance(table_or_uri, str):
+        dt = get_or_create_delta_table(table_or_uri, df.schema)
+    else:
+        dt = table_or_uri
+
+    if isinstance(df, pl.LazyFrame):
+        df = df.collect()
+
+    df = df.to_arrow()
+
+    write_deltalake(
+        table_or_uri=dt,
+        data=df,
+        mode="append",
+    )
+
+
+def incremental(table_or_uri: str | DeltaTable, df: PolarsFrame, incremental_column: str) -> None:
+    """
+    Appends new data to the Delta table based on an incremental column.
+
+    Args:
+        table_or_uri (str): The URI of the target Delta table.
+        df (PolarsFrame): The dataframe to append to the Delta table.
+        incremental_column (str): The column used to determine new data to append. The source dataframe will only append rows where the value of the `incremental_column` is greater than the max value of the `incremental_column` in the delta table.
+
+    Example:
+        ```python
+        from msfabricutils.etl import incremental
+        import polars as pl
+
+        data = pl.DataFrame({...})
+
+        incremental("path/to/delta_table", data, incremental_column="timestamp")
+        ```
+    """
+
+    if isinstance(table_or_uri, str):
+        dt = get_or_create_delta_table(table_or_uri, df.schema)
+    else:
+        dt = table_or_uri
+
+    max_value = get_max_column_value(table_or_uri, incremental_column)
+
+    df = df.filter(pl.col(incremental_column) > max_value)
+
+    if isinstance(df, pl.LazyFrame):
+        df = df.collect()
+
+    df = df.to_arrow()
+
+    write_deltalake(
+        table_or_uri=dt,
+        data=df,
+        mode="append",
+    )
+
+
diff --git a/src/msfabricutils/etl/read/__init__.py b/src/msfabricutils/etl/read/__init__.py
new file mode 100644
index 0000000..a9835c4
--- /dev/null
+++ b/src/msfabricutils/etl/read/__init__.py
@@ -0,0 +1,7 @@
+from .delta_table import read_delta
+from .parquet import read_parquet
+
+__all__ = (
+    "read_delta",
+    "read_parquet",
+)
diff --git a/src/msfabricutils/etl/sources/delta_table.py b/src/msfabricutils/etl/read/delta_table.py
similarity index 83%
rename from src/msfabricutils/etl/sources/delta_table.py
rename to src/msfabricutils/etl/read/delta_table.py
index 6128578..52289ba 100644
--- a/src/msfabricutils/etl/sources/delta_table.py
+++ b/src/msfabricutils/etl/read/delta_table.py
@@ -4,7 +4,7 @@ from msfabricutils.core.auth import get_storage_options
 from msfabricutils.etl.types import PolarsFrame
 
 
-def source_delta(table_uri: str, eager: bool = False) -> PolarsFrame:
+def read_delta(table_uri: str, eager: bool = False) -> PolarsFrame:
     """
     Reads a Delta table from the specified abfss URI. Automatically handles the authentication with OneLake.
 
@@ -17,15 +17,15 @@ def source_delta(table_uri: str, eager: bool = False) -> PolarsFrame:
 
     Example:
         ```python
-        from msfabricutils.etl import source_delta
+        from msfabricutils.etl import read_delta
 
         workspace_id = "12345678-1234-1234-1234-123456789012"
         lakehouse_id = "beefbeef-beef-beef-beef-beefbeefbeef"
         table_name = "my-delta-table"
         table_uri = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{lakehouse_id}/Tables/{table_name}"
 
-        df = source_delta(table_uri, eager=True)
-        lazy_df = source_delta(table_uri, eager=False)
+        df = read_delta(table_uri, eager=True)
+        lazy_df = read_delta(table_uri, eager=False)
         ```
     """
 
diff --git a/src/msfabricutils/etl/sources/parquet.py b/src/msfabricutils/etl/read/parquet.py
similarity index 82%
rename from src/msfabricutils/etl/sources/parquet.py
rename to src/msfabricutils/etl/read/parquet.py
index b1a8fed..fbb0d32 100644
--- a/src/msfabricutils/etl/sources/parquet.py
+++ b/src/msfabricutils/etl/read/parquet.py
@@ -4,7 +4,7 @@ from msfabricutils.core.auth import get_storage_options
 from msfabricutils.etl.types import PolarsFrame
 
 
-def source_parquet(table_uri: str, eager: bool = False) -> PolarsFrame:
+def read_parquet(table_uri: str, eager: bool = False) -> PolarsFrame:
     """
     Reads a Parquet file from the specified abfss URI. Automatically handles the authentication with OneLake.
 
@@ -18,7 +18,7 @@ def source_parquet(table_uri: str, eager: bool = False) -> PolarsFrame:
     Example:
         Reading a single file
         ```python
-        from msfabricutils.etl import source_parquet
+        from msfabricutils.etl import read_parquet
 
         workspace_id = "12345678-1234-1234-1234-123456789012"
         lakehouse_id = "beefbeef-beef-beef-beef-beefbeefbeef"
@@ -26,18 +26,18 @@ def source_parquet(table_uri: str, eager: bool = False) -> PolarsFrame:
         file_path = "my-parquet-file.parquet"
         folder_uri = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{lakehouse_id}/Files/"
 
-        df = source_parquet(folder_uri + file_path, eager=True)
+        df = read_parquet(folder_uri + file_path, eager=True)
         ```
 
         Reading all Parquet files in a folder
         ```python
-        from msfabricutils.etl import source_parquet
+        from msfabricutils.etl import read_parquet
 
         workspace_id = "12345678-1234-1234-1234-123456789012"
         lakehouse_id = "beefbeef-beef-beef-beef-beefbeefbeef"
 
         folder_uri = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{lakehouse_id}/Files/"
-        glob_df = source_parquet(folder_uri + "**/*.parquet", eager=True)
+        glob_df = read_parquet(folder_uri + "**/*.parquet", eager=True)
         ```
     """
 
diff --git a/src/msfabricutils/etl/sinks/__init__.py b/src/msfabricutils/etl/sinks/__init__.py
deleted file mode 100644
index 6b716ed..0000000
--- a/src/msfabricutils/etl/sinks/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-from .delta_table import upsert_scd_type_1
-
-__all__ = ("upsert_scd_type_1",)
diff --git a/src/msfabricutils/etl/sinks/delta_table.py b/src/msfabricutils/etl/sinks/delta_table.py
deleted file mode 100644
index 9f3ccb5..0000000
--- a/src/msfabricutils/etl/sinks/delta_table.py
+++ /dev/null
@@ -1,108 +0,0 @@
-import polars as pl
-from deltalake import DeltaTable
-
-from msfabricutils.etl import Config
-from msfabricutils.etl.helpers.merge_helpers import (
-    build_merge_predicate,
-    build_when_matched_update_columns,
-    build_when_matched_update_predicate,
-)
-from msfabricutils.etl.types import PolarsFrame
-
-
-def upsert_scd_type_1(
-    table_uri: str,
-    df: PolarsFrame,
-    primary_key_columns: str | list[str],
-    config: Config | None = None,
-    exclude_columns: str | list[str] | None = None,
-) -> dict[str:str]:
-    """
-    Upserts dataframe into a Delta table using Slowly Changing Dimension (SCD) Type 1.
-
-    Args:
-        table_uri (str): The URI of the target Delta table.
-        df (PolarsFrame): The dataframe to upsert.
-        config (Config | None): Configuration object containing audit column information.
-        primary_key_columns (str | list[str]): Primary key column(s) for the upsert.
-        exclude_columns (str | list[str] | None): Columns to exclude from the upsert.
-
-    Returns:
-        Result of the merge operation.
-
-    Example:
-        ```python
-        from msfabricutils.etl import Config, upsert_scd_type_1
-        import polars as pl
-
-
-        config = get_default_config()
-        data = pl.DataFrame({...})
-
-        upsert_scd_type_1(
-            "path/to/delta_table",
-            data,
-            config,
-            primary_key_columns=["id"]
-        )
-        ```
-    """
-
-    dynamic_audit_columns = config.get_dynamic_audit_columns() if config else []
-    static_audit_columns = config.get_static_audit_columns() if config else []
-
-    if isinstance(primary_key_columns, str):
-        primary_key_columns = [primary_key_columns]
-
-    if config:
-        primary_key_columns = [
-            config.normalization_strategy(column) for column in primary_key_columns
-        ]
-
-    if exclude_columns is None:
-        exclude_columns = []
-
-    if isinstance(exclude_columns, str):
-        exclude_columns = [exclude_columns]
-
-    if isinstance(df, pl.LazyFrame):
-        df = df.collect()
-
-    df = df.to_arrow()
-
-    if DeltaTable.is_deltatable(table_uri):
-        dt = DeltaTable(table_uri)
-    else:
-        dt = DeltaTable.create(table_uri, df.schema)
-
-    merge_predicate = build_merge_predicate(primary_key_columns)
-
-    predicate_update_columns = [
-        column
-        for column in df.column_names
-        if column
-        not in primary_key_columns + exclude_columns + static_audit_columns + dynamic_audit_columns
-    ]
-
-    when_matched_update_predicates = build_when_matched_update_predicate(predicate_update_columns)
-    update_columns = [
-        column
-        for column in df.column_names
-        if column not in primary_key_columns + exclude_columns + static_audit_columns
-    ]
-
-    when_matched_update_columns = build_when_matched_update_columns(update_columns)
-    table_merger = (
-        dt.merge(
-            df,
-            source_alias="source",
-            target_alias="target",
-            predicate=merge_predicate,
-        )
-        .when_matched_update(
-            predicate=when_matched_update_predicates, updates=when_matched_update_columns
-        )
-        .when_not_matched_insert_all()
-    )
-
-    return table_merger.execute()
diff --git a/src/msfabricutils/etl/sources/__init__.py b/src/msfabricutils/etl/sources/__init__.py
deleted file mode 100644
index 8041589..0000000
--- a/src/msfabricutils/etl/sources/__init__.py
+++ /dev/null
@@ -1,7 +0,0 @@
-from .delta_table import source_delta
-from .parquet import source_parquet
-
-__all__ = (
-    "source_delta",
-    "source_parquet",
-)
diff --git a/src/msfabricutils/etl/transform/__init__.py b/src/msfabricutils/etl/transform/__init__.py
new file mode 100644
index 0000000..6518bf0
--- /dev/null
+++ b/src/msfabricutils/etl/transform/__init__.py
@@ -0,0 +1,11 @@
+from .transforms import (
+    add_audit_columns,
+    deduplicate,
+    normalize_column_names,
+)
+
+__all__ = (
+    "add_audit_columns",
+    "deduplicate",
+    "normalize_column_names",
+)
diff --git a/src/msfabricutils/etl/transform/transforms.py b/src/msfabricutils/etl/transform/transforms.py
new file mode 100644
index 0000000..ebce5ee
--- /dev/null
+++ b/src/msfabricutils/etl/transform/transforms.py
@@ -0,0 +1,353 @@
+from typing import Callable
+
+import polars as pl
+import polars.selectors as cs
+from polars.exceptions import ColumnNotFoundError
+
+from msfabricutils.etl.config import Column
+from msfabricutils.etl.types import PolarsFrame
+
+
+def add_audit_columns(df: PolarsFrame, audit_columns: list[Column]) -> PolarsFrame:
+    """
+    Adds audit columns to the given DataFrame or LazyFrame based on the configuration.
+
+    Args:
+        df (PolarsFrame): The DataFrame or LazyFrame to which audit columns will be added.
+        audit_columns (list[Column]): A list of audit columns to put on the DataFrame.
+
+    Returns:
+        The DataFrame or LazyFrame with the added audit columns.
+
+    Example:
+        ```python
+        from msfabricutils.etl import add_audit_columns
+        import polars as pl
+
+        audit_columns = [
+            Column(
+               "created_at", pl.lit(datetime.now(timezone.utc)).cast(pl.Datetime("us", "UTC"))
+            )
+        ]
+        df = pl.DataFrame({"data": [1, 2, 3]})
+        updated_df = add_audit_columns(df, audit_columns)
+        ```
+    """
+
+    df = df.with_columns(
+        [audit_column.default_value.alias(audit_column.name) for audit_column in audit_columns]
+    )
+    return df
+
+
+def deduplicate(
+    df: PolarsFrame,
+    primary_key_columns: str | list[str] | None = None,
+    deduplication_order_columns: str | list[str] | None = None,
+    deduplication_order_descending: bool | list[bool] = True,
+) -> PolarsFrame:
+    """
+    Removes duplicate rows from the DataFrame based on primary key columns.
+
+    Args:
+        df (PolarsFrame): The DataFrame or LazyFrame from which duplicates will be removed.
+        primary_key_columns (list[str] | None): The columns to use as primary keys for deduplication.
+        deduplication_order_columns (list[str] | None): The columns to determine the order of rows for deduplication.
+        deduplication_order_descending (bool | list[bool]): Whether to sort the deduplication order in descending order.
+
+    Returns:
+        PolarsFrame: The DataFrame or LazyFrame with duplicates removed.
+
+    Example:
+        ```python
+        import polars as pl
+
+        df = pl.DataFrame({
+            "id": [1, 2, 2, 3],
+            "value": ["a", "b", "b", "c"]
+        })
+        deduped_df = deduplicate(df, primary_key_columns=["id"])
+        ```
+    """
+
+    if isinstance(primary_key_columns, str):
+        primary_key_columns = [primary_key_columns]
+
+    # Temporary fix start
+    # See GitHub issue: https://github.com/pola-rs/polars/issues/20209
+    # TODO: Remove this once the issue is fixed.
+    # .unique() does not check if subset columns exist in the dataframe if it is empty, so it's silently ignores.
+
+    if isinstance(df, pl.LazyFrame):
+        columns = df.collect_schema().names()
+    else:
+        columns = df.schema.names()
+
+    if primary_key_columns:
+        for column in primary_key_columns:
+            if column not in columns:
+                raise ColumnNotFoundError(
+                    f"unable to find column `{column}`. Valid columns: {columns}"
+                )
+
+    # Temporary fix end
+
+    if deduplication_order_columns:
+        df = df.sort(
+            deduplication_order_columns, descending=deduplication_order_descending, nulls_last=True
+        )
+
+    df = df.unique(subset=primary_key_columns, keep="first")
+
+    return df
+
+
+def normalize_column_names(
+    df: PolarsFrame, normalization_strategy: Callable[[str], str]
+) -> PolarsFrame:
+    """
+    Normalizes the column names of the DataFrame using a provided normalization strategy.
+
+    Args:
+        df (PolarsFrame): The DataFrame or LazyFrame whose column names will be normalized.
+        normalization_strategy (Callable[[str], str]): A callable which takes a string and returns a modified string.
+
+    Returns:
+        PolarsFrame: The DataFrame or LazyFrame with normalized column names.
+
+    Example:
+        ```python
+        import polars as pl
+
+        def my_strategy(old_column_name: str) -> str:
+            new_name = old_column_name.replace(" ").lower()
+            return new_name
+
+        df = pl.DataFrame({"First Name": [1, 2], "Last Name": [3, 4]})
+        normalized_df = normalize_column_names(df, my_strategy)
+        ```
+    """
+
+    if isinstance(df, pl.LazyFrame):
+        columns = df.collect_schema().names()
+    else:
+        columns = df.schema.names()
+
+    column_mapping = {old_name: normalization_strategy(old_name) for old_name in columns}
+
+    df = df.rename(column_mapping)
+
+    return df
+
+
+def reorder_columns_by_suffix(
+    df: PolarsFrame, suffix_order: list[str], sort_alphabetically_within_group: bool = True
+) -> PolarsFrame:
+    """
+    Reorders DataFrame columns based on their suffixes according to the provided order.
+
+    Args:
+        df (PolarsFrame): The DataFrame or LazyFrame whose columns will be reordered.
+        suffix_order (list[str]): List of suffixes in the desired order.
+        sort_alphabetically_within_group (bool): Whether to sort columns alphabetically within each suffix group.
+
+    Returns:
+        PolarsFrame: The DataFrame or LazyFrame with reordered columns.
+
+    Example:
+        ```python
+        import polars as pl
+
+        df = pl.DataFrame({
+            "name_key": ["a", "b"],
+            "age_key": [1, 2],
+            "name_value": ["x", "y"],
+            "age_value": [10, 20]
+        })
+        reordered_df = reorder_columns_by_suffix(df, suffix_order=["_key", "_value"])
+        ```
+    """
+    if isinstance(df, pl.LazyFrame):
+        columns = df.collect_schema().names()
+    else:
+        columns = df.schema.names()
+
+    if sort_alphabetically_within_group:
+        columns = sorted(columns)
+
+    reordered_columns = []
+    for suffix in suffix_order:
+        reordered_columns.extend(
+            [
+                column
+                for column in columns
+                if column.endswith(suffix) and column not in reordered_columns
+            ]
+        )
+
+    reordered_columns.extend([column for column in columns if column not in reordered_columns])
+
+    return df.select(reordered_columns)
+
+
+def reorder_columns_by_prefix(
+    df: PolarsFrame, prefix_order: list[str], sort_alphabetically_within_group: bool = True
+) -> PolarsFrame:
+    """
+    Reorders DataFrame columns based on their prefixes according to the provided order.
+
+    Args:
+        df (PolarsFrame): The DataFrame or LazyFrame whose columns will be reordered.
+        prefix_order (list[str]): List of prefixes in the desired order.
+        sort_alphabetically_within_group (bool): Whether to sort columns alphabetically within each prefix group.
+
+    Returns:
+        PolarsFrame: The DataFrame or LazyFrame with reordered columns.
+
+    Example:
+        ```python
+        import polars as pl
+
+        df = pl.DataFrame({
+            "dim_name": ["a", "b"],
+            "dim_age": [1, 2],
+            "fact_sales": [100, 200],
+            "fact_quantity": [5, 10]
+        })
+        reordered_df = reorder_columns_by_prefix(df, prefix_order=["dim_", "fact_"])
+        ```
+    """
+    if isinstance(df, pl.LazyFrame):
+        columns = df.collect_schema().names()
+    else:
+        columns = df.schema.names()
+
+    if sort_alphabetically_within_group:
+        columns = sorted(columns)
+
+    reordered_columns = []
+    for suffix in prefix_order:
+        reordered_columns.extend(
+            [
+                column
+                for column in columns
+                if column.startswith(suffix) and column not in reordered_columns
+            ]
+        )
+
+    reordered_columns.extend([column for column in columns if column not in reordered_columns])
+
+    return df.select(reordered_columns)
+
+
+def apply_scd_type_2(
+    source_df: PolarsFrame,
+    target_df: PolarsFrame,
+    primary_key_columns: str | list[str],
+    valid_from_column: str,
+    valid_to_column: str,
+) -> PolarsFrame:
+    """
+    Applies Slowly Changing Dimension (SCD) Type 2 logic to merge source and target DataFrames.
+    SCD Type 2 maintains historical records by creating new rows for changed data while preserving
+    the history through valid_from and valid_to dates.
+
+    Args:
+        source_df (PolarsFrame): The new/source DataFrame containing updated records.
+        target_df (PolarsFrame): The existing/target DataFrame containing current records.
+        primary_key_columns (str | list[str]): Column(s) that uniquely identify each entity.
+        valid_from_column (str): Column name containing the validity start date.
+        valid_to_column (str): Column name containing the validity end date.
+
+    Returns:
+        PolarsFrame: A DataFrame containing both current and historical records with updated
+        validity periods. The result maintains the full history of changes while ensuring
+        proper date ranges for overlapping records.
+
+    Example:
+        ```python
+        import polars as pl
+        from datetime import datetime
+
+        # Create sample source and target dataframes
+        source_df = pl.DataFrame({
+            "customer_id": [1, 2],
+            "name": ["John Updated", "Jane"],
+            "valid_from": [datetime(2024, 1, 1), datetime(2024, 1, 1)]
+        })
+
+        target_df = pl.DataFrame({
+            "customer_id": [1, 2],
+            "name": ["John", "Jane"],
+            "valid_from": [datetime(2023, 1, 1), datetime(2023, 1, 1)],
+            "valid_to": [None, None]
+        })
+
+        # Apply SCD Type 2
+        result_df = apply_scd_type_2(
+            source_df=source_df,
+            target_df=target_df,
+            primary_key_columns="customer_id",
+            valid_from_column="valid_from",
+            valid_to_column="valid_to"
+        )
+        ```
+
+    Notes:
+        - The function handles overlapping date ranges by adjusting valid_to dates
+        - NULL in valid_to indicates a currently active record
+        - Records in source_df will create new versions if they differ from target_df
+        - Historical records are preserved with appropriate valid_to dates
+    """
+    if isinstance(primary_key_columns, str):
+        primary_key_columns = [primary_key_columns]
+
+    target_records_to_be_updated = (
+        target_df.join(
+            source_df,
+            on=primary_key_columns,
+            how="left",
+            suffix="__source",
+        )
+    )
+    print(target_records_to_be_updated)
+    target_records_to_be_updated = (
+        target_records_to_be_updated
+        .filter(
+            (pl.col(valid_from_column) > pl.col(valid_from_column + "__source"))
+            | 
+            (
+                (pl.col(valid_from_column + "__source") > pl.col(valid_from_column))
+                & (
+                    (pl.col(valid_from_column + "__source") <= pl.col(valid_to_column))
+                    | pl.col(valid_to_column).is_null()
+                )
+            )
+        )
+    )
+    print(target_records_to_be_updated)
+
+    target_records_to_be_updated = target_records_to_be_updated.unique(
+        subset=primary_key_columns + [valid_from_column]
+    )
+    print(target_records_to_be_updated)
+    target_records_to_be_updated = (
+        target_records_to_be_updated
+        .select(~cs.ends_with("__source"))
+        .join(
+            source_df,
+            on=primary_key_columns + [valid_from_column],
+            how="anti",
+        )
+    )
+    print(target_records_to_be_updated)
+    upsert_df: PolarsFrame = pl.concat([target_records_to_be_updated, source_df])
+    upsert_df = upsert_df.with_columns(
+        pl.col(valid_from_column)
+        .shift(-1)
+        .over(partition_by=primary_key_columns, order_by=valid_from_column)
+        .alias(valid_to_column)
+    )
+    print(upsert_df.sort(primary_key_columns + [valid_from_column]))
+
+    return upsert_df
\ No newline at end of file
diff --git a/src/msfabricutils/etl/transforms/__init__.py b/src/msfabricutils/etl/transforms/__init__.py
deleted file mode 100644
index 3b1a521..0000000
--- a/src/msfabricutils/etl/transforms/__init__.py
+++ /dev/null
@@ -1,11 +0,0 @@
-from .transforms import (
-    add_audit_columns_transform,
-    deduplicate_transform,
-    normalize_column_names_transform,
-)
-
-__all__ = (
-    "add_audit_columns_transform",
-    "deduplicate_transform",
-    "normalize_column_names_transform",
-)
diff --git a/src/msfabricutils/etl/transforms/transforms.py b/src/msfabricutils/etl/transforms/transforms.py
deleted file mode 100644
index 3d8ceb9..0000000
--- a/src/msfabricutils/etl/transforms/transforms.py
+++ /dev/null
@@ -1,138 +0,0 @@
-import polars as pl
-from polars.exceptions import ColumnNotFoundError
-
-from msfabricutils.etl import Config
-from msfabricutils.etl.types import PolarsFrame
-
-
-def add_audit_columns_transform(df: PolarsFrame, config: Config) -> PolarsFrame:
-    """
-    Adds audit columns to the given DataFrame or LazyFrame based on the configuration.
-
-    Args:
-        df (PolarsFrame): The DataFrame or LazyFrame to which audit columns will be added.
-        config (Config): The configuration object that provides the audit column definitions.
-
-    Returns:
-        The DataFrame or LazyFrame with the added audit columns.
-
-    Example:
-        ```python
-        from msfabricutils.etl import get_default_config, add_audit_columns_transform
-        import polars as pl
-
-
-        config = get_default_config()
-        df = pl.DataFrame({"data": [1, 2, 3]})
-        updated_df = add_audit_columns_transform(df, config)
-
-        ```
-    """
-
-    audit_columns = config.get_audit_columns()
-
-    df = df.with_columns(
-        [audit_column.default_value.alias(audit_column.name) for audit_column in audit_columns]
-    )
-    return df
-
-
-def deduplicate_transform(
-    df: PolarsFrame,
-    primary_key_columns: str | list[str] | None = None,
-    deduplication_order_columns: str | list[str] | None = None,
-    deduplication_order_descending: bool | list[bool] = True,
-) -> PolarsFrame:
-    """
-    Removes duplicate rows from the DataFrame based on primary key columns.
-
-    Args:
-        df (PolarsFrame): The DataFrame or LazyFrame from which duplicates will be removed.
-        primary_key_columns (list[str] | None): The columns to use as primary keys for deduplication.
-        deduplication_order_columns (list[str] | None): The columns to determine the order of rows for deduplication.
-        deduplication_order_descending (bool | list[bool]): Whether to sort the deduplication order in descending order.
-
-    Returns:
-        PolarsFrame: The DataFrame or LazyFrame with duplicates removed.
-
-    Example:
-        ```python
-        import polars as pl
-
-        df = pl.DataFrame({
-            "id": [1, 2, 2, 3],
-            "value": ["a", "b", "b", "c"]
-        })
-        deduped_df = deduplicate_transform(df, primary_key_columns=["id"])
-        ```
-    """
-
-    if isinstance(primary_key_columns, str):
-        primary_key_columns = [primary_key_columns]
-
-    # Temporary fix start
-    # See GitHub issue: https://github.com/pola-rs/polars/issues/20209
-    # TODO: Remove this once the issue is fixed.
-    # .unique() does not check if subset columns exist in the dataframe if it is empty, so it's silently ignores.
-
-    if isinstance(df, pl.LazyFrame):
-        columns = df.collect_schema().names()
-    else:
-        columns = df.schema.names()
-
-    if primary_key_columns:
-        for column in primary_key_columns:
-            if column not in columns:
-                raise ColumnNotFoundError(
-                    f"unable to find column `{column}`. Valid columns: {columns}"
-                )
-
-    # Temporary fix end
-
-    if deduplication_order_columns:
-        df = df.sort(
-            deduplication_order_columns, descending=deduplication_order_descending, nulls_last=True
-        )
-
-    df = df.unique(subset=primary_key_columns, keep="first")
-
-    return df
-
-
-def normalize_column_names_transform(df: PolarsFrame, config: Config) -> PolarsFrame:
-    """
-    Normalizes the column names of the DataFrame using a provided normalization strategy.
-
-    Args:
-        df (PolarsFrame): The DataFrame or LazyFrame whose column names will be normalized.
-        config (Config): The configuration object that provides the normalization strategy.
-
-    Returns:
-        PolarsFrame: The DataFrame or LazyFrame with normalized column names.
-
-    Example:
-        ```python
-        import polars as pl
-        from msfabricutils.etl import get_default_config
-
-        config = get_default_config()
-
-        df = pl.DataFrame({"First Name": [1, 2], "Last Name": [3, 4]})
-        normalized_df = normalize_column_names_transform(df, config)
-        ```
-    """
-
-    if isinstance(df, pl.LazyFrame):
-        columns = df.collect_schema().names()
-    else:
-        columns = df.schema.names()
-
-    column_mapping = {old_name: config.normalization_strategy(old_name) for old_name in columns}
-
-    df = df.rename(column_mapping)
-
-    return df
-
-
-# def filter_source(df: PolarsFrame, filter: Callable[[PolarsFrame], PolarsFrame]) -> PolarsFrame:
-#     return filter(df)


Test patch:
diff --git a/tests/etl/test_end_to_end.py b/tests/etl/test_end_to_end.py
index 9792c42..934ebf5 100644
--- a/tests/etl/test_end_to_end.py
+++ b/tests/etl/test_end_to_end.py
@@ -5,13 +5,13 @@ from freezegun import freeze_time
 from polars.testing import assert_frame_equal
 
 from msfabricutils.etl import (
-    add_audit_columns_transform,
-    deduplicate_transform,
+    add_audit_columns,
+    deduplicate,
     get_default_config,
-    get_incremental_column_value,
-    normalize_column_names_transform,
-    source_delta,
-    upsert_scd_type_1,
+    get_max_column_value,
+    normalize_column_names,
+    read_delta,
+    upsert,
 )
 
 
@@ -34,27 +34,35 @@ def test_end_to_end(tmp_path):
     config = get_default_config()
 
     # Read the source delta table
-    source_df = source_delta(source_table_path)
+    source_df = read_delta(source_table_path)
 
     # Get the incremental column value
-    incremental_column_value = get_incremental_column_value(target_table_path, "batch_id")
+    incremental_column_value = get_max_column_value(target_table_path, "batch_id")
 
     # Filter the source dataframe to only get the rows with a modified_at greater than the incremental column value
     filtered_df = source_df.filter(pl.col("batch_id") > incremental_column_value)
 
     # Deduplicate the source dataframe
-    deduped_df = deduplicate_transform(
+    deduped_df = deduplicate(
         filtered_df, primary_key_columns="ID", deduplication_order_columns="batch_id"
     )
 
     # Normalize the column names
-    normalized_df = normalize_column_names_transform(deduped_df, config)
+
+    strategy = config.normalization_strategy
+    normalized_df = normalize_column_names(deduped_df, strategy)
 
     # Add audit columns
-    audit_df = add_audit_columns_transform(normalized_df, config)
+    audit_df = add_audit_columns(normalized_df, config.get_audit_columns())
 
     # Upsert to a target table
-    upsert_scd_type_1(target_table_path, audit_df, primary_key_columns="ID", config=config)
+    upsert(
+        target_table_path,
+        audit_df,
+        primary_key_columns="id",
+        update_exclusion_columns=config.get_static_audit_columns(),
+        predicate_exclusion_columns=config.get_dynamic_audit_columns(),
+    )
 
     # Read the target table
     target_df = pl.read_delta(target_table_path)
diff --git a/tests/etl/transforms/test_dimension_lookup.py b/tests/etl/transforms/test_dimension_lookup.py
new file mode 100644
index 0000000..34c1954
--- /dev/null
+++ b/tests/etl/transforms/test_dimension_lookup.py
@@ -0,0 +1,171 @@
+
+
+#from msfabricutils.etl.transforms.transforms import get_dimension_surrogate_key, get_dimension_historical_surrogate_key
+
+
+# @pytest.fixture
+# def config():
+#     return get_default_config()
+
+# @pytest.fixture
+# def mock_dimension_table(tmp_path, config):
+#     # Create a sample dimension table with both current and historical records
+#     df = pl.DataFrame({
+#         "customer_bk": ["C1", "C1", "C2"],
+#         "customer_sk": [1, 1, 2],
+#         "valid_from": [
+#             datetime(2023, 1, 1),
+#             datetime(2023, 6, 1),
+#             datetime(2023, 1, 1)
+#         ],
+#         "valid_to": [
+#             datetime(2023, 6, 1),
+#             None,
+#             None
+#         ]
+#     })
+    
+#     # Save as Delta table
+#     delta_path = tmp_path / "dimension_table"
+#     df.write_delta(delta_path)
+#     return str(delta_path)
+
+# def test_get_dimension_surrogate_key_business_key(mock_dimension_table, config):
+#     # Test data with business key
+#     fact_df = pl.DataFrame({
+#         "customer_bk": ["C1", "C2", "C3"],
+#         "amount": [100, 200, 300]
+#     })
+    
+#     result = get_dimension_surrogate_key(
+#         fact_df,
+#         mock_dimension_table,
+#         config,
+#         keep_key_columns=True
+#     )
+    
+#     assert "customer_sk" in result.columns
+#     assert "customer_bk" in result.columns
+#     assert result.filter(pl.col("customer_bk") == "C1")["customer_sk"][0] == 1
+#     assert result.filter(pl.col("customer_bk") == "C2")["customer_sk"][0] == 2
+#     # C3 should get default surrogate key (usually -1)
+#     assert result.filter(pl.col("customer_bk") == "C3")["customer_sk"][0] == config.surrogate_key_column.default_value
+
+# def test_get_dimension_surrogate_key_historical(mock_dimension_table, config):
+#     # Test data with business key and date for historical lookup
+#     fact_df = pl.DataFrame({
+#         "customer_bk": ["C1", "C1"],
+#         "date_bk": [date(2023, 3, 1), date(2023, 7, 1)],
+#         "amount": [100, 200]
+#     })
+    
+#     result = get_dimension_surrogate_key(
+#         fact_df,
+#         mock_dimension_table,
+#         config,
+#         keep_key_columns=True,
+#         create_historical_surrogate_key=True
+#     )
+    
+#     # Verify current surrogate keys
+#     assert result.filter(pl.col("customer_bk") == "C1")["customer_sk"].to_list() == [1, 1]
+    
+#     # Verify historical surrogate keys (if implemented)
+#     if "customer_hsk" in result.columns:
+#         historical_sks = result.filter(pl.col("customer_bk") == "C1")["customer_hsk"].to_list()
+#         assert len(historical_sks) == 2
+#         assert historical_sks[0] != historical_sks[1]  # Different periods should have different historical keys 
+
+# def test_get_dimension_historical_surrogate_key():
+#     # Arrange
+#     config = get_default_config()
+    
+#     # Create sample fact data
+#     fact_df = pl.DataFrame({
+#         "customer_id_bk": ["C1", "C2", "C3"],
+#         "order_date_bk": ["2024-01-01", "2024-01-02", "2024-01-03"]
+#     })
+
+#     # Create sample dimension data with historical records
+#     dimension_df = pl.DataFrame({
+#         "customer_id_bk": ["C1", "C1", "C2"],
+#         "customer_sk": [1, 2, 3],
+#         "valid_from": ["2023-12-01", "2024-01-02", "2023-12-01"],
+#         "valid_to": ["2024-01-02", None, None]
+#     })
+
+#     # Act
+#     result_df = get_dimension_historical_surrogate_key(
+#         df=fact_df,
+#         dimension_df=dimension_df,
+#         key_columns=["customer_id_bk"],
+#         config=config
+#     )
+
+#     # Assert
+#     expected_df = pl.DataFrame({
+#         "customer_id_bk": ["C1", "C2", "C3"],
+#         "order_date_bk": ["2024-01-01", "2024-01-02", "2024-01-03"],
+#         "customer_hsk": [1, 3, -1]  # -1 is the default value for non-matching records
+#     })
+
+#     assert result_df.collect() if isinstance(result_df, pl.LazyFrame) else result_df == expected_df
+
+
+# def test_get_dimension_historical_surrogate_key_with_empty_dataframe():
+#     # Arrange
+#     config = get_default_config()
+    
+#     fact_df = pl.DataFrame({
+#         "customer_id_bk": [],
+#         "order_date_bk": []
+#     })
+
+#     dimension_df = pl.DataFrame({
+#         "customer_id_bk": ["C1"],
+#         "customer_sk": [1],
+#         "valid_from": ["2023-12-01"],
+#         "valid_to": [None]
+#     })
+
+#     # Act
+#     result_df = get_dimension_historical_surrogate_key(
+#         df=fact_df,
+#         dimension_df=dimension_df,
+#         key_columns=["customer_id_bk"],
+#         config=config
+#     )
+
+#     # Assert
+#     expected_df = pl.DataFrame({
+#         "customer_id_bk": [],
+#         "order_date_bk": [],
+#         "customer_hsk": []
+#     })
+
+#     assert result_df.collect() if isinstance(result_df, pl.LazyFrame) else result_df == expected_df
+
+
+# def test_get_dimension_historical_surrogate_key_invalid_key():
+#     # Arrange
+#     config = get_default_config()
+    
+#     fact_df = pl.DataFrame({
+#         "customer_id_bk": ["C1"]
+#     })
+
+#     dimension_df = pl.DataFrame({
+#         "customer_id_bk": ["C1"],
+#         "customer_sk": [1],
+#         "valid_from": ["2023-12-01"],
+#         "valid_to": [None]
+#     })
+
+#     # Act & Assert
+#     with pytest.raises(Exception):  # Replace with specific exception if known
+#         get_dimension_historical_surrogate_key(
+#             df=fact_df,
+#             dimension_df=dimension_df,
+#             key_columns=["invalid_column"],
+#             config=config
+#         )
\ No newline at end of file
diff --git a/tests/etl/transforms/test_transforms.py b/tests/etl/transforms/test_transforms.py
index aac702d..2911093 100644
--- a/tests/etl/transforms/test_transforms.py
+++ b/tests/etl/transforms/test_transforms.py
@@ -1,11 +1,20 @@
 from datetime import datetime, timezone
 
 import polars as pl
+import pytest
 from freezegun import freeze_time
+from polars.exceptions import ColumnNotFoundError
 from polars.testing import assert_frame_equal
 
 from msfabricutils.etl import get_default_config
-from msfabricutils.etl.transforms.transforms import add_audit_columns_transform
+from msfabricutils.etl.transform.transforms import (
+    add_audit_columns,
+    apply_scd_type_2,
+    deduplicate,
+    normalize_column_names,
+    reorder_columns_by_prefix,
+    reorder_columns_by_suffix,
+)
 
 
 @freeze_time("2024-12-06 12:00:00")
@@ -14,7 +23,9 @@ def test_add_audit_columns():
 
     df = pl.DataFrame({"data": [1, 2, 3]})
 
-    actual_df = add_audit_columns_transform(df, config)
+    audit_columns = config.get_audit_columns()
+
+    actual_df = add_audit_columns(df, audit_columns)
 
     datetime_now = datetime(2024, 12, 6, 12, 0, tzinfo=timezone.utc)
 
@@ -34,3 +45,520 @@ def test_add_audit_columns():
     )
 
     assert_frame_equal(actual_df, expected_df)
+
+def test_reorder_columns_by_suffix():
+    # Test case with alphabetical sorting within groups
+    df = pl.DataFrame({
+        "name_key": ["a", "b"],
+        "age_key": [1, 2],
+        "name_value": ["x", "y"],
+        "age_value": [10, 20],
+        "other": [True, False]
+    })
+
+    actual_df = reorder_columns_by_suffix(df, suffix_order=["_key", "_value"])
+
+    expected_df = pl.DataFrame({
+        "age_key": [1, 2],
+        "name_key": ["a", "b"],
+        "age_value": [10, 20],
+        "name_value": ["x", "y"],
+        "other": [True, False]
+    })
+
+    assert_frame_equal(actual_df, expected_df)
+
+def test_reorder_columns_by_suffix_no_sorting():
+    # Test case without alphabetical sorting within groups
+    df = pl.DataFrame({
+        "name_key": ["a", "b"],
+        "age_key": [1, 2],
+        "name_value": ["x", "y"],
+        "age_value": [10, 20],
+        "other": [True, False]
+    })
+
+    actual_df = reorder_columns_by_suffix(
+        df, 
+        suffix_order=["_key", "_value"], 
+        sort_alphabetically_within_group=False
+    )
+
+    # Columns should maintain their original order within groups
+    expected_df = pl.DataFrame({
+        "name_key": ["a", "b"],
+        "age_key": [1, 2],
+        "name_value": ["x", "y"],
+        "age_value": [10, 20],
+        "other": [True, False]
+    })
+
+    assert_frame_equal(actual_df, expected_df)
+
+def test_reorder_columns_by_prefix():
+    # Test case with alphabetical sorting within groups
+    df = pl.DataFrame({
+        "dim_name": ["a", "b"],
+        "dim_age": [1, 2],
+        "fact_sales": [100, 200],
+        "fact_quantity": [5, 10],
+        "other": [True, False]
+    })
+
+    actual_df = reorder_columns_by_prefix(df, prefix_order=["dim_", "fact_"])
+
+    expected_df = pl.DataFrame({
+        "dim_age": [1, 2],
+        "dim_name": ["a", "b"],
+        "fact_quantity": [5, 10],
+        "fact_sales": [100, 200],
+        "other": [True, False]
+    })
+
+    assert_frame_equal(actual_df, expected_df)
+
+def test_reorder_columns_by_prefix_no_sorting():
+    # Test case without alphabetical sorting within groups
+    df = pl.DataFrame({
+        "dim_name": ["a", "b"],
+        "dim_age": [1, 2],
+        "fact_sales": [100, 200],
+        "fact_quantity": [5, 10],
+        "other": [True, False]
+    })
+
+    actual_df = reorder_columns_by_prefix(
+        df, 
+        prefix_order=["dim_", "fact_"], 
+        sort_alphabetically_within_group=False
+    )
+
+    # Columns should maintain their original order within groups
+    expected_df = pl.DataFrame({
+        "dim_name": ["a", "b"],
+        "dim_age": [1, 2],
+        "fact_sales": [100, 200],
+        "fact_quantity": [5, 10],
+        "other": [True, False]
+    })
+
+    assert_frame_equal(actual_df, expected_df)
+
+def test_deduplicate_single_key():
+    df = pl.DataFrame({
+        "id": [1, 2, 2, 3],
+        "value": ["a", "b1", "b2", "c"],
+        "timestamp": [1, 2, 3, 4]
+    })
+
+    actual_df = deduplicate(
+        df,
+        primary_key_columns="id",
+        deduplication_order_columns="timestamp",
+        deduplication_order_descending=True
+    )
+
+    expected_df = pl.DataFrame({
+        "id": [1, 2, 3],
+        "value": ["a", "b2", "c"],
+        "timestamp": [1, 3, 4]
+    })
+
+    assert_frame_equal(actual_df, expected_df, check_row_order=False)
+
+def test_deduplicate_multiple_keys():
+    df = pl.DataFrame({
+        "id": [1, 1, 2, 2],
+        "type": ["A", "A", "B", "B"],
+        "value": ["old", "new", "old", "new"],
+        "timestamp": [1, 2, 3, 4]
+    })
+
+    actual_df = deduplicate(
+        df,
+        primary_key_columns=["id", "type"],
+        deduplication_order_columns="timestamp",
+        deduplication_order_descending=True
+    )
+
+    expected_df = pl.DataFrame({
+        "id": [1, 2],
+        "type": ["A", "B"],
+        "value": ["new", "new"],
+        "timestamp": [2, 4]
+    })
+
+    assert_frame_equal(actual_df, expected_df, check_row_order=False)
+
+def test_deduplicate_no_duplicates():
+    df = pl.DataFrame({
+        "id": [1, 2, 3],
+        "value": ["a", "b", "c"]
+    })
+
+    actual_df = deduplicate(df, primary_key_columns="id")
+    assert_frame_equal(actual_df, df, check_row_order=False)
+
+def test_deduplicate_invalid_column():
+    df = pl.DataFrame({
+        "id": [1, 2, 3],
+        "value": ["a", "b", "c"]
+    })
+
+    with pytest.raises(ColumnNotFoundError):
+        deduplicate(df, primary_key_columns="non_existent")
+
+def test_normalize_column_names():
+    df = pl.DataFrame({
+        "First Name": ["John", "Jane"],
+        "Last Name": ["Doe", "Smith"],
+        "AGE": [30, 25],
+        "phone_number": [123, 456]
+    })
+
+    def normalize_strategy(col: str) -> str:
+        return col.lower().replace(" ", "_")
+
+    actual_df = normalize_column_names(df, normalize_strategy)
+
+    expected_df = pl.DataFrame({
+        "first_name": ["John", "Jane"],
+        "last_name": ["Doe", "Smith"],
+        "age": [30, 25],
+        "phone_number": [123, 456]
+    })
+
+    assert_frame_equal(actual_df, expected_df, check_row_order=False)
+
+def test_normalize_column_names_custom_strategy():
+    df = pl.DataFrame({
+        "First Name": ["John", "Jane"],
+        "Last-Name": ["Doe", "Smith"]
+    })
+
+    def custom_strategy(col: str) -> str:
+        return col.upper().replace(" ", "_").replace("-", "_")
+
+    actual_df = normalize_column_names(df, custom_strategy)
+
+    expected_df = pl.DataFrame({
+        "FIRST_NAME": ["John", "Jane"],
+        "LAST_NAME": ["Doe", "Smith"]
+    })
+
+    assert_frame_equal(actual_df, expected_df, check_row_order=False)
+
+def test_normalize_column_names_empty_df():
+    df = pl.DataFrame({
+        "First Name": [],
+        "Last Name": []
+    })
+
+    def normalize_strategy(col: str) -> str:
+        return col.lower().replace(" ", "_")
+
+    actual_df = normalize_column_names(df, normalize_strategy)
+
+    expected_df = pl.DataFrame({
+        "first_name": [],
+        "last_name": []
+    })
+
+    assert_frame_equal(actual_df, expected_df, check_row_order=False)
+
+
+def test_apply_scd_type_2_old_records():
+    # Arrange
+    # Create source dataframe with new and updated records
+    source_df = pl.DataFrame(
+        {
+            "id": [1, 3],
+            "name": ["Alice", "Charlie"],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None],
+        },
+        schema={
+            "id": pl.Int64,
+            "name": pl.Utf8,
+            "valid_from": pl.Datetime(time_zone="UTC"),
+            "valid_to": pl.Datetime(time_zone="UTC"),
+        },
+    )
+
+    # Create target dataframe with existing records
+    target_df = pl.DataFrame(
+        {
+            "id": [1, 2],
+            "name": ["Alice Updated", "Bob"],
+            "valid_from": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None],
+        },
+        schema={
+            "id": pl.Int64,
+            "name": pl.Utf8,
+            "valid_from": pl.Datetime(time_zone="UTC"),
+            "valid_to": pl.Datetime(time_zone="UTC"),
+        },        
+    )
+
+    # Act
+    result_df = apply_scd_type_2(
+        source_df=source_df,
+        target_df=target_df,
+        primary_key_columns="id",
+        valid_from_column="valid_from",
+        valid_to_column="valid_to",
+    )
+
+    # Assert
+    expected_df = pl.DataFrame(
+        {
+            "id": [1, 1, 3],
+            "name": ["Alice", "Alice Updated", "Charlie"],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                None,
+                None,
+            ],
+        }
+    )
+
+    assert_frame_equal(result_df, expected_df, check_row_order=False)
+
+
+def test_apply_scd_type_2_new_records():
+    # Arrange
+    # Create source dataframe with new and updated records
+    source_df = pl.DataFrame(
+        {
+            "id": [1, 2],
+            "name": ["Alice Updated", "Bob"],
+            "valid_from": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None],
+        }
+    )
+
+    # Create target dataframe with existing records
+    target_df = pl.DataFrame(
+        {
+            "id": [1, 3],
+            "name": ["Alice", "Charlie"],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None],
+        },
+        schema={
+            "id": pl.Int64,
+            "name": pl.Utf8,
+            "valid_from": pl.Datetime(time_zone="UTC"),
+            "valid_to": pl.Datetime(time_zone="UTC"),
+        },
+    )
+
+    # Act
+    result_df = apply_scd_type_2(
+        source_df=source_df,
+        target_df=target_df,
+        primary_key_columns="id",
+        valid_from_column="valid_from",
+        valid_to_column="valid_to",
+    )
+
+    # Assert
+    expected_df = pl.DataFrame(
+        {
+            "id": [1, 1, 2],
+            "name": ["Alice", "Alice Updated", "Bob"],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                None,
+                None,
+            ],
+        }
+    )
+
+    assert_frame_equal(result_df, expected_df, check_row_order=False)
+
+
+def test_apply_scd_type_2_empty_target():
+    # Arrange
+    schema = {
+        "id": pl.Int64,
+        "name": pl.Utf8,
+        "valid_from": pl.Datetime(time_zone="UTC"),
+        "valid_to": pl.Datetime(time_zone="UTC"),
+    }
+
+    source_df = pl.DataFrame(
+        {
+            "id": [1],
+            "name": ["Alice"],
+            "valid_from": [datetime(2024, 1, 1, tzinfo=timezone.utc)],
+            "valid_to": [None],
+        },
+        schema=schema,
+    )
+
+    target_df = pl.DataFrame(
+        {"id": [], "name": [], "valid_from": [], "valid_to": []}, schema=schema
+    )
+
+    # Act
+    result_df = apply_scd_type_2(
+        source_df=source_df,
+        target_df=target_df,
+        primary_key_columns="id",
+        valid_from_column="valid_from",
+        valid_to_column="valid_to",
+    )
+
+    # Assert
+    expected_df = source_df
+
+    assert_frame_equal(result_df, expected_df, check_row_order=False)
+
+
+def test_apply_scd_type_2_multiple_primary_keys():
+    # Arrange
+    source_df = pl.DataFrame(
+        {
+            "customer_id": [1, 1],
+            "product_id": [100, 101],
+            "quantity": [5, 10],
+            "valid_from": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None],
+        }
+    )
+
+    target_df = pl.DataFrame(
+        {
+            "customer_id": [1],
+            "product_id": [100],
+            "quantity": [3],
+            "valid_from": [datetime(2023, 1, 1, tzinfo=timezone.utc)],
+            "valid_to": [None],
+        }
+    )
+
+    # Act
+    result_df = apply_scd_type_2(
+        source_df=source_df,
+        target_df=target_df,
+        primary_key_columns=["customer_id", "product_id"],
+        valid_from_column="valid_from",
+        valid_to_column="valid_to",
+    )
+
+    # Assert
+    expected_df = pl.DataFrame(
+        {
+            "customer_id": [1, 1, 1],
+            "product_id": [100, 100, 101],
+            "quantity": [3, 5, 10],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [datetime(2024, 1, 1, tzinfo=timezone.utc), None, None],
+        }
+    )
+
+    assert_frame_equal(result_df, expected_df, check_row_order=False)
+
+
+def test_apply_scd_type_2_target_records_gets_updated():
+    # Arrange
+    schema = {
+        "id": pl.Int64,
+        "name": pl.Utf8,
+        "valid_from": pl.Datetime(time_zone="UTC"),
+        "valid_to": pl.Datetime(time_zone="UTC"),
+    }
+
+    source_df = pl.DataFrame(
+        {
+            "id": [1, 1, 1, 1],
+            "name": ["Alice 3", "Alice 1", "Alice 5", "Alice 6"],
+            "valid_from": [
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2025, 1, 1, tzinfo=timezone.utc),
+                datetime(2026, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, None, None, None],
+        },
+        schema=schema,
+    )
+
+    target_df = pl.DataFrame(
+        {
+            "id": [1, 1],
+            "name": ["Alice 4", "Alice 2"],
+            "valid_from": [
+                datetime(2024, 6, 1, tzinfo=timezone.utc),
+                datetime(2023, 6, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [None, datetime(2024, 6, 1, tzinfo=timezone.utc)],
+        },
+        schema=schema,
+    )
+
+    # Act
+    result_df = apply_scd_type_2(
+        source_df=source_df,
+        target_df=target_df,
+        primary_key_columns="id",
+        valid_from_column="valid_from",
+        valid_to_column="valid_to",
+    )
+
+    # Assert
+    expected_df = pl.DataFrame(
+        {
+            "id": [1, 1, 1, 1, 1, 1],
+            "name": ["Alice 1", "Alice 2", "Alice 3", "Alice 4", "Alice 5", "Alice 6"],
+            "valid_from": [
+                datetime(2023, 1, 1, tzinfo=timezone.utc),
+                datetime(2023, 6, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 6, 1, tzinfo=timezone.utc),
+                datetime(2025, 1, 1, tzinfo=timezone.utc),
+                datetime(2026, 1, 1, tzinfo=timezone.utc),
+            ],
+            "valid_to": [
+                datetime(2023, 6, 1, tzinfo=timezone.utc),
+                datetime(2024, 1, 1, tzinfo=timezone.utc),
+                datetime(2024, 6, 1, tzinfo=timezone.utc),
+                datetime(2025, 1, 1, tzinfo=timezone.utc),
+                datetime(2026, 1, 1, tzinfo=timezone.utc),
+                None,
+            ],
+        }
+    )
+
+    assert_frame_equal(result_df, expected_df, check_row_order=False)