spark

Author: Zouxxyy

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/BaseTable.scala

@@ -18,7 +18,7 @@
 
 package org.apache.paimon.spark
 
-import org.apache.paimon.table.Table
+import org.apache.paimon.table.{FileStoreTable, Table}
 import org.apache.paimon.utils.StringUtils
 
 import org.apache.spark.sql.connector.catalog.TableCapability
@@ -37,7 +37,7 @@ abstract class BaseTable
 
   override def capabilities(): JSet[TableCapability] = JCollections.emptySet[TableCapability]()
 
-  override def name: String = table.fullName
+  override def name: String = BaseTable.tableNameWithCatalog(table)
 
   override lazy val schema: StructType = SparkTypeUtils.fromPaimonRowType(table.rowType)
 
@@ -48,6 +48,20 @@ abstract class BaseTable
   override def properties: JMap[String, String] = table.options()
 
   override def toString: String = {
-    s"${table.getClass.getSimpleName}[${table.fullName()}]"
+    s"${table.getClass.getSimpleName}[$name]"
+  }
+}
+
+object BaseTable {
+
+  /** Returns the full table name with catalog prefix if available. */
+  def tableNameWithCatalog(table: Table): String = {
+    val fullName = table.fullName
+    table match {
+      case t: FileStoreTable =>
+        Option(t.catalogEnvironment().catalogName())
+          .fold(fullName)(catalog => s"$catalog.$fullName")
+      case _ => fullName
+    }
   }
 }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala

@@ -19,6 +19,7 @@
 package org.apache.paimon.spark.execution
 
 import org.apache.paimon.partition.PartitionPredicate
+import org.apache.paimon.spark.BaseTable
 import org.apache.paimon.spark.leafnode.PaimonLeafV2CommandExec
 import org.apache.paimon.table.{FileStoreTable, Table}
 import org.apache.paimon.utils.InternalRowPartitionComputer
@@ -69,7 +70,7 @@ case class TruncatePaimonTableWithFilterExec(
   override def output: Seq[Attribute] = Nil
 
   override def simpleString(maxFields: Int): String = {
-    s"TruncatePaimonTableWithFilterExec: ${table.fullName()}" +
+    s"TruncatePaimonTableWithFilterExec: ${BaseTable.tableNameWithCatalog(table)}" +
       partitionPredicate.map(p => s", PartitionPredicate: [$p]").getOrElse("")
   }
 }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala

@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions
 import org.apache.paimon.partition.PartitionPredicate
 import org.apache.paimon.predicate.{Predicate, TopN, VectorSearch}
 import org.apache.paimon.spark.{PaimonBatch, PaimonInputPartition, PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric, PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric, SparkTypeUtils}
+import org.apache.paimon.spark.BaseTable
 import org.apache.paimon.spark.schema.PaimonMetadataColumn
 import org.apache.paimon.spark.schema.PaimonMetadataColumn._
 import org.apache.paimon.spark.util.{OptionUtils, SplitUtils}
@@ -182,7 +183,7 @@ trait BaseScan extends Scan with SupportsReportStatistics with Logging {
     } else {
       ""
     }
-    s"${getClass.getSimpleName}: [${table.name}]" +
+    s"${getClass.getSimpleName}: [${BaseTable.tableNameWithCatalog(table)}]" +
       pushedPartitionFiltersStr +
       pushedRuntimePartitionFiltersStr +
       pushedDataFiltersStr +

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonLocalScan.scala

@@ -19,6 +19,7 @@
 package org.apache.paimon.spark.read
 
 import org.apache.paimon.partition.PartitionPredicate
+import org.apache.paimon.spark.BaseTable
 import org.apache.paimon.table.Table
 
 import org.apache.spark.sql.catalyst.InternalRow
@@ -39,6 +40,6 @@ case class PaimonLocalScan(
     } else {
       ""
     }
-    s"PaimonLocalScan: [${table.name}]" + pushedPartitionFiltersStr
+    s"PaimonLocalScan: [${BaseTable.tableNameWithCatalog(table)}]" + pushedPartitionFiltersStr
   }
 }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala

@@ -104,7 +104,7 @@ class PaimonV2Write(
       case Some(_) if !overwriteDynamic => ", overwriteTable=true"
       case _ => ""
     }
-    s"PaimonWrite(table=${table.fullName()}$overwriteDynamicStr$overwritePartitionsStr)"
+    s"PaimonWrite(table=${BaseTable.tableNameWithCatalog(table)}$overwriteDynamicStr$overwritePartitionsStr)"
   }
 
   override def description(): String = toString

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWrite.scala

@@ -19,7 +19,7 @@
 package org.apache.paimon.spark.write
 
 import org.apache.paimon.options.Options
-import org.apache.paimon.spark.SaveMode
+import org.apache.paimon.spark.{BaseTable, SaveMode}
 import org.apache.paimon.spark.commands.WriteIntoPaimonTable
 import org.apache.paimon.table.FileStoreTable
 
@@ -38,6 +38,6 @@ class PaimonWrite(val table: FileStoreTable, saveMode: SaveMode, options: Option
   }
 
   override def toString: String = {
-    s"table: ${table.fullName()}, saveMode: $saveMode, options: ${options.toMap}"
+    s"table: ${BaseTable.tableNameWithCatalog(table)}, saveMode: $saveMode, options: ${options.toMap}"
   }
 }

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteRequireDistributionTest.scala

@@ -49,7 +49,7 @@ class V2WriteRequireDistributionTest extends PaimonSparkTestBase with AdaptiveSp
         val node1 = nodes(0)
         assert(
           node1.isInstanceOf[AppendDataExec] &&
-            node1.toString.contains("PaimonWrite(table=test.t1"),
+            node1.toString.contains("PaimonWrite(table=paimon.test.t1"),
           s"Expected AppendDataExec with specific paimon write, but got: $node1"
         )
 
@@ -92,7 +92,7 @@ class V2WriteRequireDistributionTest extends PaimonSparkTestBase with AdaptiveSp
         val node1 = nodes(0)
         assert(
           node1.isInstanceOf[AppendDataExec] &&
-            node1.toString.contains("PaimonWrite(table=test.t1"),
+            node1.toString.contains("PaimonWrite(table=paimon.test.t1"),
           s"Expected AppendDataExec with specific paimon write, but got: $node1"
         )
 
@@ -136,7 +136,7 @@ class V2WriteRequireDistributionTest extends PaimonSparkTestBase with AdaptiveSp
         val node1 = nodes(0)
         assert(
           node1.isInstanceOf[AppendDataExecV1] &&
-            node1.toString.contains("AppendDataExecV1 PrimaryKeyFileStoreTable[test.t1]"),
+            node1.toString.contains("AppendDataExecV1 PrimaryKeyFileStoreTable[paimon.test.t1]"),
           s"Expected AppendDataExec with specific paimon write, but got: $node1"
         )
       }