Add ability to compare changes in CI jobs

docs/data-sources/job.md

@@ -29,6 +29,7 @@ description: |-
 - `id` (String) The ID of this resource.
 - `job_completion_trigger_condition` (Set of Object) Which other job should trigger this job when it finishes, and on which conditions. (see [below for nested schema](#nestedatt--job_completion_trigger_condition))
 - `name` (String) Given name for the job
+- `run_compare_changes` (Boolean) Whether the CI job should compare data changes introduced by the code change in the PR.
 - `self_deferring` (Boolean) Whether this job defers on a previous run of itself (overrides value in deferring_job_id)
 - `timeout_seconds` (Number) Number of seconds before the job times out
 - `triggers` (Map of Boolean) Flags for which types of triggers to use, keys of github_webhook, git_provider_webhook, schedule, on_merge

docs/data-sources/jobs.md

@@ -61,6 +61,7 @@ Read-Only:
 - `job_type` (String) The type of job (e.g. CI, scheduled)
 - `name` (String) The name of the job
 - `project_id` (Number) The ID of the project
+- `run_compare_changes` (Boolean) Whether the job should compare data changes introduced by the code change in the PR
 - `run_generate_sources` (Boolean) Whether the job test source freshness
 - `schedule` (Attributes) (see [below for nested schema](#nestedatt--jobs--schedule))
 - `settings` (Attributes) (see [below for nested schema](#nestedatt--jobs--settings))

docs/resources/job.md

@@ -122,6 +122,7 @@ resource "dbtcloud_job" "downstream_job" {
 - `is_active` (Boolean) Should always be set to true as setting it to false is the same as creating a job in a deleted state. To create/keep a job in a 'deactivated' state, check  the `triggers` config.
 - `job_completion_trigger_condition` (Block Set, Max: 1) Which other job should trigger this job when it finishes, and on which conditions (sometimes referred as 'job chaining'). (see [below for nested schema](#nestedblock--job_completion_trigger_condition))
 - `num_threads` (Number) Number of threads to use in the job
+- `run_compare_changes` (Boolean) Whether the CI job should compare data changes introduced by the code changes. Requires `deferring_environment_id` to be set. (Advanced CI needs to be activated in the dbt Cloud Account Settings first as well)
 - `run_generate_sources` (Boolean) Flag for whether the job should add a `dbt source freshness` step to the job. The difference between manually adding a step with `dbt source freshness` in the job steps or using this flag is that with this flag, a failed freshness will still allow the following steps to run.
 - `schedule_cron` (String) Custom cron expression for schedule
 - `schedule_days` (List of Number) List of days of week as numbers (0 = Sunday, 7 = Saturday) to execute the job at if running on a schedule

pkg/dbt_cloud/job.go

@@ -78,6 +78,7 @@ type Job struct {
 	Execution              JobExecution          `json:"execution"`
 	TriggersOnDraftPR      bool                  `json:"triggers_on_draft_pr"`
 	JobCompletionTrigger   *JobCompletionTrigger `json:"job_completion_trigger_condition"`
+	RunCompareChanges      bool                  `json:"run_compare_changes"`
 }
 
 type JobWithEnvironment struct {
@@ -133,6 +134,7 @@ func (c *Client) CreateJob(
 	timeoutSeconds int,
 	triggersOnDraftPR bool,
 	jobCompletionTriggerCondition map[string]any,
+	runCompareChanges bool,
 ) (*Job, error) {
 	state := STATE_ACTIVE
 	if !isActive {
@@ -232,6 +234,7 @@ func (c *Client) CreateJob(
 		TriggersOnDraftPR:    triggersOnDraftPR,
 		JobCompletionTrigger: jobCompletionTrigger,
 		JobType:              jobType,
+		RunCompareChanges:    runCompareChanges,
 	}
 	if dbtVersion != "" {
 		newJob.Dbt_Version = &dbtVersion

pkg/framework/objects/job/data_source_all.go

@@ -131,6 +131,7 @@ func (d *jobsDataSource) Read(
 			},
 			JobType:           types.StringValue(job.JobType),
 			TriggersOnDraftPr: types.BoolValue(job.TriggersOnDraftPR),
+			RunCompareChanges: types.BoolValue(job.RunCompareChanges),
 			Environment: JobEnvironment{
 				ProjectID:      types.Int64Value(int64(job.Environment.Project_Id)),
 				ID:             types.Int64Value(int64(*job.Environment.ID)),

pkg/framework/objects/job/model.go

@@ -66,4 +66,5 @@ type JobDataSourceModel struct {
 	TriggersOnDraftPr             types.Bool            `tfsdk:"triggers_on_draft_pr"`
 	Environment                   JobEnvironment        `tfsdk:"environment"`
 	JobCompletionTriggerCondition *JobCompletionTrigger `tfsdk:"job_completion_trigger_condition"`
+	RunCompareChanges             types.Bool            `tfsdk:"run_compare_changes"`
 }

pkg/framework/objects/job/schema.go

@@ -77,6 +77,10 @@ func (d *jobsDataSource) Schema(
 							Computed:    true,
 							Description: "Whether the job test source freshness",
 						},
+						"run_compare_changes": schema.BoolAttribute{
+							Computed:    true,
+							Description: "Whether the job should compare data changes introduced by the code change in the PR",
+						},
 						"id": schema.Int64Attribute{
 							Computed:    true,
 							Description: "The ID of the job",

pkg/sdkv2/data_sources/job.go

@@ -98,6 +98,11 @@ var jobSchema = map[string]*schema.Schema{
 		},
 		Description: "Which other job should trigger this job when it finishes, and on which conditions.",
 	},
+	"run_compare_changes": {
+		Type:        schema.TypeBool,
+		Computed:    true,
+		Description: "Whether the CI job should compare data changes introduced by the code change in the PR.",
+	},
 }
 
 func DatasourceJob() *schema.Resource {
@@ -184,6 +189,9 @@ func datasourceJobRead(
 			return diag.FromErr(err)
 		}
 	}
+	if err := d.Set("run_compare_changes", job.RunCompareChanges); err != nil {
+		return diag.FromErr(err)
+	}
 
 	d.SetId(jobId)
 

pkg/sdkv2/resources/job.go

@@ -198,6 +198,13 @@ var jobSchema = map[string]*schema.Schema{
 		},
 		Description: "Which other job should trigger this job when it finishes, and on which conditions (sometimes referred as 'job chaining').",
 	},
+	"run_compare_changes": {
+		Type:     schema.TypeBool,
+		Optional: true,
+		Default:  false,
+		// Once on the plugin framework, put a validation to check that `deferring_environment_id` is set
+		Description: "Whether the CI job should compare data changes introduced by the code changes. Requires `deferring_environment_id` to be set. (Advanced CI needs to be activated in the dbt Cloud Account Settings first as well)",
+	},
 }
 
 func ResourceJob() *schema.Resource {
@@ -390,6 +397,9 @@ func resourceJobRead(ctx context.Context, d *schema.ResourceData, m interface{})
 			return diag.FromErr(err)
 		}
 	}
+	if err := d.Set("run_compare_changes", job.RunCompareChanges); err != nil {
+		return diag.FromErr(err)
+	}
 
 	return diags
 }
@@ -426,6 +436,7 @@ func resourceJobCreate(
 	selfDeferring := d.Get("self_deferring").(bool)
 	timeoutSeconds := d.Get("timeout_seconds").(int)
 	triggersOnDraftPR := d.Get("triggers_on_draft_pr").(bool)
+	runCompareChanges := d.Get("run_compare_changes").(bool)
 
 	var jobCompletionTrigger map[string]any
 	empty, completionJobID, completionProjectID, completionStatuses := utils.ExtractJobConditionSet(
@@ -476,6 +487,7 @@ func resourceJobCreate(
 		timeoutSeconds,
 		triggersOnDraftPR,
 		jobCompletionTrigger,
+		runCompareChanges,
 	)
 	if err != nil {
 		return diag.FromErr(err)
@@ -516,8 +528,9 @@ func resourceJobUpdate(
 		d.HasChange("deferring_environment_id") ||
 		d.HasChange("self_deferring") ||
 		d.HasChange("timeout_seconds") ||
-		d.HasChange("triggers_on_drat_pr") ||
-		d.HasChange("job_completion_trigger_condition") {
+		d.HasChange("triggers_on_draft_pr") ||
+		d.HasChange("job_completion_trigger_condition") ||
+		d.HasChange("run_compare_changes") {
 		job, err := c.GetJob(jobId)
 		if err != nil {
 			return diag.FromErr(err)
@@ -690,6 +703,10 @@ func resourceJobUpdate(
 				job.JobCompletionTrigger = &jobCondTrigger
 			}
 		}
+		if d.HasChange("run_compare_changes") {
+			runCompareChanges := d.Get("run_compare_changes").(bool)
+			job.RunCompareChanges = runCompareChanges
+		}
 
 		_, err = c.UpdateJob(jobId, *job)
 		if err != nil {

pkg/sdkv2/resources/job_acceptance_test.go

@@ -580,9 +580,17 @@ func testAccDbtCloudJobResourceBasicConfigTriggers(
 	git_trigger := "false"
 	schedule_trigger := "false"
 	on_merge_trigger := "false"
+	run_compare_changes := "false"
+	deferringConfig := ""
 
 	if trigger == "git" {
 		git_trigger = "true"
+		deferringConfig = "deferring_environment_id = dbtcloud_environment.test_job_environment.environment_id"
+		if !isDbtCloudPR() {
+			// we don't want to activate it in Cloud PRs as the setting need to be ON
+			// TODO: When TF supports account settings, activate the setting in this test and remove this logic
+			run_compare_changes = "true"
+		}
 	}
 	if trigger == "schedule" {
 		schedule_trigger = "true"
@@ -600,7 +608,7 @@ resource "dbtcloud_environment" "test_job_environment" {
     project_id = dbtcloud_project.test_job_project.id
     name = "%s"
     dbt_version = "%s"
-    type = "development"
+    type = "deployment"
 }
 
 resource "dbtcloud_job" "test_job" {
@@ -616,8 +624,10 @@ resource "dbtcloud_job" "test_job" {
     "schedule": %s,
 	"on_merge": %s
   }
+  run_compare_changes = %s
+  %s
 }
-`, projectName, environmentName, DBT_CLOUD_VERSION, jobName, git_trigger, git_trigger, schedule_trigger, on_merge_trigger)
+`, projectName, environmentName, DBT_CLOUD_VERSION, jobName, git_trigger, git_trigger, schedule_trigger, on_merge_trigger, run_compare_changes, deferringConfig)
 }
 
 func testAccCheckDbtCloudJobExists(resource string) resource.TestCheckFunc {