support real uuid for iceberg table

Author: mingfeng

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

@@ -264,7 +264,7 @@ public static Table loadTable(
         }
 
         if (options.type() == TableType.ICEBERG_TABLE) {
-            return toIcebergTable(identifier, schema, dataFileIO);
+            return toIcebergTable(identifier, schema, dataFileIO, metadata.uuid());
         }
 
         Identifier tableIdentifier = identifier;
@@ -443,7 +443,10 @@ private static LanceTable toLanceTable(
     }
 
     private static IcebergTable toIcebergTable(
-            Identifier identifier, TableSchema schema, Function<Path, FileIO> fileIO) {
+            Identifier identifier,
+            TableSchema schema,
+            Function<Path, FileIO> fileIO,
+            @Nullable String uuid) {
         Map<String, String> options = schema.options();
         String location = options.get(CoreOptions.PATH.key());
         return IcebergTable.builder()
@@ -454,6 +457,7 @@ private static IcebergTable toIcebergTable(
                 .partitionKeys(schema.partitionKeys())
                 .options(options)
                 .comment(schema.comment())
+                .uuid(uuid)
                 .build();
     }
 }

paimon-core/src/main/java/org/apache/paimon/table/iceberg/IcebergTable.java

@@ -50,6 +50,7 @@ class Builder {
         private String location;
         private Map<String, String> options;
         private String comment;
+        private String uuid;
 
         public Builder identifier(Identifier identifier) {
             this.identifier = identifier;
@@ -86,9 +87,14 @@ public Builder comment(String comment) {
             return this;
         }
 
+        public Builder uuid(String uuid) {
+            this.uuid = uuid;
+            return this;
+        }
+
         public IcebergTable build() {
             return new IcebergTableImpl(
-                    identifier, fileIO, rowType, partitionKeys, location, options, comment);
+                    identifier, fileIO, rowType, partitionKeys, location, options, comment, uuid);
         }
     }
 }

paimon-core/src/main/java/org/apache/paimon/table/iceberg/IcebergTableImpl.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.InnerTableScan;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
 
 import javax.annotation.Nullable;
 
@@ -44,6 +45,7 @@ public class IcebergTableImpl implements ReadonlyTable, IcebergTable {
     private final String location;
     private final Map<String, String> options;
     @Nullable private final String comment;
+    @Nullable private final String uuid;
 
     public IcebergTableImpl(
             Identifier identifier,
@@ -52,14 +54,16 @@ public IcebergTableImpl(
             List<String> partitionKeys,
             String location,
             Map<String, String> options,
-            @Nullable String comment) {
+            @Nullable String comment,
+            @Nullable String uuid) {
         this.identifier = identifier;
         this.fileIO = fileIO;
         this.rowType = rowType;
         this.partitionKeys = partitionKeys;
         this.location = location;
-        this.options = options;
+        this.options = options == null ? Collections.emptyMap() : options;
         this.comment = comment;
+        this.uuid = uuid;
     }
 
     @Override
@@ -72,6 +76,11 @@ public String fullName() {
         return identifier.getFullName();
     }
 
+    @Override
+    public String uuid() {
+        return StringUtils.isEmpty(uuid) ? fullName() : uuid;
+    }
+
     @Override
     public RowType rowType() {
         return rowType;
@@ -117,7 +126,7 @@ public IcebergTable copy(Map<String, String> dynamicOptions) {
         Map<String, String> newOptions = new HashMap<>(options);
         newOptions.putAll(dynamicOptions);
         return new IcebergTableImpl(
-                identifier, fileIO, rowType, partitionKeys, location, newOptions, comment);
+                identifier, fileIO, rowType, partitionKeys, location, newOptions, comment, uuid);
     }
 
     @Override