Scd2 incremental key column (#1712)

* Add PostgreSQL and Snowflake unit tests for scd2_by_column with incremental_key

Co-authored-by: Sabri Karagönen <[email protected]>

* Add unit tests for scd2_by_column with incremental_key across all platforms

Co-authored-by: Sabri Karagönen <[email protected]>

* Update documentation for scd2_by_column incremental_key feature

Co-authored-by: Sabri Karagönen <[email protected]>

* Fix dupword lint warnings in test files

Co-authored-by: Sabri Karagönen <[email protected]>

* Add integration tests for scd2_by_column with incremental_key using DuckDB

Co-authored-by: Sabri Karagönen <[email protected]>

* Allow incremental_key with scd2_by_column strategy in linter validation

Co-authored-by: Sabri Karagönen <[email protected]>

* Add scd2-by-column-incremental-key connection to default environment

Co-authored-by: Sabri Karagönen <[email protected]>

* Fix timestamp format in integration test expectations

Co-authored-by: Sabri Karagönen <[email protected]>

* Update expected connections JSON to include scd2-by-column-incremental-key connection

Co-authored-by: Sabri Karagönen <[email protected]>

---------

Co-authored-by: Cursor Agent <[email protected]>

Author: sabrikaragonen

docs/assets/materialization.md

@@ -435,10 +435,28 @@ When changes are detected in non-primary key columns:
 
 **Automatically added columns:**
 
-- `_valid_from`: TIMESTAMP when the record version became active (set to `CURRENT_TIMESTAMP()`)
-- `_valid_until`: TIMESTAMP when the record version became inactive (set to `TIMESTAMP('9999-12-31')` for current records)
+- `_valid_from`: TIMESTAMP when the record version became active (defaults to `CURRENT_TIMESTAMP()`, or uses `incremental_key` value if specified)
+- `_valid_until`: TIMESTAMP when the record version became inactive (set to `TIMESTAMP('9999-12-31')` for current records, or uses `incremental_key` value when a record is expired due to changes)
 - `_is_current`: BOOLEAN indicating if this is the current version of the record
 
+**Optional: Using `incremental_key` for timestamps:**
+
+By default, `_valid_from` and `_valid_until` are set using `CURRENT_TIMESTAMP()`. However, if your source data has a column that indicates when changes actually occurred (e.g., an `updated_at` timestamp), you can specify it using the `incremental_key` option:
+
+```yaml
+materialization:
+  type: table
+  strategy: scd2_by_column
+  incremental_key: updated_at
+```
+
+When `incremental_key` is specified:
+- `_valid_from` for new/updated records will be set to the value of the `incremental_key` column
+- `_valid_until` for records being expired (due to changes) will be set to the value of the `incremental_key` column from the new record
+- Records expiring because they're no longer in the source data will still use `CURRENT_TIMESTAMP()` for `_valid_until`
+
+This is useful when you want the SCD2 timeline to reflect the actual business timestamps from your source data rather than the processing time.
+
 **NOTE:***
 
 - Unless otherwise specified by `partition_by`, the SCD2 table will be partitioned by `_valid_from` for platforms which support partitioning (BigQuery, Athena, Snowflake).
@@ -475,6 +493,43 @@ UNION ALL
 SELECT 3 AS ID, 'Keyboard' AS Name, 89.99 AS Price
 ```
 
+**Example with `incremental_key`:**
+
+When you want `_valid_from` and `_valid_until` to reflect actual business timestamps instead of processing time:
+
+```bruin-sql
+/* @bruin
+name: test.product_catalog
+type: bq.sql
+
+materialization:
+  type: table
+  strategy: scd2_by_column
+  incremental_key: updated_at
+
+columns:
+  - name: ID
+    type: INTEGER
+    description: "Unique identifier for Product"
+    primary_key: true
+  - name: Name
+    type: VARCHAR
+    description: "Name of the Product"
+  - name: Price
+    type: FLOAT
+    description: "Price of the Product"
+  - name: updated_at
+    type: TIMESTAMP
+    description: "When the product was last modified in the source system"
+@bruin */
+
+SELECT 1 AS ID, 'Wireless Mouse' AS Name, 29.99 AS Price, TIMESTAMP '2024-01-15 10:30:00' AS updated_at
+UNION ALL
+SELECT 2 AS ID, 'USB Cable' AS Name, 12.99 AS Price, TIMESTAMP '2024-01-14 14:00:00' AS updated_at
+```
+
+In this case, `_valid_from` will be set to the `updated_at` value from each record, preserving the actual business timeline of when changes occurred.
+
 **Example behavior:**
 
 Let's say you want to create a new table to track product catalog with SCD2. If the table doesn't exist yet, you'll need an initial run with the `--full-refresh` flag:
@@ -628,9 +683,9 @@ Notice how:
 | Aspect | scd2_by_column | scd2_by_time |
 |--------|----------------|--------------|
 | **Change Detection** | Automatically detects changes in any non-primary key column | Based on time values in the incremental_key column |
-| **_valid_from Value** | Set to `CURRENT_TIMESTAMP()` when change is processed | Derived from the incremental_key column value |
-| **Use Case** | When you want to track any column changes regardless of when they occurred | When your source data has reliable timestamps indicating when changes happened |
-| **Configuration** | Only requires primary_key columns | Requires both primary_key columns and incremental_key |
+| **_valid_from Value** | Set to `CURRENT_TIMESTAMP()` by default, or uses `incremental_key` value if specified | Always derived from the incremental_key column value |
+| **Use Case** | When you want to track any column changes; optionally use `incremental_key` for business timestamps | When your source data has reliable timestamps indicating when changes happened |
+| **Configuration** | Only requires primary_key columns; `incremental_key` is optional | Requires both primary_key columns and incremental_key |
 
 > [!WARNING]
 > SCD2 materializations are currently only supported for BigQuery, Snowflake, Postgres, Amazon Redshift, MySQL, DuckDB, and Databricks.

integration-tests/.bruin.yml

@@ -11,6 +11,8 @@ environments:
           path: "duckdb-files/scd2-by-column.db"
         - name: "duckdb-scd2-by-time"
           path: "duckdb-files/scd2-by-time.db"
+        - name: "duckdb-scd2-by-column-incremental-key"
+          path: "duckdb-files/scd2-by-column-incremental-key.db"
         - name: "duckdb-start-date-flags"
           path: "duckdb-files/start-date-flags.db"
         - name: "duckdb-mat-test"
@@ -244,6 +246,12 @@ environments:
         - name: "duckdb-scd2-by-time"
           path: "duckdb-files/scd2-by-time.db"
 
+  env-scd2-by-column-incremental-key:
+    connections:
+      duckdb:
+        - name: "duckdb-scd2-by-column-incremental-key"
+          path: "duckdb-files/scd2-by-column-incremental-key.db"
+
   env-duckdb-decimal:
     connections:
       duckdb:

integration-tests/expectations/expected_connections.json

@@ -26,6 +26,10 @@
           "name": "duckdb-scd2-by-time",
           "path": "duckdb-files/scd2-by-time.db"
         },
+        {
+          "name": "duckdb-scd2-by-column-incremental-key",
+          "path": "duckdb-files/scd2-by-column-incremental-key.db"
+        },
         {
           "name": "duckdb-start-date-flags",
           "path": "duckdb-files/start-date-flags.db"
@@ -77,6 +81,10 @@
             "name": "duckdb-scd2-by-time",
             "path": "duckdb-files/scd2-by-time.db"
           },
+          {
+            "name": "duckdb-scd2-by-column-incremental-key",
+            "path": "duckdb-files/scd2-by-column-incremental-key.db"
+          },
           {
             "name": "duckdb-start-date-flags",
             "path": "duckdb-files/start-date-flags.db"
@@ -494,6 +502,17 @@
       },
       "schema_prefix": ""
     },
+    "env-scd2-by-column-incremental-key": {
+      "connections": {
+        "duckdb": [
+          {
+            "name": "duckdb-scd2-by-column-incremental-key",
+            "path": "duckdb-files/scd2-by-column-incremental-key.db"
+          }
+        ]
+      },
+      "schema_prefix": ""
+    },
     "env-scd2-by-time": {
       "connections": {
         "duckdb": [

integration-tests/integration_test.go

@@ -2168,6 +2168,147 @@ func TestWorkflowTasks(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "run_pipeline_with_scd2_by_column_incremental_key",
+			workflow: e2e.Workflow{
+				Name: "run_pipeline_with_scd2_by_column_incremental_key",
+				Steps: []e2e.Task{
+					{
+						Name:    "scd2-col-ik-01a: create test directory",
+						Command: "mkdir",
+						Args:    []string{"-p", filepath.Join(tempdir, "test-scd2-by-column-incremental-key")},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:       "scd2-col-ik-01b: initialize git repository",
+						Command:    "git",
+						Args:       []string{"init"},
+						WorkingDir: filepath.Join(tempdir, "test-scd2-by-column-incremental-key"),
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:       "scd2-col-ik-01c: copy pipeline files",
+						Command:    "cp",
+						Args:       []string{"-a", filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/scd2-by-column-incremental-key-pipeline"), "."},
+						WorkingDir: filepath.Join(tempdir, "test-scd2-by-column-incremental-key"),
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-02: run pipeline with full refresh",
+						Command: binary,
+						Args:    []string{"run", "--full-refresh", "--config-file", filepath.Join(currentFolder, ".bruin.yml"), "--env", "env-scd2-by-column-incremental-key", filepath.Join(tempdir, "test-scd2-by-column-incremental-key/scd2-by-column-incremental-key-pipeline")},
+						Env:     []string{},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-03: query the initial table",
+						Command: binary,
+						Args:    []string{"query", "--connection", "duckdb-scd2-by-column-incremental-key", "--query", "SELECT product_id, product_name, price, _is_current, _valid_from FROM test.products ORDER BY product_id, _valid_from;", "--output", "csv"},
+						Env:     []string{},
+						Expected: e2e.Output{
+							ExitCode: 0,
+							CSVFile:  filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/scd2-by-column-incremental-key-pipeline/expectations/scd2_by_col_ik_expected_initial.csv"),
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+							e2e.AssertByCSV,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-04a: copy products_incremental_key_updated_01.sql",
+						Command: "cp",
+						Args:    []string{filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/resources/products_incremental_key_updated_01.sql"), filepath.Join(tempdir, "test-scd2-by-column-incremental-key/scd2-by-column-incremental-key-pipeline/assets/products.sql")},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-04b: run pipeline with updated products",
+						Command: binary,
+						Args:    []string{"run", "--config-file", filepath.Join(currentFolder, ".bruin.yml"), "--env", "env-scd2-by-column-incremental-key", filepath.Join(tempdir, "test-scd2-by-column-incremental-key/scd2-by-column-incremental-key-pipeline")},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-05: query the updated table 01",
+						Command: binary,
+						Args:    []string{"query", "--connection", "duckdb-scd2-by-column-incremental-key", "--query", "SELECT product_id, product_name, price, _is_current, _valid_from FROM test.products ORDER BY product_id, _valid_from;", "--output", "csv"},
+						Env:     []string{},
+						Expected: e2e.Output{
+							ExitCode: 0,
+							CSVFile:  filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/scd2-by-column-incremental-key-pipeline/expectations/scd2_by_col_ik_expected_updated_01.csv"),
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+							e2e.AssertByCSV,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-06a: copy products_incremental_key_updated_02.sql",
+						Command: "cp",
+						Args:    []string{filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/resources/products_incremental_key_updated_02.sql"), filepath.Join(tempdir, "test-scd2-by-column-incremental-key/scd2-by-column-incremental-key-pipeline/assets/products.sql")},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-06b: run pipeline with updated products 02",
+						Command: binary,
+						Args:    []string{"run", "--config-file", filepath.Join(currentFolder, ".bruin.yml"), "--env", "env-scd2-by-column-incremental-key", filepath.Join(tempdir, "test-scd2-by-column-incremental-key/scd2-by-column-incremental-key-pipeline")},
+						Expected: e2e.Output{
+							ExitCode: 0,
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+						},
+					},
+					{
+						Name:    "scd2-col-ik-07: query the updated table 02",
+						Command: binary,
+						Args:    []string{"query", "--connection", "duckdb-scd2-by-column-incremental-key", "--query", "SELECT product_id, product_name, price, _is_current, _valid_from FROM test.products ORDER BY product_id, _valid_from;", "--output", "csv"},
+						Env:     []string{},
+						Expected: e2e.Output{
+							ExitCode: 0,
+							CSVFile:  filepath.Join(currentFolder, "test-pipelines/duckdb-scd2-tests/scd2-by-column-incremental-key-pipeline/expectations/scd2_by_col_ik_expected_updated_02.csv"),
+						},
+						Asserts: []func(*e2e.Task) error{
+							e2e.AssertByExitCode,
+							e2e.AssertByCSV,
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "start_date_flags_workflow",
 			workflow: e2e.Workflow{

integration-tests/test-pipelines/duckdb-scd2-tests/resources/products_incremental_key_original.sql

@@ -0,0 +1,29 @@
+/* @bruin
+name: test.products
+type: duckdb.sql
+materialization:
+  type: table
+  strategy: scd2_by_column
+  incremental_key: updated_at
+
+columns:
+  - name: product_id
+    type: INTEGER
+    description: "Unique identifier for Product"
+    primary_key: true
+  - name: product_name
+    type: VARCHAR
+    description: "Name of the Product"
+  - name: price
+    type: FLOAT
+    description: "Price of the Product"
+  - name: updated_at
+    type: TIMESTAMP
+    description: "When the product was last updated"
+@bruin */
+
+SELECT 1 AS product_id, 'Laptop' AS product_name, 999.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at
+UNION ALL
+SELECT 2 AS product_id, 'Mouse' AS product_name, 29.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at
+UNION ALL
+SELECT 3 AS product_id, 'Keyboard' AS product_name, 79.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at

integration-tests/test-pipelines/duckdb-scd2-tests/resources/products_incremental_key_updated_01.sql

@@ -0,0 +1,30 @@
+/* @bruin
+name: test.products
+type: duckdb.sql
+materialization:
+  type: table
+  strategy: scd2_by_column
+  incremental_key: updated_at
+
+columns:
+  - name: product_id
+    type: INTEGER
+    description: "Unique identifier for Product"
+    primary_key: true
+  - name: product_name
+    type: VARCHAR
+    description: "Name of the Product"
+  - name: price
+    type: FLOAT
+    description: "Price of the Product"
+  - name: updated_at
+    type: TIMESTAMP
+    description: "When the product was last updated"
+@bruin */
+
+-- Update 1: Laptop price changed from 999.99 to 1099.99 at 2024-02-01
+SELECT 1 AS product_id, 'Laptop' AS product_name, 1099.99 AS price, TIMESTAMP '2024-02-01 14:30:00' AS updated_at
+UNION ALL
+SELECT 2 AS product_id, 'Mouse' AS product_name, 29.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at
+UNION ALL
+SELECT 3 AS product_id, 'Keyboard' AS product_name, 79.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at

integration-tests/test-pipelines/duckdb-scd2-tests/resources/products_incremental_key_updated_02.sql

@@ -0,0 +1,30 @@
+/* @bruin
+name: test.products
+type: duckdb.sql
+materialization:
+  type: table
+  strategy: scd2_by_column
+  incremental_key: updated_at
+
+columns:
+  - name: product_id
+    type: INTEGER
+    description: "Unique identifier for Product"
+    primary_key: true
+  - name: product_name
+    type: VARCHAR
+    description: "Name of the Product"
+  - name: price
+    type: FLOAT
+    description: "Price of the Product"
+  - name: updated_at
+    type: TIMESTAMP
+    description: "When the product was last updated"
+@bruin */
+
+-- Update 2: Keyboard removed, Monitor added at 2024-03-01
+SELECT 1 AS product_id, 'Laptop' AS product_name, 1099.99 AS price, TIMESTAMP '2024-02-01 14:30:00' AS updated_at
+UNION ALL
+SELECT 2 AS product_id, 'Mouse' AS product_name, 29.99 AS price, TIMESTAMP '2024-01-15 10:00:00' AS updated_at
+UNION ALL
+SELECT 4 AS product_id, 'Monitor' AS product_name, 299.99 AS price, TIMESTAMP '2024-03-01 09:00:00' AS updated_at