[#3607] feat(core): Support to import the entities when loading entit…

…ies (#3623)

### What changes were proposed in this pull request?
Support to import the entities when loading entities

### Why are the changes needed?

Fix: #3607 

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Add ut.

---------

Co-authored-by: Heng Qin <[email protected]>
Co-authored-by: Rory <[email protected]>
Co-authored-by: Jerry Shao <[email protected]>

common/src/main/java/com/datastrato/gravitino/dto/rel/TableDTO.java

@@ -58,7 +58,7 @@ private TableDTO() {}
    * @param properties The properties associated with the table.
    * @param audit The audit information for the table.
    * @param partitioning The partitioning of the table.
-   * @param indexes Teh indexes of the table.
+   * @param indexes The indexes of the table.
    */
   private TableDTO(
       String name,

core/src/main/java/com/datastrato/gravitino/GravitinoEnv.java

@@ -40,7 +40,6 @@
 import com.datastrato.gravitino.metrics.source.JVMMetricsSource;
 import com.datastrato.gravitino.storage.IdGenerator;
 import com.datastrato.gravitino.storage.RandomIdGenerator;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -98,40 +97,6 @@ public static GravitinoEnv getInstance() {
     return InstanceHolder.INSTANCE;
   }
 
-  /**
-   * This method is used for testing purposes only to set the lock manager for test in package
-   * `com.datastrato.gravitino.server.web.rest`, as tree lock depends on the lock manager and we did
-   * not mock the lock manager in the test, so we need to set the lock manager for test.
-   *
-   * @param lockManager The lock manager to be set.
-   */
-  @VisibleForTesting
-  public void setLockManager(LockManager lockManager) {
-    this.lockManager = lockManager;
-  }
-
-  /**
-   * This method is used for testing purposes only to set the access manager for test in package
-   * `com.datastrato.gravitino.server.web.rest` and `com.datastrato.gravitino.authorization`.
-   *
-   * @param accessControlManager The access control manager to be set.
-   */
-  @VisibleForTesting
-  public void setAccessControlManager(AccessControlManager accessControlManager) {
-    this.accessControlManager = accessControlManager;
-  }
-
-  /**
-   * This method is used for testing purposes only to set the entity store for test in package
-   * `com.datastrato.gravitino.authorization`.
-   *
-   * @param entityStore The entity store to be set.
-   */
-  @VisibleForTesting
-  public void setEntityStore(EntityStore entityStore) {
-    this.entityStore = entityStore;
-  }
-
   /**
    * Initialize the Gravitino environment.
    *

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedSchema.java

@@ -6,6 +6,7 @@
 
 import com.datastrato.gravitino.Audit;
 import com.datastrato.gravitino.Schema;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.SchemaEntity;
 import java.util.Map;
@@ -19,14 +20,23 @@
 public final class EntityCombinedSchema implements Schema {
 
   private final Schema schema;
+
   private final SchemaEntity schemaEntity;
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
 
+  // Field "imported" is used to indicate whether the entity has been imported to Gravitino
+  // managed storage backend. If "imported" is true, it means that storage backend have stored
+  // the correct entity. Otherwise, we should import the external entity to the storage backend.
+  // This is used for tag/access control related purposes, only the imported entities have the
+  // unique id, and based on this id, we can label and control the access to the entities.
+  private boolean imported;
+
   private EntityCombinedSchema(Schema schema, SchemaEntity schemaEntity) {
     this.schema = schema;
     this.schemaEntity = schemaEntity;
+    this.imported = false;
   }
 
   public static EntityCombinedSchema of(Schema schema, SchemaEntity schemaEntity) {
@@ -42,6 +52,11 @@ public EntityCombinedSchema withHiddenPropertiesSet(Set<String> hiddenProperties
     return this;
   }
 
+  public EntityCombinedSchema withImported(boolean imported) {
+    this.imported = imported;
+    return this;
+  }
+
   @Override
   public String name() {
     return schema.name();
@@ -73,4 +88,12 @@ public Audit auditInfo() {
         ? schema.auditInfo()
         : mergedAudit.merge(schemaEntity.auditInfo(), true /* overwrite */);
   }
+
+  public boolean imported() {
+    return imported;
+  }
+
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(schema.properties());
+  }
 }

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedTable.java

@@ -5,6 +5,7 @@
 package com.datastrato.gravitino.catalog;
 
 import com.datastrato.gravitino.Audit;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.TableEntity;
 import com.datastrato.gravitino.rel.Column;
@@ -31,9 +32,17 @@ public final class EntityCombinedTable implements Table {
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
 
+  // Field "imported" is used to indicate whether the entity has been imported to Gravitino
+  // managed storage backend. If "imported" is true, it means that storage backend have stored
+  // the correct entity. Otherwise, we should import the external entity to the storage backend.
+  // This is used for tag/access control related purposes, only the imported entities have the
+  // unique id, and based on this id, we can label and control the access to the entities.
+  private boolean imported;
+
   private EntityCombinedTable(Table table, TableEntity tableEntity) {
     this.table = table;
     this.tableEntity = tableEntity;
+    this.imported = false;
   }
 
   public static EntityCombinedTable of(Table table, TableEntity tableEntity) {
@@ -49,6 +58,11 @@ public EntityCombinedTable withHiddenPropertiesSet(Set<String> hiddenProperties)
     return this;
   }
 
+  public EntityCombinedTable withImported(boolean imported) {
+    this.imported = imported;
+    return this;
+  }
+
   @Override
   public String name() {
     return table.name();
@@ -96,6 +110,10 @@ public Index[] index() {
     return table.index();
   }
 
+  public boolean imported() {
+    return imported;
+  }
+
   @Override
   public Audit auditInfo() {
     AuditInfo mergedAudit =
@@ -110,4 +128,8 @@ public Audit auditInfo() {
         ? table.auditInfo()
         : mergedAudit.merge(tableEntity.auditInfo(), true /* overwrite */);
   }
+
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(table.properties());
+  }
 }

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedTopic.java

@@ -5,6 +5,7 @@
 package com.datastrato.gravitino.catalog;
 
 import com.datastrato.gravitino.Audit;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.messaging.Topic;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.TopicEntity;
@@ -24,9 +25,17 @@ public class EntityCombinedTopic implements Topic {
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
 
+  // Field "imported" is used to indicate whether the entity has been imported to Gravitino
+  // managed storage backend. If "imported" is true, it means that storage backend have stored
+  // the correct entity. Otherwise, we should import the external entity to the storage backend.
+  // This is used for tag/access control related purposes, only the imported entities have the
+  // unique id, and based on this id, we can label and control the access to the entities.
+  private boolean imported;
+
   private EntityCombinedTopic(Topic topic, TopicEntity topicEntity) {
     this.topic = topic;
     this.topicEntity = topicEntity;
+    this.imported = false;
   }
 
   public static EntityCombinedTopic of(Topic topic, TopicEntity topicEntity) {
@@ -42,6 +51,11 @@ public EntityCombinedTopic withHiddenPropertiesSet(Set<String> hiddenProperties)
     return this;
   }
 
+  public EntityCombinedTopic withImported(boolean imported) {
+    this.imported = imported;
+    return this;
+  }
+
   @Override
   public String name() {
     return topic.name();
@@ -73,4 +87,12 @@ public Audit auditInfo() {
         ? topic.auditInfo()
         : mergedAudit.merge(topicEntity.auditInfo(), true /* overwrite */);
   }
+
+  public boolean imported() {
+    return imported;
+  }
+
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(topic.properties());
+  }
 }