[SPARK-52346][SDP] Fix counter initialization and increment logic for flow retries

### What changes were proposed in this pull request?

Fixes flow failure counter increments by switching to `updateWith`. This prevents first increment from throwing `NoSuchElementException` when key does not exist.

### Why are the changes needed?

Current impl `flowToNumConsecutiveFailure(flowIdentifier)` throws exception when a flow is retried for first time:
```
java.util.NoSuchElementException: key not found: `spark_catalog`.`test_db`.`mv`
	at scala.collection.MapOps.default(Map.scala:289)
	at scala.collection.MapOps.default$(Map.scala:288)
	at scala.collection.AbstractMap.default(Map.scala:420)
	at scala.collection.MapOps.apply(Map.scala:176)
	at scala.collection.MapOps.apply$(Map.scala:175)
	at scala.collection.AbstractMap.apply(Map.scala:420)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.incrementFlowToNumConsecutiveFailure(GraphExecution.scala:52)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.$anonfun$planAndStartFlow$1(GraphExecution.scala:92)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.$anonfun$planAndStartFlow$1$adapted(GraphExecution.scala:90)
	at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:484)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
```

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Run pipeline manually with flow failure to test.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #52094 from JiaqiWang18/sdp-graphexecution-incrementflow-map-default.

Authored-by: Jacky Wang <jacky.wang@databricks.com>
Signed-off-by: Gengliang Wang <gengliang@apache.org>

Author: jackywang-db

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala

@@ -49,7 +49,10 @@ abstract class GraphExecution(
 
   /** Increments flow execution retry count for `flow`. */
   private def incrementFlowToNumConsecutiveFailure(flowIdentifier: TableIdentifier): Unit = {
-    flowToNumConsecutiveFailure.put(flowIdentifier, flowToNumConsecutiveFailure(flowIdentifier) + 1)
+    flowToNumConsecutiveFailure.updateWith(flowIdentifier) {
+      case Some(count) => Some(count + 1)
+      case None => Some(1)
+    }
   }
 
   /**

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala

@@ -26,7 +26,7 @@ import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, TableC
 import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
 import org.apache.spark.sql.pipelines.common.{FlowStatus, RunState}
 import org.apache.spark.sql.pipelines.graph.TriggeredGraphExecution.StreamState
-import org.apache.spark.sql.pipelines.logging.EventLevel
+import org.apache.spark.sql.pipelines.logging.{EventLevel, FlowProgress}
 import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
@@ -1027,4 +1027,33 @@ class TriggeredGraphExecutionSuite extends ExecutionTest with SharedSparkSession
       }
     )
   }
+
+  test("consecutive failure event level is correct") {
+    val session = spark
+    import session.implicits._
+
+    val pipelineDef = new TestGraphRegistrationContext(spark) {
+      registerMaterializedView(
+        "retry_test",
+        partitionCols = Some(Seq("nonexistent_col")),
+        query = dfFlowFunc(spark.range(5).withColumn("id_mod", ($"id" % 2).cast("int")))
+      )
+    }
+
+    val graph = pipelineDef.toDataflowGraph
+    val updateContext = TestPipelineUpdateContext(spark, graph)
+    updateContext.pipelineExecution.runPipeline()
+    updateContext.pipelineExecution.awaitCompletion()
+
+    val failedEvents = updateContext.eventBuffer.getEvents.filter { e =>
+      e.details.isInstanceOf[FlowProgress] &&
+      e.details.asInstanceOf[FlowProgress].status == FlowStatus.FAILED
+    }
+
+    val warnCount = failedEvents.count(_.level == EventLevel.WARN)
+    // flowToNumConsecutiveFailure controls that the last failure should be logged as ERROR
+    val errorCount = failedEvents.count(_.level == EventLevel.ERROR)
+
+    assert(warnCount == 2 && errorCount == 1)
+  }
 }