[core] Add Table.uuid method

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

@@ -423,16 +423,17 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
 
     protected Table getDataOrFormatTable(Identifier identifier) throws TableNotExistException {
         Preconditions.checkArgument(identifier.getSystemTableName() == null);
-        TableSchema tableSchema = getDataTableSchema(identifier);
+        TableMeta tableMeta = getDataTableMeta(identifier);
         return FileStoreTableFactory.create(
                 fileIO,
                 getTableLocation(identifier),
-                tableSchema,
+                tableMeta.schema,
                 new CatalogEnvironment(
                         identifier,
+                        tableMeta.uuid,
                         Lock.factory(
                                 lockFactory().orElse(null), lockContext().orElse(null), identifier),
-                        metastoreClientFactory(identifier, tableSchema).orElse(null),
+                        metastoreClientFactory(identifier, tableMeta.schema).orElse(null),
                         lineageMetaFactory));
     }
 
@@ -474,6 +475,10 @@ public Map<String, Map<String, Path>> allTablePaths() {
         }
     }
 
+    protected TableMeta getDataTableMeta(Identifier identifier) throws TableNotExistException {
+        return new TableMeta(getDataTableSchema(identifier), null);
+    }
+
     protected abstract TableSchema getDataTableSchema(Identifier identifier)
             throws TableNotExistException;
 
@@ -626,4 +631,25 @@ public Optional<TableSchema> tableSchemaInFileSystem(Path tablePath, String bran
                             }
                         });
     }
+
+    /** Table metadata. */
+    protected static class TableMeta {
+
+        private final TableSchema schema;
+        @Nullable private final String uuid;
+
+        public TableMeta(TableSchema schema, @Nullable String uuid) {
+            this.schema = schema;
+            this.uuid = uuid;
+        }
+
+        public TableSchema schema() {
+            return schema;
+        }
+
+        @Nullable
+        public String uuid() {
+            return uuid;
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -78,6 +78,7 @@
 import static org.apache.paimon.catalog.Identifier.UNKNOWN_DATABASE;
 import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
 import static org.apache.paimon.utils.FileUtils.listVersionedFiles;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
 import static org.apache.paimon.utils.Preconditions.checkState;
 
 /** Schema Manager to manage schema versions. */
@@ -123,6 +124,24 @@ public Optional<TableSchema> latest() {
         }
     }
 
+    public long earliestCreationTime() {
+        try {
+            long earliest = 0;
+            if (!schemaExists(0)) {
+                Optional<Long> min =
+                        listVersionedFiles(fileIO, schemaDirectory(), SCHEMA_PREFIX)
+                                .reduce(Math::min);
+                checkArgument(min.isPresent());
+                earliest = min.get();
+            }
+
+            Path schemaPath = toSchemaPath(earliest);
+            return fileIO.getFileStatus(schemaPath).getModificationTime();
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
     public List<TableSchema> listAll() {
         return listAllIds().stream().map(this::schema).collect(Collectors.toList());
     }

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -167,6 +167,15 @@ public Identifier identifier() {
                 : identifier;
     }
 
+    @Override
+    public String uuid() {
+        if (catalogEnvironment.uuid() != null) {
+            return catalogEnvironment.uuid();
+        }
+        long earliestCreationTime = schemaManager().earliestCreationTime();
+        return fullName() + "." + earliestCreationTime;
+    }
+
     @Override
     public Optional<Statistics> statistics() {
         Snapshot snapshot = TimeTravelUtil.resolveSnapshot(this);

paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java

@@ -36,30 +36,38 @@ public class CatalogEnvironment implements Serializable {
     private static final long serialVersionUID = 1L;
 
     @Nullable private final Identifier identifier;
+    @Nullable private final String uuid;
     private final Lock.Factory lockFactory;
     @Nullable private final MetastoreClient.Factory metastoreClientFactory;
     @Nullable private final LineageMetaFactory lineageMetaFactory;
 
     public CatalogEnvironment(
             @Nullable Identifier identifier,
+            @Nullable String uuid,
             Lock.Factory lockFactory,
             @Nullable MetastoreClient.Factory metastoreClientFactory,
             @Nullable LineageMetaFactory lineageMetaFactory) {
         this.identifier = identifier;
+        this.uuid = uuid;
         this.lockFactory = lockFactory;
         this.metastoreClientFactory = metastoreClientFactory;
         this.lineageMetaFactory = lineageMetaFactory;
     }
 
     public static CatalogEnvironment empty() {
-        return new CatalogEnvironment(null, Lock.emptyFactory(), null, null);
+        return new CatalogEnvironment(null, null, Lock.emptyFactory(), null, null);
     }
 
     @Nullable
     public Identifier identifier() {
         return identifier;
     }
 
+    @Nullable
+    public String uuid() {
+        return uuid;
+    }
+
     public Lock.Factory lockFactory() {
         return lockFactory;
     }

paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java

@@ -72,6 +72,11 @@ public String fullName() {
         return wrapped.fullName();
     }
 
+    @Override
+    public String uuid() {
+        return wrapped.uuid();
+    }
+
     @Override
     public SnapshotReader newSnapshotReader() {
         return wrapped.newSnapshotReader();

paimon-core/src/main/java/org/apache/paimon/table/Table.java

@@ -52,10 +52,19 @@ public interface Table extends Serializable {
     /** A name to identify this table. */
     String name();
 
+    /** Full name of the table, default is database.tableName. */
     default String fullName() {
         return name();
     }
 
+    /**
+     * UUID of the table, metastore can provide the true UUID of this table, default is the full
+     * name.
+     */
+    default String uuid() {
+        return fullName();
+    }
+
     /** Returns the row type of this table. */
     RowType rowType();
 

paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java

@@ -948,4 +948,16 @@ public void testFormatTable() throws Exception {
                 .isInstanceOf(Catalog.TableNotExistException.class);
         assertThat(catalog.getTable(newIdentifier)).isInstanceOf(FormatTable.class);
     }
+
+    @Test
+    public void testTableUUID() throws Exception {
+        catalog.createDatabase("test_db", false);
+        Identifier identifier = Identifier.create("test_db", "test_table");
+        catalog.createTable(identifier, DEFAULT_TABLE_SCHEMA, false);
+        Table table = catalog.getTable(identifier);
+        String uuid = table.uuid();
+        assertThat(uuid).startsWith(identifier.getFullName() + ".");
+        assertThat(Long.parseLong(uuid.substring((identifier.getFullName() + ".").length())))
+                .isGreaterThan(0);
+    }
 }

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -411,6 +411,18 @@ protected List<String> listTablesImpl(String databaseName) {
         }
     }
 
+    @Override
+    protected TableMeta getDataTableMeta(Identifier identifier) throws TableNotExistException {
+        return getDataTableMeta(identifier, getHmsTable(identifier));
+    }
+
+    private TableMeta getDataTableMeta(Identifier identifier, Table table)
+            throws TableNotExistException {
+        return new TableMeta(
+                getDataTableSchema(identifier, table),
+                identifier.getFullName() + "." + table.getCreateTime());
+    }
+
     @Override
     public TableSchema getDataTableSchema(Identifier identifier) throws TableNotExistException {
         Table table = getHmsTable(identifier);
@@ -567,18 +579,19 @@ public org.apache.paimon.table.Table getDataOrFormatTable(Identifier identifier)
         Preconditions.checkArgument(identifier.getSystemTableName() == null);
         Table table = getHmsTable(identifier);
         try {
-            TableSchema tableSchema = getDataTableSchema(identifier, table);
+            TableMeta tableMeta = getDataTableMeta(identifier, table);
             return FileStoreTableFactory.create(
                     fileIO,
                     getTableLocation(identifier, table),
-                    tableSchema,
+                    tableMeta.schema(),
                     new CatalogEnvironment(
                             identifier,
+                            tableMeta.uuid(),
                             Lock.factory(
                                     lockFactory().orElse(null),
                                     lockContext().orElse(null),
                                     identifier),
-                            metastoreClientFactory(identifier, tableSchema).orElse(null),
+                            metastoreClientFactory(identifier, tableMeta.schema()).orElse(null),
                             lineageMetaFactory));
         } catch (TableNotExistException ignore) {
         }