v1

Author: Zouxxyy

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 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,40 @@ 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, 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);
+            Options options, FileIOLoader fallbackIOLoader, String catalogName) {
+        return new CatalogContext(options, null, null, fallbackIOLoader, catalogName);
     }
 
     public static CatalogContext create(
             Options options,
             Configuration hadoopConf,
             FileIOLoader preferIOLoader,
-            FileIOLoader fallbackIOLoader) {
-        return new CatalogContext(options, hadoopConf, preferIOLoader, fallbackIOLoader);
+            FileIOLoader fallbackIOLoader,
+            String catalogName) {
+        return new CatalogContext(
+                options, hadoopConf, preferIOLoader, fallbackIOLoader, catalogName);
+    }
+
+    @Nullable
+    public String catalogName() {
+        return catalogName;
     }
 
     public Options options() {

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

paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -176,7 +176,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) {

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -81,17 +81,25 @@ public abstract class AbstractCatalog implements Catalog {
     protected final FileIO fileIO;
     protected final Map<String, String> tableDefaultOptions;
     protected final Options catalogOptions;
+    @Nullable protected final String name;
 
     protected AbstractCatalog(FileIO fileIO) {
         this.fileIO = fileIO;
         this.tableDefaultOptions = new HashMap<>();
         this.catalogOptions = new Options();
+        this.name = null;
     }
 
-    protected AbstractCatalog(FileIO fileIO, Options options) {
+    protected AbstractCatalog(FileIO fileIO, Options options, @Nullable String name) {
         this.fileIO = fileIO;
         this.tableDefaultOptions = CatalogUtils.tableDefaultOptions(options.toMap());
         this.catalogOptions = options;
+        this.name = name;
+    }
+
+    @Override
+    public String name() {
+        return name;
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -51,6 +51,16 @@
 @Public
 public interface Catalog extends AutoCloseable {
 
+    /**
+     * Return the name of this catalog.
+     *
+     * @return this catalog's name
+     */
+    @Nullable
+    default String name() {
+        return null;
+    }
+
     // ======================= database methods ===============================
 
     /**

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogFactory.java

@@ -63,8 +63,8 @@ static Path warehouse(CatalogContext context) {
      * If the ClassLoader is not specified, using the context ClassLoader of current thread as
      * default.
      */
-    static Catalog createCatalog(CatalogContext options) {
-        return createCatalog(options, CatalogFactory.class.getClassLoader());
+    static Catalog createCatalog(CatalogContext context) {
+        return createCatalog(context, CatalogFactory.class.getClassLoader());
     }
 
     static Catalog createCatalog(CatalogContext context, ClassLoader classLoader) {

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -237,6 +237,7 @@ public static Table loadTable(
 
         CatalogEnvironment catalogEnv =
                 new CatalogEnvironment(
+                        catalog.name(),
                         tableIdentifier,
                         metadata.uuid(),
                         catalog.catalogLoader(),

paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java

@@ -51,6 +51,11 @@ public Catalog wrapped() {
         return wrapped;
     }
 
+    @Override
+    public String name() {
+        return wrapped.name();
+    }
+
     @Override
     public boolean caseSensitive() {
         return wrapped.caseSensitive();

paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java

@@ -50,7 +50,11 @@ public FileSystemCatalog(FileIO fileIO, Path warehouse) {
     }
 
     public FileSystemCatalog(FileIO fileIO, Path warehouse, Options options) {
-        super(fileIO, options);
+        this(fileIO, warehouse, options, null);
+    }
+
+    public FileSystemCatalog(FileIO fileIO, Path warehouse, Options options, String name) {
+        super(fileIO, options, name);
         this.warehouse = warehouse;
     }
 
@@ -198,7 +202,7 @@ public String warehouse() {
 
     @Override
     public CatalogLoader catalogLoader() {
-        return new FileSystemCatalogLoader(fileIO, warehouse, catalogOptions);
+        return new FileSystemCatalogLoader(fileIO, warehouse, catalogOptions, name);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogFactory.java

@@ -41,6 +41,6 @@ public Catalog create(FileIO fileIO, Path warehouse, CatalogContext context) {
                     "Only managed table is supported in File system catalog.");
         }
 
-        return new FileSystemCatalog(fileIO, warehouse, context.options());
+        return new FileSystemCatalog(fileIO, warehouse, context.options(), context.catalogName());
     }
 }