fix: set depends on past value set by users in transformation task level (#488) (#509)

Co-authored-by: Arinda Arif <79823430+arinda-arif@users.noreply.github.com>

Author: sravankorumilli

ext/scheduler/airflow2/compiler/compiler_test.go

@@ -24,7 +24,7 @@ func TestCompiler(t *testing.T) {
 		Owner: "mee@mee",
 		Behavior: models.JobSpecBehavior{
 			CatchUp:       true,
-			DependsOnPast: false,
+			DependsOnPast: true,
 			Retry: models.JobSpecBehaviorRetry{
 				Count:              2,
 				Delay:              time.Second * 2,

ext/scheduler/airflow2/compiler_test.go

@@ -71,7 +71,7 @@ func TestCompilerIntegration(t *testing.T) {
 		Owner: "mee@mee",
 		Behavior: models.JobSpecBehavior{
 			CatchUp:       true,
-			DependsOnPast: false,
+			DependsOnPast: true,
 		},
 		Schedule: models.JobSpecSchedule{
 			StartDate: time.Date(2000, 11, 11, 0, 0, 0, 0, time.UTC),
@@ -94,7 +94,7 @@ func TestCompilerIntegration(t *testing.T) {
 		Owner: "mee@mee",
 		Behavior: models.JobSpecBehavior{
 			CatchUp:       true,
-			DependsOnPast: false,
+			DependsOnPast: true,
 		},
 		Schedule: models.JobSpecSchedule{
 			StartDate: time.Date(2000, 11, 11, 0, 0, 0, 0, time.UTC),
@@ -142,7 +142,7 @@ func TestCompilerIntegration(t *testing.T) {
 		Owner: "mee@mee",
 		Behavior: models.JobSpecBehavior{
 			CatchUp:       true,
-			DependsOnPast: false,
+			DependsOnPast: true,
 			Retry: models.JobSpecBehaviorRetry{
 				Count:              4,
 				Delay:              0,

ext/scheduler/airflow2/resources/base_dag.py

@@ -43,7 +43,7 @@
     "queue": "{{ .Metadata.Airflow.Queue }}",
     {{- end }}
     "owner": {{.Job.Owner | quote}},
-    "depends_on_past": {{ if .Job.Behavior.DependsOnPast }} True {{- else -}} False {{- end -}},
+    "depends_on_past": False,
     "retries": {{ if gt .Job.Behavior.Retry.Count 0 -}} {{.Job.Behavior.Retry.Count}} {{- else -}} DAG_RETRIES {{- end}},
     "retry_delay": {{ if gt .Job.Behavior.Retry.Delay.Nanoseconds 0 -}} timedelta(seconds={{.Job.Behavior.Retry.Delay.Seconds}}) {{- else -}} timedelta(seconds=DAG_RETRY_DELAY) {{- end}},
     "retry_exponential_backoff": {{if .Job.Behavior.Retry.ExponentialBackoff -}}True{{- else -}}False{{- end -}},
@@ -135,6 +135,7 @@
     task_id={{$baseTaskSchema.Name | quote}},
     get_logs=True,
     dag=dag,
+    depends_on_past={{ if .Job.Behavior.DependsOnPast }}True{{- else -}}False{{- end -}},
     in_cluster=True,
     is_delete_operator_pod=True,
     do_xcom_push=False,

ext/scheduler/airflow2/resources/expected_compiled_template.py

@@ -93,6 +93,7 @@
     task_id="bq",
     get_logs=True,
     dag=dag,
+    depends_on_past=True,
     in_cluster=True,
     is_delete_operator_pod=True,
     do_xcom_push=False,