Avoid using d.Get("task") for each task in appflow_flow

.changelog/44543

@@ -0,0 +1,3 @@
+```release-note:bug
+resource/aws_appflow_flow: Fix slowness caused by o(n^2) logic relating to number of tasks in a flow
+```

internal/service/appflow/flow.go

@@ -1208,16 +1208,12 @@ func resourceFlow() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(0, 2048),
 							},
 							DiffSuppressFunc: func(k, oldValue, newValue string, d *schema.ResourceData) bool {
-								if v, ok := d.Get("task").(*schema.Set); ok && v.Len() == 1 {
-									if tl, ok := v.List()[0].(map[string]any); ok && len(tl) > 0 {
-										if sf, ok := tl["source_fields"].([]any); ok && len(sf) == 1 {
-											if sf[0] == "" {
-												return oldValue == "0" && newValue == "1"
-											}
-										}
-									}
+								marker, ok := d.Get("single_task_flag").(bool)
+								if !ok || !marker {
+									return false
 								}
-								return false
+
+								return oldValue == "0" && newValue == "1"
 							},
 						},
 						"task_properties": {
@@ -1236,6 +1232,16 @@ func resourceFlow() *schema.Resource {
 					},
 				},
 			},
+			"single_task_flag": {
+				Type:     schema.TypeBool,
+				Computed: true,
+				Optional: true,
+				ForceNew: false,
+				DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
+					// Ignore internal field
+					return true
+				},
+			},
 			"trigger_config": {
 				Type:     schema.TypeList,
 				Required: true,
@@ -1415,9 +1421,13 @@ func resourceFlowRead(ctx context.Context, d *schema.ResourceData, meta any) dia
 	} else {
 		d.Set("source_flow_config", nil)
 	}
-	if err := d.Set("task", flattenTasks(output.Tasks)); err != nil {
+	tasks := flattenTasks(output.Tasks)
+	if err := d.Set("task", tasks); err != nil {
 		return sdkdiag.AppendErrorf(diags, "setting task: %s", err)
 	}
+	if err := d.Set("single_task_flag", len(tasks) == 1); err != nil {
+		return sdkdiag.AppendErrorf(diags, "setting single_task_flag: %s", err)
+	}
 	if output.TriggerConfig != nil {
 		if err := d.Set("trigger_config", []any{flattenTriggerConfig(output.TriggerConfig)}); err != nil {
 			return sdkdiag.AppendErrorf(diags, "setting trigger_config: %s", err)

internal/service/appflow/flow_test.go

@@ -50,6 +50,7 @@ func TestAccAppFlowFlow_basic(t *testing.T) {
 					resource.TestCheckResourceAttrSet(resourceName, "task.#"),
 					resource.TestCheckResourceAttrSet(resourceName, "task.0.source_fields.#"),
 					resource.TestCheckResourceAttrSet(resourceName, "task.0.task_type"),
+					resource.TestCheckResourceAttr(resourceName, "single_task_flag", acctest.CtTrue),
 					resource.TestCheckResourceAttr(resourceName, "trigger_config.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "trigger_config.0.trigger_type", "Scheduled"),
 					resource.TestCheckResourceAttr(resourceName, "trigger_config.0.trigger_properties.#", "1"),
@@ -217,6 +218,7 @@ func TestAccAppFlowFlow_taskUpdate(t *testing.T) {
 						"task_properties.SOURCE_DATA_TYPE":      names.AttrID,
 						"task_type":                             "Map",
 					}),
+					resource.TestCheckResourceAttr(resourceName, "single_task_flag", acctest.CtFalse),
 				),
 			},
 			{