feat: build only what changed from consumed-input records

Makefile

@@ -145,6 +145,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/own_change_replay \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_append_cursor_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_capped_microbatch_build.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_changed_builds.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_concurrent_microbatch_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_cursor_runtime_failures.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_dropped_relation_rebuild.py \

benchmarks/duckshop/edge_cases/changed_builds/models/customer_orders.sql

@@ -0,0 +1,5 @@
+MODEL (description "Orders per customer, two models downstream of both sources.", materialized table);
+SELECT c.customer_id, count(o.order_id) AS orders
+FROM __ref("stg_customers") AS c
+LEFT JOIN __ref("stg_orders") AS o USING (customer_id)
+GROUP BY c.customer_id

benchmarks/duckshop/edge_cases/changed_builds/models/stg_customers.sql

@@ -0,0 +1,2 @@
+MODEL (description "Staged customers.", materialized table);
+SELECT customer_id FROM __source("raw_customers")

benchmarks/duckshop/edge_cases/changed_builds/models/stg_orders.sql

@@ -0,0 +1,2 @@
+MODEL (description "Staged orders.", materialized table);
+SELECT order_id, customer_id FROM __source("raw_orders")

benchmarks/duckshop/edge_cases/changed_builds/sources/raw.yml

@@ -0,0 +1,17 @@
+sources:
+  - name: raw_orders
+    description: Synthetic orders, numbered by load batch.
+    schema: raw
+    table: orders
+    freshness:
+      strategy: column
+      column: load_id
+      type: integer
+  - name: raw_customers
+    description: Synthetic customers, numbered by load batch.
+    schema: raw
+    table: customers
+    freshness:
+      strategy: column
+      column: load_id
+      type: integer

benchmarks/duckshop/edge_cases/changed_builds_udf/functions/sql/table_fn__customer_order_keys.sql

@@ -0,0 +1,9 @@
+FUNCTION (
+  description "Order keys of one customer, read from staged orders",
+  arguments (p_customer_id INTEGER),
+  returns table (order_key INTEGER, customer_id INTEGER)
+);
+
+SELECT o.order_key, o.customer_id
+FROM __ref("stg_orders") AS o
+WHERE o.customer_id = p_customer_id

benchmarks/duckshop/edge_cases/changed_builds_udf/functions/sql/table_fn__raw_customers_from.sql

@@ -0,0 +1,9 @@
+FUNCTION (
+  description "Customers from an id on, read straight from the raw table without declaring it",
+  arguments (p_customer_id INTEGER),
+  returns table (customer_id INTEGER)
+);
+
+SELECT c.customer_id
+FROM raw.customers AS c
+WHERE c.customer_id >= p_customer_id

benchmarks/duckshop/edge_cases/changed_builds_udf/functions/sql/udf__order_key.sql

@@ -0,0 +1,7 @@
+FUNCTION (
+  description "Display key of an order, ten times its number",
+  arguments (p_order_id INTEGER),
+  returns INTEGER,
+);
+
+p_order_id * 10

benchmarks/duckshop/edge_cases/changed_builds_udf/functions/sql/udf__raw_customer_count.sql

@@ -0,0 +1,7 @@
+FUNCTION (
+  description "Number of customers from an id on, counted straight from the raw table without declaring it",
+  arguments (p_customer_id INTEGER),
+  returns BIGINT,
+);
+
+(SELECT count(*) FROM raw.customers AS c WHERE c.customer_id >= p_customer_id)

benchmarks/duckshop/edge_cases/changed_builds_udf/models/customer_count_via_udf.sql

@@ -0,0 +1,2 @@
+MODEL (description "Customer count read only through a SQL function whose body reads an undeclared raw table.", materialized table);
+SELECT __udf("udf__raw_customer_count")(1) AS customer_count

benchmarks/duckshop/edge_cases/changed_builds_udf/models/customers_via_table_fn.sql

@@ -0,0 +1,3 @@
+MODEL (description "Customers read only through a table function whose body reads an undeclared raw table.", materialized table);
+SELECT k.customer_id
+FROM __table_fn("table_fn__raw_customers_from")(1) AS k

benchmarks/duckshop/edge_cases/changed_builds_udf/models/first_customer_order_keys.sql

@@ -0,0 +1,3 @@
+MODEL (description "Order keys of the first customer, read only through a table function.", materialized table);
+SELECT k.order_key
+FROM __table_fn("table_fn__customer_order_keys")(1) AS k

benchmarks/duckshop/edge_cases/changed_builds_udf/models/stg_orders.sql

@@ -0,0 +1,3 @@
+MODEL (description "Staged orders with a display key from a SQL function.", materialized table);
+SELECT order_id, customer_id, __udf("udf__order_key")(order_id) AS order_key
+FROM __source("raw_orders")

benchmarks/duckshop/edge_cases/changed_builds_unqualified/models/customers.sql

@@ -0,0 +1,2 @@
+MODEL (description "Customers in the target schema, sharing a table name with the raw customers source.", materialized table);
+SELECT customer_id FROM __ref("stg_customers")

benchmarks/duckshop/edge_cases/changed_builds_unqualified/models/unqualified_customer_reader.sql

@@ -0,0 +1,5 @@
+MODEL (description "Declares the raw customers source but also reads the customers model by its unqualified name.", materialized table);
+SELECT r.customer_id, count(c.customer_id) AS modelled
+FROM __source("raw_customers") AS r
+LEFT JOIN customers AS c USING (customer_id)
+GROUP BY r.customer_id

benchmarks/duckshop/edge_cases/orchestration/consumption_history.json

@@ -0,0 +1,70 @@
+{
+  "record": {
+    "state_database": null,
+    "state_schema": "main",
+    "event_id": "consumption-000",
+    "execution_id": "0190a1b2-0000-7000-8000-000000000001",
+    "phase": "succeeded",
+    "run_id": "external-build",
+    "node_type": "model",
+    "node_name": "customer_orders",
+    "target_database": null,
+    "target_schema": "main",
+    "target_name": "customer_orders_🧾",
+    "version_hash": "customer-orders-version",
+    "input_count": 2,
+    "input_kind": "source",
+    "input_name": "shop.orders",
+    "input_version": "load 'orders' 🧾 #7",
+    "recorded_at": "2026-02-03T04:05:06Z"
+  },
+  "source_rows": 104,
+  "duplicate_rows": 3,
+  "later_time": "2026-02-03T04:05:06.001Z",
+  "retained_result": "consumption-103",
+  "kind": "consumption",
+  "source_key": "event_id",
+  "timestamp_key": "recorded_at",
+  "id_prefix": "consumption",
+  "page_lengths": [
+    100,
+    7
+  ],
+  "later_index": 106,
+  "later_record": {
+    "event_id": "consumption-103",
+    "phase": "started",
+    "execution_id": "0190a1b2-0000-7000-8000-000000000002",
+    "input_kind": null,
+    "input_name": null,
+    "input_version": null,
+    "recorded_at": "2026-02-03T04:05:06.001Z"
+  },
+  "last_record": {
+    "event_id": "consumption-103",
+    "phase": "started",
+    "execution_id": "0190a1b2-0000-7000-8000-000000000002",
+    "input_kind": null,
+    "input_name": null,
+    "input_version": null,
+    "recorded_at": "2026-02-03T04:05:06.001Z"
+  },
+  "record_overrides": {
+    "consumption-102": {
+      "input_kind": "node",
+      "input_name": "model.stg_customers",
+      "input_version": "0190a1b2-0000-7000-8000-0000000000aa"
+    },
+    "consumption-103": {
+      "event_id": "consumption-103",
+      "phase": "started",
+      "execution_id": "0190a1b2-0000-7000-8000-000000000002",
+      "input_kind": null,
+      "input_name": null,
+      "input_version": null,
+      "recorded_at": "2026-02-03T04:05:06.001Z"
+    }
+  },
+  "source_count": 107,
+  "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id"
+}

clients/typescript/src/schema.d.ts

@@ -499,6 +499,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/locations/{location_id}/targets/{target}/metadata/consumption": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        post: operations["report_consumption"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/locations/{location_id}/targets/{target}/metadata/fingerprints": {
         parameters: {
             query?: never;
@@ -997,6 +1013,14 @@ export interface components {
             /** Format: uuid */
             request_id: string;
         };
+        ConsumptionBatch: {
+            /** Format: int64 */
+            generation: number;
+            reconciliation?: null | components["schemas"]["WarehouseReconciliationPage"];
+            records: components["schemas"]["WarehouseConsumptionRecord"][];
+            /** Format: uuid */
+            session_id: string;
+        };
         /** @description The current successful snapshot, read with a consistent generation and change set. */
         CurrentSnapshot: {
             changes: components["schemas"]["SnapshotChange"][];
@@ -1929,6 +1953,28 @@ export interface components {
             state_database?: string | null;
             state_schema: string;
         };
+        /** @description What one execution consumed, as history; `--changed` reads the warehouse table, never this mirror. */
+        WarehouseConsumptionRecord: {
+            event_id: string;
+            execution_id: string;
+            /** Format: int64 */
+            input_count: number;
+            input_kind?: string | null;
+            input_name?: string | null;
+            input_version?: string | null;
+            node_name: string;
+            node_type: string;
+            phase: string;
+            /** Format: date-time */
+            recorded_at: string;
+            run_id: string;
+            state_database?: string | null;
+            state_schema: string;
+            target_database?: string | null;
+            target_name: string;
+            target_schema: string;
+            version_hash: string;
+        };
         /** @description Fingerprints remain history here; latest-row selection belongs to the existing planner reader. */
         WarehouseFingerprintRecord: {
             definition: string;
@@ -1969,7 +2015,7 @@ export interface components {
          * @description Catalog observations require full scans; only immutable history supports a watermark.
          * @enum {string}
          */
-        WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "catalog" | "statistics";
+        WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "consumption" | "catalog" | "statistics";
         /** @description Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. */
         WarehouseJanitorRecord: {
             archive_name: string;
@@ -3437,6 +3483,54 @@ export interface operations {
             };
         };
     };
+    report_consumption: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["ConsumptionBatch"];
+            };
+        };
+        responses: {
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["WarehouseSyncResponse"];
+                };
+            };
+            400: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+            /** @description Invalid credential */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Location not authorized */
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+        };
+    };
     report_fingerprints: {
         parameters: {
             query?: never;

clients/typescript/tests/types.ts

@@ -67,6 +67,14 @@ void client.POST("/api/locations/{location_id}/targets/{target}/metadata/microba
   params: { path: { location_id: "synthetic-location", target: "development" } },
   body: { session_id: "synthetic-session", generation: 1, records: [] },
 });
+const consumptionHistory: components["schemas"]["WarehouseHistoryKind"] = "consumption";
+const startedInput: components["schemas"]["WarehouseConsumptionRecord"]["input_version"] = null;
+void consumptionHistory;
+void startedInput;
+void client.POST("/api/locations/{location_id}/targets/{target}/metadata/consumption", {
+  params: { path: { location_id: "synthetic-location", target: "development" } },
+  body: { session_id: "synthetic-session", generation: 1, records: [] },
+});
 const catalogHistory: components["schemas"]["WarehouseHistoryKind"] = "catalog";
 const catalogColumn: components["schemas"]["WarehouseCatalogColumn"] = {
   name: "order_id", data_type: "INTEGER",

crates/sqlbuild-analysis/src/janitor_planning/constants.rs

@@ -1,16 +1,18 @@
 //! Janitor planning names, prefixes and skip reasons.
 
 use sqlbuild_core::constants::COLUMN_MIGRATION_TABLE_NAME;
+use sqlbuild_core::constants::CONSUMPTION_TABLE_NAME;
 use sqlbuild_core::constants::FINGERPRINT_TABLE_NAME;
 use sqlbuild_core::constants::MIGRATION_TABLE_NAME;
 use sqlbuild_core::constants::OLD_NAME_VIEW_TABLE_NAME;
 use sqlbuild_core::constants::SOURCE_FRESHNESS_TABLE_NAME;
 
 /// SQLBuild state tables the janitor never acts on.
-pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 9] = [
+pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 10] = [
     FINGERPRINT_TABLE_NAME,
     "_sqlbuild_microbatches",
     SOURCE_FRESHNESS_TABLE_NAME,
+    CONSUMPTION_TABLE_NAME,
     "_sqlbuild_node_results",
     "_sqlbuild_audit_results",
     "_sqlbuild_janitor_events",

crates/sqlbuild-cli/src/argv/command_table.json

@@ -2360,6 +2360,17 @@
         "default": false,
         "group": 1
        },
+       {
+        "kind": "store_true",
+        "options": [
+         "--changed"
+        ],
+        "dest": "changed",
+        "nargs": 0,
+        "const": true,
+        "default": false,
+        "group": 1
+       },
        {
         "kind": "store_true",
         "options": [

crates/sqlbuild-cli/src/argv/constants.rs

@@ -71,7 +71,7 @@ pub(crate) const INT_CONVERTER: &str = "int";
 pub(crate) const PATH_CONVERTER: &str = "Path";
 
 /// Every `CliNamespace` attribute with a default, in declaration order.
-pub(crate) const CLI_NAMESPACE_DEFAULTS: [(&str, NamespaceDefault); 184] = [
+pub(crate) const CLI_NAMESPACE_DEFAULTS: [(&str, NamespaceDefault); 185] = [
     ("command", NamespaceDefault::None),
     ("project_dir", NamespaceDefault::None),
     ("dbt_project_dir", NamespaceDefault::None),
@@ -126,6 +126,7 @@ pub(crate) const CLI_NAMESPACE_DEFAULTS: [(&str, NamespaceDefault); 184] = [
     ("no_color", NamespaceDefault::Bool(false)),
     ("fail_fast", NamespaceDefault::Bool(false)),
     ("full_refresh", NamespaceDefault::Bool(false)),
+    ("changed", NamespaceDefault::Bool(false)),
     ("as_target", NamespaceDefault::None),
     ("include_python", NamespaceDefault::Bool(true)),
     ("load_sources", NamespaceDefault::None),

crates/sqlbuild-cli/src/dispatch/constants.rs

@@ -139,7 +139,7 @@ pub(crate) const PLAN_ROUTE: Route = routed(
     REQUEST_KEYWORD_SHAPE,
 );
 
-const BUILD_FIELDS: [Field; 35] = [
+const BUILD_FIELDS: [Field; 36] = [
     PROJECT_DIR,
     NO_SQL_VALIDATION,
     same("no_cache"),
@@ -151,6 +151,7 @@ const BUILD_FIELDS: [Field; 35] = [
     NO_COLOR,
     same("fail_fast"),
     same("full_refresh"),
+    same("changed"),
     same("load_sources"),
     attr("reload_sources", "reload"),
     same("include_python"),