[SPARK-54358][SDP] Checkpoint dirs collide when streaming tables in different schemas have same name

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

Updates the per-streaming table checkpoint path to use the fully qualified table path, instead of just its name.

### Why are the changes needed?

A streaming table is a table fed by a stream. Streaming tables have checkpoint directories underneath their pipeline's storage root. These directories don't currently take the table namespace into account, which means that two tables with different schemas but the same name will be mapped to the same checkpoint directory. This could be very bad and cause data loss.

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

Yes, but for an unreleased feature.

### How was this patch tested?

Added a test for the collision case. Verified that it fails with the prior logic and now passes.

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

Closes #53070 from sryza/collide.

Authored-by: Sandy Ryza <sandy.ryza@databricks.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
(cherry picked from commit e09c999)
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: sryza

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

@@ -40,14 +40,14 @@ case class FlowSystemMetadata(
    * which is storage/_checkpoints/flow_destination_table/flow_name.
    * @return the checkpoint root directory for `flow`
    */
-  private def flowCheckpointsDirOpt(): Option[Path] = {
+  def flowCheckpointsDirOpt(): Option[Path] = {
     Option(if (graph.table.contains(flow.destinationIdentifier) ||
       graph.sink.contains(flow.destinationIdentifier)) {
       val checkpointRoot = new Path(context.storageRoot, "_checkpoints")
-      val flowTableName = flow.destinationIdentifier.table
+      val flowTableId = flow.destinationIdentifier.nameParts.mkString(Path.SEPARATOR)
       val flowName = flow.identifier.table
       val checkpointDir = new Path(
-        new Path(checkpointRoot, flowTableName),
+        new Path(checkpointRoot, flowTableId),
         flowName
       )
       logInfo(

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

@@ -227,6 +227,38 @@ class SystemMetadataSuite
       updateContext2
     )
   }
+
+  test("checkpoint dirs for tables with same name but different schema don't collide") {
+    val session = spark
+    import session.implicits._
+
+    // create a pipeline with only a single ST
+    val graph = new TestGraphRegistrationContext(spark) {
+      implicit val sparkSession: SparkSession = spark
+      val mem: MemoryStream[Int] = MemoryStream[Int]
+      mem.addData(1, 2, 3)
+      registerView("a", query = dfFlowFunc(mem.toDF()))
+      registerTable("st")
+      registerFlow("st", "st", query = readStreamFlowFunc("a"))
+      registerTable("schema2.st")
+      registerFlow("schema2.st", "schema2.st", query = readStreamFlowFunc("a"))
+    }.toDataflowGraph
+
+    val updateContext = TestPipelineUpdateContext(
+      unresolvedGraph = graph,
+      spark = spark,
+      storageRoot = storageRoot,
+      failOnErrorEvent = true
+    )
+
+    val stFlow = graph.flow(fullyQualifiedIdentifier("st"))
+    val schema2StFlow = graph.flow(fullyQualifiedIdentifier("st", database = Option("schema2")))
+    val stSystemMetadata = FlowSystemMetadata(updateContext, stFlow, graph)
+    val schema2StSystemMetadata = FlowSystemMetadata(updateContext, schema2StFlow, graph)
+    assert(
+      stSystemMetadata.flowCheckpointsDirOpt() != schema2StSystemMetadata.flowCheckpointsDirOpt()
+    )
+  }
 }
 
 trait SystemMetadataTestHelpers {
@@ -242,8 +274,9 @@ trait SystemMetadataTestHelpers {
   ): Path = {
     val expectedRawCheckPointDir = tableOrSinkElement match {
       case t if t.isInstanceOf[Table] || t.isInstanceOf[Sink] =>
+        val tableId = t.identifier.nameParts.mkString(Path.SEPARATOR)
         new Path(updateContext.storageRoot)
-        .suffix(s"/_checkpoints/${t.identifier.table}/${flowElement.identifier.table}")
+        .suffix(s"/_checkpoints/$tableId/${flowElement.identifier.table}")
         .toString
       case _ =>
         fail(