[ISSUE-3607] Support to sync the entities

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

@@ -99,9 +99,9 @@ public static GravitinoEnv getInstance() {
   }
 
   /**
-   * 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.
+   * This method is used for testing purposes only to set the lock manager for test in package 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.
    */
@@ -121,6 +121,17 @@ public void setAccessControlManager(AccessControlManager accessControlManager) {
     this.accessControlManager = accessControlManager;
   }
 
+  /**
+   * This method is used for testing purposes only to set the access manager for test in package
+   * `com.datastrato.gravitino.server.web.rest`.
+   *
+   * @param catalogDispatcher The catalog dispatcher to be set.
+   */
+  @VisibleForTesting
+  public void setCatalogDispatcher(CatalogDispatcher catalogDispatcher) {
+    this.catalogDispatcher = catalogDispatcher;
+  }
+
   /**
    * This method is used for testing purposes only to set the entity store for test in package
    * `com.datastrato.gravitino.authorization`.

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

@@ -8,6 +8,7 @@
 import com.datastrato.gravitino.Schema;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.SchemaEntity;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -23,6 +24,7 @@ public final class EntityCombinedSchema implements Schema {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  private boolean imported;
 
   private EntityCombinedSchema(Schema schema, SchemaEntity schemaEntity) {
     this.schema = schema;
@@ -42,6 +44,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 +80,12 @@ public Audit auditInfo() {
         ? schema.auditInfo()
         : mergedAudit.merge(schemaEntity.auditInfo(), true /* overwrite */);
   }
+
+  public boolean imported() {
+    return imported;
+  }
+
+  Map<String, String> schemaProperties() {
+    return Collections.unmodifiableMap(schema.properties());
+  }
 }

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

@@ -14,6 +14,7 @@
 import com.datastrato.gravitino.rel.expressions.sorts.SortOrder;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -30,6 +31,7 @@ public final class EntityCombinedTable implements Table {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  private boolean imported;
 
   private EntityCombinedTable(Table table, TableEntity tableEntity) {
     this.table = table;
@@ -49,6 +51,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 +103,10 @@ public Index[] index() {
     return table.index();
   }
 
+  public boolean imported() {
+    return imported;
+  }
+
   @Override
   public Audit auditInfo() {
     AuditInfo mergedAudit =
@@ -110,4 +121,8 @@ public Audit auditInfo() {
         ? table.auditInfo()
         : mergedAudit.merge(tableEntity.auditInfo(), true /* overwrite */);
   }
+
+  Map<String, String> tableProperties() {
+    return Collections.unmodifiableMap(table.properties());
+  }
 }

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

@@ -8,6 +8,7 @@
 import com.datastrato.gravitino.messaging.Topic;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.TopicEntity;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -23,6 +24,7 @@ public class EntityCombinedTopic implements Topic {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  private boolean imported;
 
   private EntityCombinedTopic(Topic topic, TopicEntity topicEntity) {
     this.topic = topic;
@@ -42,6 +44,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 +80,12 @@ public Audit auditInfo() {
         ? topic.auditInfo()
         : mergedAudit.merge(topicEntity.auditInfo(), true /* overwrite */);
   }
+
+  public boolean imported() {
+    return imported;
+  }
+
+  Map<String, String> topicProperties() {
+    return Collections.unmodifiableMap(topic.properties());
+  }
 }

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

@@ -5,6 +5,7 @@
 
 package com.datastrato.gravitino.catalog;
 
+import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.connector.SupportsSchemas;
 
 /**
@@ -13,4 +14,6 @@
  * to dispatching or handling schema-related events or actions that are not covered by the standard
  * {@code SupportsSchemas} operations.
  */
-public interface SchemaDispatcher extends SupportsSchemas {}
+public interface SchemaDispatcher extends SupportsSchemas {
+  boolean importSchema(NameIdentifier identifier);
+}

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

@@ -71,6 +71,12 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
         applyCaseSensitive(ident, Capability.Scope.SCHEMA, dispatcher), cascade);
   }
 
+  @Override
+  public boolean importSchema(NameIdentifier identifier) {
+    return dispatcher.importSchema(
+        applyCaseSensitive(identifier, Capability.Scope.SCHEMA, dispatcher));
+  }
+
   private NameIdentifier normalizeNameIdentifier(NameIdentifier ident) {
     Capability capability = dispatcher.getCatalogCapability(ident);
     return applyCapabilities(ident, Capability.Scope.SCHEMA, capability);