diff --git a/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java b/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
index 9eb8a5538920..962508368385 100644
--- a/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
+++ b/paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java
@@ -48,18 +48,21 @@ public class CatalogContext implements Serializable {
     private final SerializableConfiguration hadoopConf;
     @Nullable private final FileIOLoader preferIOLoader;
     @Nullable private final FileIOLoader fallbackIOLoader;
+    @Nullable private final String catalogName;
 
     private CatalogContext(
             Options options,
             @Nullable Configuration hadoopConf,
             @Nullable FileIOLoader preferIOLoader,
-            @Nullable FileIOLoader fallbackIOLoader) {
+            @Nullable FileIOLoader fallbackIOLoader,
+            @Nullable String catalogName) {
         this.options = checkNotNull(options);
         this.hadoopConf =
                 new SerializableConfiguration(
                         hadoopConf == null ? getHadoopConfiguration(options) : hadoopConf);
         this.preferIOLoader = preferIOLoader;
         this.fallbackIOLoader = fallbackIOLoader;
+        this.catalogName = catalogName;
     }
 
     public static CatalogContext create(Path warehouse) {
@@ -69,28 +72,43 @@ public static CatalogContext create(Path warehouse) {
     }
 
     public static CatalogContext create(Options options) {
-        return new CatalogContext(options, null, null, null);
+        return new CatalogContext(options, null, null, null, null);
     }
 
     public static CatalogContext create(Options options, Configuration hadoopConf) {
-        return new CatalogContext(options, hadoopConf, null, null);
+        return new CatalogContext(options, hadoopConf, null, null, null);
+    }
+
+    public static CatalogContext create(
+            Options options, Configuration hadoopConf, @Nullable String catalogName) {
+        return new CatalogContext(options, hadoopConf, null, null, catalogName);
     }
 
     public static CatalogContext create(Options options, FileIOLoader fallbackIOLoader) {
-        return new CatalogContext(options, null, null, fallbackIOLoader);
+        return new CatalogContext(options, null, null, fallbackIOLoader, null);
     }
 
     public static CatalogContext create(
             Options options, FileIOLoader preferIOLoader, FileIOLoader fallbackIOLoader) {
-        return new CatalogContext(options, null, preferIOLoader, fallbackIOLoader);
+        return new CatalogContext(options, null, preferIOLoader, fallbackIOLoader, null);
+    }
+
+    public static CatalogContext create(
+            Options options,
+            FileIOLoader preferIOLoader,
+            FileIOLoader fallbackIOLoader,
+            @Nullable String catalogName) {
+        return new CatalogContext(options, null, preferIOLoader, fallbackIOLoader, catalogName);
     }
 
     public static CatalogContext create(
             Options options,
             Configuration hadoopConf,
             FileIOLoader preferIOLoader,
-            FileIOLoader fallbackIOLoader) {
-        return new CatalogContext(options, hadoopConf, preferIOLoader, fallbackIOLoader);
+            FileIOLoader fallbackIOLoader,
+            @Nullable String catalogName) {
+        return new CatalogContext(
+                options, hadoopConf, preferIOLoader, fallbackIOLoader, catalogName);
     }
 
     public Options options() {
@@ -111,4 +129,9 @@ public FileIOLoader preferIO() {
     public FileIOLoader fallbackIO() {
         return fallbackIOLoader;
     }
+
+    @Nullable
+    public String catalogName() {
+        return catalogName;
+    }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
index 6a46d1a49008..85f2e4dd87f9 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
@@ -64,7 +64,11 @@ public void configure(CatalogContext context) {
         options.set(RESOLVING_FILE_IO_ENABLED, false);
         this.context =
                 CatalogContext.create(
-                        options, context.hadoopConf(), context.preferIO(), context.fallbackIO());
+                        options,
+                        context.hadoopConf(),
+                        context.preferIO(),
+                        context.fallbackIO(),
+                        context.catalogName());
     }
 
     @Override
diff --git a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
index f9f0d283bbba..927bc5fbad07 100644
--- a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
@@ -181,7 +181,8 @@ public FileIO fileIO() throws IOException {
                             options,
                             catalogContext.hadoopConf(),
                             catalogContext.preferIO(),
-                            catalogContext.fallbackIO());
+                            catalogContext.fallbackIO(),
+                            catalogContext.catalogName());
             try {
                 fileIO = FileIO.get(path, context);
             } catch (IOException e) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 709010d2ce85..7af46a666933 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -95,6 +95,12 @@ protected AbstractCatalog(FileIO fileIO, CatalogContext context) {
         this.context = context;
     }
 
+    @Nullable
+    @Override
+    public String name() {
+        return context.catalogName();
+    }
+
     @Override
     public Map<String, String> options() {
         return context.options().toMap();
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 3947a950e59a..d55c465a30f6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -1166,6 +1166,10 @@ TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String
 
     // ==================== Catalog Information ==========================
 
+    /** The name of this catalog. */
+    @Nullable
+    String name();
+
     /** Catalog options for re-creating this catalog. */
     Map<String, String> options();
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
index a33dc2c0c106..fad1e35c0972 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
@@ -292,6 +292,7 @@ public static Table loadTable(
 
         CatalogEnvironment catalogEnv =
                 new CatalogEnvironment(
+                        catalog.name(),
                         tableIdentifier,
                         metadata.uuid(),
                         isRestCatalog && metadata.isExternal() ? null : catalog.catalogLoader(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index 81558ac68a48..edd2176b8de1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -53,6 +53,12 @@ public Catalog wrapped() {
         return wrapped;
     }
 
+    @Nullable
+    @Override
+    public String name() {
+        return wrapped.name();
+    }
+
     @Override
     public boolean caseSensitive() {
         return wrapped.caseSensitive();
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 7b053d120c90..ad82c6d7f3eb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -113,11 +113,17 @@ public RESTCatalog(CatalogContext context, boolean configRequired) {
                         api.options(),
                         context.hadoopConf(),
                         context.preferIO(),
-                        context.fallbackIO());
+                        context.fallbackIO(),
+                        context.catalogName());
         this.dataTokenEnabled = api.options().get(RESTTokenFileIO.DATA_TOKEN_ENABLED);
         this.tableDefaultOptions = CatalogUtils.tableDefaultOptions(this.context.options().toMap());
     }
 
+    @Override
+    public String name() {
+        return context.catalogName();
+    }
+
     @Override
     public Map<String, String> options() {
         return context.options().toMap();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
index 882ad952b5f3..0c7348a287dc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -43,8 +43,9 @@
 /** Catalog environment in table which contains log factory, metastore client factory. */
 public class CatalogEnvironment implements Serializable {
 
-    private static final long serialVersionUID = 2L;
+    private static final long serialVersionUID = 3L;
 
+    @Nullable private final String catalogName;
     @Nullable private final Identifier identifier;
     @Nullable private final String uuid;
     @Nullable private final CatalogLoader catalogLoader;
@@ -55,6 +56,7 @@ public class CatalogEnvironment implements Serializable {
     private final boolean supportsPartitionModification;
 
     public CatalogEnvironment(
+            @Nullable String catalogName,
             @Nullable Identifier identifier,
             @Nullable String uuid,
             @Nullable CatalogLoader catalogLoader,
@@ -63,6 +65,7 @@ public CatalogEnvironment(
             @Nullable CatalogContext catalogContext,
             boolean supportsVersionManagement,
             boolean supportsPartitionModification) {
+        this.catalogName = catalogName;
         this.identifier = identifier;
         this.uuid = uuid;
         this.catalogLoader = catalogLoader;
@@ -74,7 +77,12 @@ public CatalogEnvironment(
     }
 
     public static CatalogEnvironment empty() {
-        return new CatalogEnvironment(null, null, null, null, null, null, false, false);
+        return new CatalogEnvironment(null, null, null, null, null, null, null, false, false);
+    }
+
+    @Nullable
+    public String catalogName() {
+        return catalogName;
     }
 
     @Nullable
@@ -173,6 +181,7 @@ public CatalogContext catalogContext() {
 
     public CatalogEnvironment copy(Identifier identifier) {
         return new CatalogEnvironment(
+                catalogName,
                 identifier,
                 uuid,
                 catalogLoader,
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogFactoryTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogFactoryTest.java
index 7f93b8d61c55..e45319f90ccc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogFactoryTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogFactoryTest.java
@@ -93,7 +93,7 @@ public void testContextSerializable() throws IOException, ClassNotFoundException
         conf.set("my_key", "my_value");
         CatalogContext context =
                 CatalogContext.create(
-                        new Options(), conf, new TestFileIOLoader(), new TestFileIOLoader());
+                        new Options(), conf, new TestFileIOLoader(), new TestFileIOLoader(), null);
         context = InstantiationUtil.clone(context);
         assertThat(context.hadoopConf().get("my_key")).isEqualTo(conf.get("my_key"));
     }
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
index 4eef95b55b14..06516cb308fa 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
@@ -133,7 +133,7 @@ public void close() throws Exception {}
                 };
 
         CatalogEnvironment env =
-                new CatalogEnvironment(null, null, null, null, null, null, false, false) {
+                new CatalogEnvironment(null, null, null, null, null, null, null, false, false) {
 
                     @Override
                     public PartitionModification partitionModification() {
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 81cb097f5adb..c36a1e00e17f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -2663,6 +2663,7 @@ private FileStoreTable getFileTable(Identifier identifier) {
         TableSchema schema = tableMetadata.schema();
         CatalogEnvironment catalogEnv =
                 new CatalogEnvironment(
+                        catalog.name(),
                         identifier,
                         tableMetadata.uuid(),
                         catalog.catalogLoader(),
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
index 293320721385..53b6ef435c72 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java
@@ -53,7 +53,10 @@ public FlinkCatalog createCatalog(Context context) {
         return createCatalog(
                 context.getName(),
                 CatalogContext.create(
-                        Options.fromMap(context.getOptions()), new FlinkFileIOLoader()),
+                        Options.fromMap(context.getOptions()),
+                        new FlinkFileIOLoader(),
+                        null,
+                        context.getName()),
                 context.getClassLoader());
     }
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalogFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalogFactory.java
index 6ec0d8a8df08..c70acb80aad2 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalogFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalogFactory.java
@@ -93,7 +93,8 @@ public static FlinkGenericCatalog createCatalog(
         FlinkCatalog paimon =
                 new FlinkCatalog(
                         org.apache.paimon.catalog.CatalogFactory.createCatalog(
-                                CatalogContext.create(options, new FlinkFileIOLoader()), cl),
+                                CatalogContext.create(options, new FlinkFileIOLoader(), null, name),
+                                cl),
                         name,
                         options.get(DEFAULT_DATABASE),
                         options);
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 9d12201c91a2..d281b273cdc8 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -128,7 +128,8 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
         CatalogContext catalogContext =
                 CatalogContext.create(
                         Options.fromMap(options.asCaseSensitiveMap()),
-                        sparkSession.sessionState().newHadoopConf());
+                        sparkSession.sessionState().newHadoopConf(),
+                        name);
         this.catalog = CatalogFactory.createCatalog(catalogContext);
         this.defaultDatabase =
                 options.getOrDefault(DEFAULT_DATABASE.key(), DEFAULT_DATABASE.defaultValue());
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/BaseTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/BaseTable.scala
index 9ffa6b194d60..5a358ea548a8 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/BaseTable.scala
+++ b/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
+    }
   }
 }
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
index 41c2187ce37a..2d9c28ccedb5 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
+++ b/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("")
   }
 }
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
index 78cde8752e3f..f4100204efdd 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
+++ b/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 +
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonLocalScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonLocalScan.scala
index a176cd568646..eca9efc71a7e 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonLocalScan.scala
+++ b/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
   }
 }
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
index e4676ae4af1f..26a3f16431a7 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
+++ b/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
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWrite.scala
index 3ed93b0dd3aa..7bab8a51eec9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWrite.scala
+++ b/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}"
   }
 }
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DescribeTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DescribeTableTestBase.scala
index 63efa3f7e0a6..06f369075e52 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DescribeTableTestBase.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DescribeTableTestBase.scala
@@ -224,4 +224,21 @@ abstract class DescribeTableTestBase extends PaimonSparkTestBase {
       parameters.head.contains(PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME))
     Assertions.assertTrue(parameters.head.contains(PartitionStatistics.FIELD_RECORD_COUNT))
   }
+
+  test("Paimon table name and scan description should include catalog name") {
+    spark.sql("CREATE TABLE T (id INT, name STRING)")
+    spark.sql("INSERT INTO T VALUES (1, 'a')")
+
+    // Table name should include catalog prefix
+    val relation = createRelationV2("T")
+    Assertions.assertTrue(
+      relation.table.name().startsWith("paimon."),
+      s"Table name should start with 'paimon.', but got: ${relation.table.name()}")
+
+    // Scan description should include catalog prefix
+    val scan = getScan("SELECT * FROM T")
+    Assertions.assertTrue(
+      scan.description().contains("paimon."),
+      s"Scan description should contain 'paimon.', but got: ${scan.description()}")
+  }
 }
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteRequireDistributionTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteRequireDistributionTest.scala
index 02a5b9a83015..8b7eee52dea3 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteRequireDistributionTest.scala
+++ b/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"
         )
       }