diff --git a/bundles/aliyun/src/main/java/org/apache/gravitino/oss/credential/OSSTokenProvider.java b/bundles/aliyun/src/main/java/org/apache/gravitino/oss/credential/OSSTokenProvider.java
index 04ef0022a10..79d7f51f780 100644
--- a/bundles/aliyun/src/main/java/org/apache/gravitino/oss/credential/OSSTokenProvider.java
+++ b/bundles/aliyun/src/main/java/org/apache/gravitino/oss/credential/OSSTokenProvider.java
@@ -138,9 +138,10 @@ private String createPolicy(Set<String> readLocations, Set<String> writeLocation
             .effect(Effect.ALLOW)
             .addAction("oss:GetObject")
             .addAction("oss:GetObjectVersion");
+
     // Add support for bucket-level policies
     Map<String, Statement.Builder> bucketListStatementBuilder = new HashMap<>();
-    Map<String, Statement.Builder> bucketGetLocationStatementBuilder = new HashMap<>();
+    Map<String, Statement.Builder> bucketMetadataStatementBuilder = new HashMap<>();
 
     String arnPrefix = getArnPrefix();
     Stream.concat(readLocations.stream(), writeLocations.stream())
@@ -150,22 +151,24 @@ private String createPolicy(Set<String> readLocations, Set<String> writeLocation
               URI uri = URI.create(location);
               allowGetObjectStatementBuilder.addResource(getOssUriWithArn(arnPrefix, uri));
               String bucketArn = arnPrefix + getBucketName(uri);
-              // ListBucket
+              // OSS use 'oss:ListObjects' to list objects in a bucket while s3 use 's3:ListBucket'
               bucketListStatementBuilder.computeIfAbsent(
                   bucketArn,
                   key ->
                       Statement.builder()
                           .effect(Effect.ALLOW)
-                          .addAction("oss:ListBucket")
+                          .addAction("oss:ListObjects")
                           .addResource(key)
                           .condition(getCondition(uri)));
-              // GetBucketLocation
-              bucketGetLocationStatementBuilder.computeIfAbsent(
+              // Add get bucket location and bucket info action.
+              bucketMetadataStatementBuilder.computeIfAbsent(
                   bucketArn,
                   key ->
                       Statement.builder()
                           .effect(Effect.ALLOW)
                           .addAction("oss:GetBucketLocation")
+                          // Required for OSS Hadoop connector to get bucket information
+                          .addAction("oss:GetBucketInfo")
                           .addResource(key));
             });
 
@@ -192,7 +195,7 @@ private String createPolicy(Set<String> readLocations, Set<String> writeLocation
       policyBuilder.addStatement(
           Statement.builder().effect(Effect.ALLOW).addAction("oss:ListBucket").build());
     }
-    bucketGetLocationStatementBuilder
+    bucketMetadataStatementBuilder
         .values()
         .forEach(statementBuilder -> policyBuilder.addStatement(statementBuilder.build()));
 
diff --git a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/S3TokenProvider.java b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/S3TokenProvider.java
index 24b88875de9..56d293d046f 100644
--- a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/S3TokenProvider.java
+++ b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/S3TokenProvider.java
@@ -20,6 +20,7 @@
 package org.apache.gravitino.s3.credential;
 
 import java.net.URI;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -49,6 +50,7 @@
 
 /** Generates S3 token to access S3 data. */
 public class S3TokenProvider implements CredentialProvider {
+
   private StsClient stsClient;
   private String roleArn;
   private String externalID;
@@ -134,6 +136,7 @@ private IamPolicy createPolicy(
               allowGetObjectStatementBuilder.addResource(
                   IamResource.create(getS3UriWithArn(arnPrefix, uri)));
               String bucketArn = arnPrefix + getBucketName(uri);
+              String rawPath = trimLeadingSlash(uri.getPath());
               bucketListStatmentBuilder
                   .computeIfAbsent(
                       bucketArn,
@@ -142,10 +145,14 @@ private IamPolicy createPolicy(
                               .effect(IamEffect.ALLOW)
                               .addAction("s3:ListBucket")
                               .addResource(key))
-                  .addCondition(
+                  .addConditions(
                       IamConditionOperator.STRING_LIKE,
                       "s3:prefix",
-                      concatPathWithSep(trimLeadingSlash(uri.getPath()), "*", "/"));
+                      Arrays.asList(
+                          // Get raw path metadata information for AWS hadoop connector
+                          rawPath,
+                          // Listing objects in raw path
+                          concatPathWithSep(rawPath, "*", "/")));
               bucketGetLocationStatmentBuilder.computeIfAbsent(
                   bucketArn,
                   key ->
diff --git a/catalogs/catalog-model/build.gradle.kts b/catalogs/catalog-model/build.gradle.kts
index 95af305fcae..5c125532263 100644
--- a/catalogs/catalog-model/build.gradle.kts
+++ b/catalogs/catalog-model/build.gradle.kts
@@ -29,17 +29,15 @@ dependencies {
     exclude(group = "*")
   }
 
-  implementation(project(":core")) {
+  implementation(project(":catalogs:catalog-common")) {
     exclude(group = "*")
   }
   implementation(project(":common")) {
     exclude(group = "*")
   }
-
-  implementation(project(":catalogs:catalog-common")) {
+  implementation(project(":core")) {
     exclude(group = "*")
   }
-
   implementation(libs.guava)
   implementation(libs.slf4j.api)
 
@@ -47,14 +45,17 @@ dependencies {
   testImplementation(project(":integration-test-common", "testArtifacts"))
   testImplementation(project(":server"))
   testImplementation(project(":server-common"))
-
   testImplementation(libs.bundles.log4j)
   testImplementation(libs.commons.io)
   testImplementation(libs.commons.lang3)
   testImplementation(libs.mockito.core)
   testImplementation(libs.mockito.inline)
+  testImplementation(libs.mysql.driver)
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.junit.jupiter.params)
+  testImplementation(libs.postgresql.driver)
+  testImplementation(libs.testcontainers)
+  testImplementation(libs.testcontainers.mysql)
 
   testRuntimeOnly(libs.junit.jupiter.engine)
 }
@@ -68,8 +69,9 @@ tasks {
   val copyCatalogLibs by registering(Copy::class) {
     dependsOn("jar", "runtimeJars")
     from("build/libs") {
-      exclude("slf4j-*.jar")
       exclude("guava-*.jar")
+      exclude("log4j-*.jar")
+      exclude("slf4j-*.jar")
     }
     into("$rootDir/distribution/package/catalogs/model/libs")
   }
diff --git a/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/integration/test/ModelCatalogOperationsIT.java b/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/integration/test/ModelCatalogOperationsIT.java
new file mode 100644
index 00000000000..6e7adac5516
--- /dev/null
+++ b/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/integration/test/ModelCatalogOperationsIT.java
@@ -0,0 +1,364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravtitino.catalog.model.integration.test;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.exceptions.ModelAlreadyExistsException;
+import org.apache.gravitino.exceptions.ModelVersionAliasesAlreadyExistException;
+import org.apache.gravitino.exceptions.NoSuchModelException;
+import org.apache.gravitino.exceptions.NoSuchModelVersionException;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.model.Model;
+import org.apache.gravitino.model.ModelVersion;
+import org.apache.gravitino.utils.RandomNameUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class ModelCatalogOperationsIT extends BaseIT {
+
+  private final String metalakeName = RandomNameUtils.genRandomName("model_it_metalake");
+  private final String catalogName = RandomNameUtils.genRandomName("model_it_catalog");
+  private final String schemaName = RandomNameUtils.genRandomName("model_it_schema");
+
+  private GravitinoMetalake gravitinoMetalake;
+  private Catalog gravitinoCatalog;
+
+  @BeforeAll
+  public void setUp() {
+    createMetalake();
+    createCatalog();
+  }
+
+  @AfterAll
+  public void tearDown() {
+    gravitinoMetalake.dropCatalog(catalogName, true);
+    client.dropMetalake(metalakeName, true);
+  }
+
+  @BeforeEach
+  public void beforeEach() {
+    createSchema();
+  }
+
+  @AfterEach
+  public void afterEach() {
+    dropSchema();
+  }
+
+  @Test
+  public void testRegisterAndGetModel() {
+    String modelName = RandomNameUtils.genRandomName("model1");
+    NameIdentifier modelIdent = NameIdentifier.of(schemaName, modelName);
+    String comment = "comment";
+    Map<String, String> properties = ImmutableMap.of("key1", "val1", "key2", "val2");
+
+    Model model = gravitinoCatalog.asModelCatalog().registerModel(modelIdent, comment, properties);
+    Assertions.assertEquals(modelName, model.name());
+    Assertions.assertEquals(comment, model.comment());
+    Assertions.assertEquals(properties, model.properties());
+
+    Model loadModel = gravitinoCatalog.asModelCatalog().getModel(modelIdent);
+    Assertions.assertEquals(modelName, loadModel.name());
+    Assertions.assertEquals(comment, loadModel.comment());
+    Assertions.assertEquals(properties, loadModel.properties());
+
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().modelExists(modelIdent));
+
+    // Test register existing model
+    Assertions.assertThrows(
+        ModelAlreadyExistsException.class,
+        () -> gravitinoCatalog.asModelCatalog().registerModel(modelIdent, comment, properties));
+
+    // Test register model in a non-existent schema
+    NameIdentifier nonExistentSchemaIdent = NameIdentifier.of("non_existent_schema", modelName);
+    Assertions.assertThrows(
+        NoSuchSchemaException.class,
+        () ->
+            gravitinoCatalog
+                .asModelCatalog()
+                .registerModel(nonExistentSchemaIdent, comment, properties));
+
+    // Test get non-existent model
+    NameIdentifier nonExistentModelIdent = NameIdentifier.of(schemaName, "non_existent_model");
+    Assertions.assertThrows(
+        NoSuchModelException.class,
+        () -> gravitinoCatalog.asModelCatalog().getModel(nonExistentModelIdent));
+
+    // Test get model from non-existent schema
+    NameIdentifier nonExistentModelIdent2 = NameIdentifier.of("non_existent_schema", modelName);
+    Assertions.assertThrows(
+        NoSuchModelException.class,
+        () -> gravitinoCatalog.asModelCatalog().getModel(nonExistentModelIdent2));
+  }
+
+  @Test
+  public void testRegisterAndListModels() {
+    String modelName1 = RandomNameUtils.genRandomName("model1");
+    String modelName2 = RandomNameUtils.genRandomName("model2");
+    NameIdentifier modelIdent1 = NameIdentifier.of(schemaName, modelName1);
+    NameIdentifier modelIdent2 = NameIdentifier.of(schemaName, modelName2);
+
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent1, null, null);
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent2, null, null);
+
+    NameIdentifier[] models =
+        gravitinoCatalog.asModelCatalog().listModels(Namespace.of(schemaName));
+    Set<NameIdentifier> resultSet = Sets.newHashSet(models);
+
+    Assertions.assertEquals(2, resultSet.size());
+    Assertions.assertTrue(resultSet.contains(modelIdent1));
+    Assertions.assertTrue(resultSet.contains(modelIdent2));
+
+    // Test delete and list models
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModel(modelIdent1));
+    NameIdentifier[] modelsAfterDelete =
+        gravitinoCatalog.asModelCatalog().listModels(Namespace.of(schemaName));
+
+    Assertions.assertEquals(1, modelsAfterDelete.length);
+    Assertions.assertEquals(modelIdent2, modelsAfterDelete[0]);
+
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModel(modelIdent2));
+    NameIdentifier[] modelsAfterDeleteAll =
+        gravitinoCatalog.asModelCatalog().listModels(Namespace.of(schemaName));
+
+    Assertions.assertEquals(0, modelsAfterDeleteAll.length);
+
+    // Test list models from non-existent schema
+    Assertions.assertThrows(
+        NoSuchSchemaException.class,
+        () -> gravitinoCatalog.asModelCatalog().listModels(Namespace.of("non_existent_schema")));
+  }
+
+  @Test
+  public void testRegisterAndDeleteModel() {
+    String modelName = RandomNameUtils.genRandomName("model1");
+    NameIdentifier modelIdent = NameIdentifier.of(schemaName, modelName);
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent, null, null);
+
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModel(modelIdent));
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().modelExists(modelIdent));
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().deleteModel(modelIdent));
+
+    // Test delete non-existent model
+    NameIdentifier nonExistentModelIdent = NameIdentifier.of(schemaName, "non_existent_model");
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().deleteModel(nonExistentModelIdent));
+
+    // Test delete model from non-existent schema
+    NameIdentifier nonExistentSchemaIdent = NameIdentifier.of("non_existent_schema", modelName);
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().deleteModel(nonExistentSchemaIdent));
+  }
+
+  @Test
+  public void testLinkAndGerModelVersion() {
+    String modelName = RandomNameUtils.genRandomName("model1");
+    Map<String, String> properties = ImmutableMap.of("key1", "val1", "key2", "val2");
+    NameIdentifier modelIdent = NameIdentifier.of(schemaName, modelName);
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent, null, null);
+
+    gravitinoCatalog
+        .asModelCatalog()
+        .linkModelVersion(modelIdent, "uri", new String[] {"alias1"}, "comment", properties);
+
+    ModelVersion modelVersion =
+        gravitinoCatalog.asModelCatalog().getModelVersion(modelIdent, "alias1");
+
+    Assertions.assertEquals(0, modelVersion.version());
+    Assertions.assertEquals("uri", modelVersion.uri());
+    Assertions.assertArrayEquals(new String[] {"alias1"}, modelVersion.aliases());
+    Assertions.assertEquals("comment", modelVersion.comment());
+    Assertions.assertEquals(properties, modelVersion.properties());
+    Assertions.assertTrue(
+        gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, "alias1"));
+
+    ModelVersion modelVersion1 = gravitinoCatalog.asModelCatalog().getModelVersion(modelIdent, 0);
+
+    Assertions.assertEquals(0, modelVersion1.version());
+    Assertions.assertEquals("uri", modelVersion1.uri());
+    Assertions.assertArrayEquals(new String[] {"alias1"}, modelVersion1.aliases());
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, 0));
+
+    // Test link a version to a non-existent model
+    NameIdentifier nonExistentModelIdent = NameIdentifier.of(schemaName, "non_existent_model");
+    Assertions.assertThrows(
+        NoSuchModelException.class,
+        () ->
+            gravitinoCatalog
+                .asModelCatalog()
+                .linkModelVersion(
+                    nonExistentModelIdent, "uri", new String[] {"alias1"}, "comment", properties));
+
+    // Test link a version using existing alias
+    Assertions.assertThrows(
+        ModelVersionAliasesAlreadyExistException.class,
+        () ->
+            gravitinoCatalog
+                .asModelCatalog()
+                .linkModelVersion(
+                    modelIdent, "uri", new String[] {"alias1"}, "comment", properties));
+
+    // Test get non-existent model version
+    Assertions.assertThrows(
+        NoSuchModelVersionException.class,
+        () -> gravitinoCatalog.asModelCatalog().getModelVersion(modelIdent, "non_existent_alias"));
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, "non_existent_alias"));
+
+    Assertions.assertThrows(
+        NoSuchModelVersionException.class,
+        () -> gravitinoCatalog.asModelCatalog().getModelVersion(modelIdent, 1));
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, 1));
+  }
+
+  @Test
+  public void testLinkAndDeleteModelVersions() {
+    String modelName = RandomNameUtils.genRandomName("model1");
+    NameIdentifier modelIdent = NameIdentifier.of(schemaName, modelName);
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent, null, null);
+
+    gravitinoCatalog
+        .asModelCatalog()
+        .linkModelVersion(modelIdent, "uri1", new String[] {"alias1"}, "comment1", null);
+    gravitinoCatalog
+        .asModelCatalog()
+        .linkModelVersion(modelIdent, "uri2", new String[] {"alias2"}, "comment2", null);
+
+    Assertions.assertTrue(
+        gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, "alias1"));
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, "alias1"));
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, 0));
+
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, 1));
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().modelVersionExists(modelIdent, "alias2"));
+    Assertions.assertFalse(gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, 1));
+
+    // Test delete non-existent model version
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, "non_existent_alias"));
+
+    // Test delete model version of non-existent model
+    NameIdentifier nonExistentModelIdent = NameIdentifier.of(schemaName, "non_existent_model");
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().deleteModelVersion(nonExistentModelIdent, "alias1"));
+
+    // Test delete model version of non-existent schema
+    NameIdentifier nonExistentSchemaIdent = NameIdentifier.of("non_existent_schema", modelName);
+    Assertions.assertFalse(
+        gravitinoCatalog.asModelCatalog().deleteModelVersion(nonExistentSchemaIdent, "alias1"));
+  }
+
+  @Test
+  public void testLinkAndListModelVersions() {
+    String modelName = RandomNameUtils.genRandomName("model1");
+    NameIdentifier modelIdent = NameIdentifier.of(schemaName, modelName);
+    gravitinoCatalog.asModelCatalog().registerModel(modelIdent, null, null);
+
+    gravitinoCatalog
+        .asModelCatalog()
+        .linkModelVersion(modelIdent, "uri1", new String[] {"alias1"}, "comment1", null);
+    gravitinoCatalog
+        .asModelCatalog()
+        .linkModelVersion(modelIdent, "uri2", new String[] {"alias2"}, "comment2", null);
+
+    int[] modelVersions = gravitinoCatalog.asModelCatalog().listModelVersions(modelIdent);
+    Set<Integer> resultSet = Arrays.stream(modelVersions).boxed().collect(Collectors.toSet());
+
+    Assertions.assertEquals(2, resultSet.size());
+    Assertions.assertTrue(resultSet.contains(0));
+    Assertions.assertTrue(resultSet.contains(1));
+
+    // Test list model versions of non-existent model
+    NameIdentifier nonExistentModelIdent = NameIdentifier.of(schemaName, "non_existent_model");
+    Assertions.assertThrows(
+        NoSuchModelException.class,
+        () -> gravitinoCatalog.asModelCatalog().listModelVersions(nonExistentModelIdent));
+
+    // Test list model versions of non-existent schema
+    NameIdentifier nonExistentSchemaIdent = NameIdentifier.of("non_existent_schema", modelName);
+    Assertions.assertThrows(
+        NoSuchModelException.class,
+        () -> gravitinoCatalog.asModelCatalog().listModelVersions(nonExistentSchemaIdent));
+
+    // Test delete and list model versions
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, 1));
+    int[] modelVersionsAfterDelete =
+        gravitinoCatalog.asModelCatalog().listModelVersions(modelIdent);
+
+    Assertions.assertEquals(1, modelVersionsAfterDelete.length);
+    Assertions.assertEquals(0, modelVersionsAfterDelete[0]);
+
+    Assertions.assertTrue(gravitinoCatalog.asModelCatalog().deleteModelVersion(modelIdent, 0));
+    int[] modelVersionsAfterDeleteAll =
+        gravitinoCatalog.asModelCatalog().listModelVersions(modelIdent);
+
+    Assertions.assertEquals(0, modelVersionsAfterDeleteAll.length);
+  }
+
+  private void createMetalake() {
+    GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
+    Assertions.assertEquals(0, gravitinoMetalakes.length);
+
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
+
+    gravitinoMetalake = loadMetalake;
+  }
+
+  private void createCatalog() {
+    gravitinoMetalake.createCatalog(catalogName, Catalog.Type.MODEL, "comment", ImmutableMap.of());
+    gravitinoCatalog = gravitinoMetalake.loadCatalog(catalogName);
+  }
+
+  private void createSchema() {
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put("key1", "val1");
+    properties.put("key2", "val2");
+    String comment = "comment";
+
+    gravitinoCatalog.asSchemas().createSchema(schemaName, comment, properties);
+    Schema loadSchema = gravitinoCatalog.asSchemas().loadSchema(schemaName);
+    Assertions.assertEquals(schemaName, loadSchema.name());
+    Assertions.assertEquals(comment, loadSchema.comment());
+    Assertions.assertEquals("val1", loadSchema.properties().get("key1"));
+    Assertions.assertEquals("val2", loadSchema.properties().get("key2"));
+  }
+
+  private void dropSchema() {
+    gravitinoCatalog.asSchemas().dropSchema(schemaName, true);
+  }
+}
diff --git a/catalogs/catalog-model/src/test/resources/log4j2.properties b/catalogs/catalog-model/src/test/resources/log4j2.properties
new file mode 100644
index 00000000000..88da637c15d
--- /dev/null
+++ b/catalogs/catalog-model/src/test/resources/log4j2.properties
@@ -0,0 +1,73 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Set to debug or trace if log4j initialization is failing
+status = info
+
+# Name of the configuration
+name = ConsoleLogConfig
+
+# Console appender configuration
+appender.console.type = Console
+appender.console.name = consoleLogger
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p [%t] %c{1}:%L - %m%n
+
+# Log files location
+property.logPath = ${sys:gravitino.log.path:-build/catalog-model-integration-test.log}
+
+# File appender configuration
+appender.file.type = File
+appender.file.name = fileLogger
+appender.file.fileName = ${logPath}
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5p %c - %m%n
+
+# Root logger level
+rootLogger.level = info
+
+# Root logger referring to console and file appenders
+rootLogger.appenderRef.stdout.ref = consoleLogger
+rootLogger.appenderRef.file.ref = fileLogger
+
+# File appender configuration for testcontainers
+appender.testcontainersFile.type = File
+appender.testcontainersFile.name = testcontainersLogger
+appender.testcontainersFile.fileName = build/testcontainers.log
+appender.testcontainersFile.layout.type = PatternLayout
+appender.testcontainersFile.layout.pattern = %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5p %c - %m%n
+
+# Logger for testcontainers
+logger.testcontainers.name = org.testcontainers
+logger.testcontainers.level = debug
+logger.testcontainers.additivity = false
+logger.testcontainers.appenderRef.file.ref = testcontainersLogger
+
+logger.tc.name = tc
+logger.tc.level = debug
+logger.tc.additivity = false
+logger.tc.appenderRef.file.ref = testcontainersLogger
+
+logger.docker.name = com.github.dockerjava
+logger.docker.level = warn
+logger.docker.additivity = false
+logger.docker.appenderRef.file.ref = testcontainersLogger
+
+logger.http.name = com.github.dockerjava.zerodep.shaded.org.apache.hc.client5.http.wire
+logger.http.level = off
diff --git a/clients/client-python/gravitino/api/types/types.py b/clients/client-python/gravitino/api/types/types.py
index 63684211a9a..b204fa82ad7 100644
--- a/clients/client-python/gravitino/api/types/types.py
+++ b/clients/client-python/gravitino/api/types/types.py
@@ -35,7 +35,7 @@ class Types:
     class NullType(Type):
         """The data type representing `NULL` values."""
 
-        _instance: "NullType" = None
+        _instance: Types.NullType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -43,7 +43,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "NullType":
+        def get(cls) -> Types.NullType:
             return cls()
 
         def name(self) -> Name:
@@ -55,7 +55,7 @@ def simple_string(self) -> str:
     class BooleanType(PrimitiveType):
         """The boolean type in Gravitino."""
 
-        _instance: "BooleanType" = None
+        _instance: Types.BooleanType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -63,7 +63,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "BooleanType":
+        def get(cls) -> Types.BooleanType:
             return cls()
 
         def name(self) -> Name:
@@ -75,8 +75,8 @@ def simple_string(self) -> str:
     class ByteType(IntegralType):
         """The byte type in Gravitino."""
 
-        _instance: "ByteType" = None
-        _unsigned_instance: "ByteType" = None
+        _instance: Types.ByteType = None
+        _unsigned_instance: Types.ByteType = None
 
         def __new__(cls, signed: bool = True):
             if signed:
@@ -90,11 +90,11 @@ def __new__(cls, signed: bool = True):
             return cls._unsigned_instance
 
         @classmethod
-        def get(cls) -> "ByteType":
+        def get(cls) -> Types.ByteType:
             return cls(True)
 
         @classmethod
-        def unsigned(cls) -> "ByteType":
+        def unsigned(cls) -> Types.ByteType:
             return cls(False)
 
         def name(self) -> Name:
@@ -104,8 +104,8 @@ def simple_string(self) -> str:
             return "byte" if self.signed() else "byte unsigned"
 
     class ShortType(IntegralType):
-        _instance: "ShortType" = None
-        _unsigned_instance: "ShortType" = None
+        _instance: Types.ShortType = None
+        _unsigned_instance: Types.ShortType = None
 
         def __new__(cls, signed=True):
             if signed:
@@ -119,11 +119,11 @@ def __new__(cls, signed=True):
             return cls._unsigned_instance
 
         @classmethod
-        def get(cls) -> "ShortType":
+        def get(cls) -> Types.ShortType:
             return cls(True)
 
         @classmethod
-        def unsigned(cls) -> "ShortType":
+        def unsigned(cls) -> Types.ShortType:
             return cls(False)
 
         def name(self) -> Name:
@@ -133,8 +133,8 @@ def simple_string(self) -> str:
             return "short" if self.signed() else "short unsigned"
 
     class IntegerType(IntegralType):
-        _instance: "IntegerType" = None
-        _unsigned_instance: "IntegerType" = None
+        _instance: Types.IntegerType = None
+        _unsigned_instance: Types.IntegerType = None
 
         def __new__(cls, signed=True):
             if signed:
@@ -148,7 +148,7 @@ def __new__(cls, signed=True):
             return cls._unsigned_instance
 
         @classmethod
-        def get(cls) -> "IntegerType":
+        def get(cls) -> Types.IntegerType:
             return cls(True)
 
         @classmethod
@@ -162,8 +162,8 @@ def simple_string(self) -> str:
             return "integer" if self.signed() else "integer unsigned"
 
     class LongType(IntegralType):
-        _instance: "LongType" = None
-        _unsigned_instance: "LongType" = None
+        _instance: Types.LongType = None
+        _unsigned_instance: Types.LongType = None
 
         def __new__(cls, signed=True):
             if signed:
@@ -177,7 +177,7 @@ def __new__(cls, signed=True):
             return cls._unsigned_instance
 
         @classmethod
-        def get(cls) -> "LongType":
+        def get(cls) -> Types.LongType:
             return cls(True)
 
         @classmethod
@@ -191,7 +191,7 @@ def simple_string(self) -> str:
             return "long" if self.signed() else "long unsigned"
 
     class FloatType(FractionType):
-        _instance: "FloatType" = None
+        _instance: Types.FloatType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -200,7 +200,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "FloatType":
+        def get(cls) -> Types.FloatType:
             return cls()
 
         def name(self) -> Name:
@@ -210,7 +210,7 @@ def simple_string(self) -> str:
             return "float"
 
     class DoubleType(FractionType):
-        _instance: "DoubleType" = None
+        _instance: Types.DoubleType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -219,7 +219,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "DoubleType":
+        def get(cls) -> Types.DoubleType:
             return cls()
 
         def name(self) -> Name:
@@ -265,7 +265,7 @@ def check_precision_scale(precision: int, scale: int):
                 )
 
         @classmethod
-        def of(cls, precision: int, scale: int) -> "DecimalType":
+        def of(cls, precision: int, scale: int) -> Types.DecimalType:
             return cls(precision, scale)
 
         def name(self) -> Name:
@@ -300,7 +300,7 @@ def __hash__(self):
     class DateType(DateTimeType):
         """The date time type in Gravitino."""
 
-        _instance: "DateType" = None
+        _instance: Types.DateType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -309,7 +309,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "DateType":
+        def get(cls) -> Types.DateType:
             return cls()
 
         def name(self) -> Name:
@@ -319,7 +319,7 @@ def simple_string(self) -> str:
             return "date"
 
     class TimeType(DateTimeType):
-        _instance: "TimeType" = None
+        _instance: Types.TimeType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -328,7 +328,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "TimeType":
+        def get(cls) -> Types.TimeType:
             return cls()
 
         def name(self) -> Name:
@@ -338,8 +338,8 @@ def simple_string(self) -> str:
             return "time"
 
     class TimestampType(DateTimeType):
-        _instance_with_tz: "TimestampType" = None
-        _instance_without_tz: "TimestampType" = None
+        _instance_with_tz: Types.TimestampType = None
+        _instance_without_tz: Types.TimestampType = None
         _with_time_zone: bool
 
         def __new__(cls, with_time_zone: bool):
@@ -354,11 +354,11 @@ def __new__(cls, with_time_zone: bool):
             return cls._instance_without_tz
 
         @classmethod
-        def with_time_zone(cls) -> "TimestampType":
+        def with_time_zone(cls) -> Types.TimestampType:
             return cls(True)
 
         @classmethod
-        def without_time_zone(cls) -> "TimestampType":
+        def without_time_zone(cls) -> Types.TimestampType:
             return cls(False)
 
         def __init__(self, with_time_zone: bool):
@@ -377,7 +377,7 @@ def simple_string(self) -> str:
     class IntervalYearType(IntervalType):
         """The interval year type in Gravitino."""
 
-        _instance: "IntervalYearType" = None
+        _instance: Types.IntervalYearType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -386,7 +386,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "IntervalYearType":
+        def get(cls) -> Types.IntervalYearType:
             return cls()
 
         def name(self) -> Name:
@@ -398,7 +398,7 @@ def simple_string(self) -> str:
     class IntervalDayType(IntervalType):
         """The interval day type in Gravitino."""
 
-        _instance: "IntervalDayType" = None
+        _instance: Types.IntervalDayType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -407,7 +407,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "IntervalDayType":
+        def get(cls) -> Types.IntervalDayType:
             return cls()
 
         def name(self) -> Name:
@@ -420,7 +420,7 @@ class StringType(PrimitiveType):
         """The string type in Gravitino, equivalent to varchar(MAX),
         which the MAX is determined by the underlying catalog."""
 
-        _instance: "StringType" = None
+        _instance: Types.StringType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -429,7 +429,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "StringType":
+        def get(cls) -> Types.StringType:
             return cls()
 
         def name(self) -> Name:
@@ -441,7 +441,7 @@ def simple_string(self) -> str:
     class UUIDType(PrimitiveType):
         """The uuid type in Gravitino."""
 
-        _instance: "UUIDType" = None
+        _instance: Types.UUIDType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -450,7 +450,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "UUIDType":
+        def get(cls) -> Types.UUIDType:
             return cls()
 
         def name(self) -> Name:
@@ -475,7 +475,7 @@ def __init__(self, length: int):
             self._length = length
 
         @classmethod
-        def of(cls, length: int) -> "FixedType":
+        def of(cls, length: int) -> Types.FixedType:
             """
             Args:
                 length: The length of the fixed type.
@@ -520,7 +520,7 @@ def __init__(self, length: int):
             self._length = length
 
         @classmethod
-        def of(cls, length: int) -> "VarCharType":
+        def of(cls, length: int) -> Types.VarCharType:
             return cls(length)
 
         def name(self) -> Name:
@@ -558,7 +558,7 @@ def __init__(self, length: int):
             self._length = length
 
         @classmethod
-        def of(cls, length: int) -> "FixedCharType":
+        def of(cls, length: int) -> Types.FixedCharType:
             return cls(length)
 
         def name(self) -> Name:
@@ -579,7 +579,7 @@ def __hash__(self):
             return hash(self._length)
 
     class BinaryType(PrimitiveType):
-        _instance: "BinaryType" = None
+        _instance: Types.BinaryType = None
 
         def __new__(cls):
             if cls._instance is None:
@@ -588,7 +588,7 @@ def __new__(cls):
             return cls._instance
 
         @classmethod
-        def get(cls) -> "BinaryType":
+        def get(cls) -> Types.BinaryType:
             return cls()
 
         def name(self) -> Name:
@@ -601,15 +601,15 @@ class StructType(ComplexType):
         """The struct type in Gravitino.
         Note, this type is not supported in the current version of Gravitino."""
 
-        _fields: List["Field"]
+        _fields: List[Field]
 
-        def __init__(self, fields: List["Field"]):
+        def __init__(self, fields: List[Field]):
             if not fields or len(fields) == 0:
                 raise ValueError("fields cannot be null or empty")
             self._fields = fields
 
         @classmethod
-        def of(cls, *fields) -> "StructType":
+        def of(cls, *fields) -> Types.StructType:
             """
             Args:
                 fields: The fields of the struct type.
@@ -617,9 +617,9 @@ def of(cls, *fields) -> "StructType":
             Returns:
                 A StructType instance with the given fields.
             """
-            return cls(fields)
+            return cls(list(fields))
 
-        def fields(self) -> List["Field"]:
+        def fields(self) -> List[Field]:
             return self._fields
 
         def name(self) -> Name:
@@ -677,7 +677,7 @@ def __init__(
             @classmethod
             def not_null_field(
                 cls, name: str, field_type: Type, comment: str = None
-            ) -> "Field":
+            ) -> Types.StructType.Field:
                 """
                 Args:
                     name: The name of the field.
@@ -689,7 +689,7 @@ def not_null_field(
             @classmethod
             def nullable_field(
                 cls, name: str, field_type: Type, comment: str = None
-            ) -> "Field":
+            ) -> Types.StructType.Field:
                 """
                 Args:
                     name: The name of the field.
@@ -760,7 +760,7 @@ def __init__(self, element_type: Type, element_nullable: bool):
             self._element_nullable = element_nullable
 
         @classmethod
-        def nullable(cls, element_type: Type) -> "ListType":
+        def nullable(cls, element_type: Type) -> Types.ListType:
             """
             Create a new ListType with the given element type and the type is nullable.
 
@@ -773,7 +773,7 @@ def nullable(cls, element_type: Type) -> "ListType":
             return cls.of(element_type, True)
 
         @classmethod
-        def not_null(cls, element_type: Type) -> "ListType":
+        def not_null(cls, element_type: Type) -> Types.ListType:
             """
             Create a new ListType with the given element type.
 
@@ -786,7 +786,7 @@ def not_null(cls, element_type: Type) -> "ListType":
             return cls.of(element_type, False)
 
         @classmethod
-        def of(cls, element_type: Type, element_nullable: bool) -> "ListType":
+        def of(cls, element_type: Type, element_nullable: bool) -> Types.ListType:
             """
             Create a new ListType with the given element type and whether the element is nullable.
 
@@ -847,7 +847,7 @@ def __init__(self, key_type: Type, value_type: Type, value_nullable: bool):
             self._value_nullable = value_nullable
 
         @classmethod
-        def value_nullable(cls, key_type: Type, value_type: Type) -> "MapType":
+        def value_nullable(cls, key_type: Type, value_type: Type) -> Types.MapType:
             """
             Create a new MapType with the given key type, value type, and the value is nullable.
 
@@ -861,7 +861,7 @@ def value_nullable(cls, key_type: Type, value_type: Type) -> "MapType":
             return cls.of(key_type, value_type, True)
 
         @classmethod
-        def value_not_null(cls, key_type: Type, value_type: Type) -> "MapType":
+        def value_not_null(cls, key_type: Type, value_type: Type) -> Types.MapType:
             """
             Create a new MapType with the given key type, value type, and the value is not nullable.
 
@@ -877,7 +877,7 @@ def value_not_null(cls, key_type: Type, value_type: Type) -> "MapType":
         @classmethod
         def of(
             cls, key_type: Type, value_type: Type, value_nullable: bool
-        ) -> "MapType":
+        ) -> Types.MapType:
             """
             Create a new MapType with the given key type, value type, and whether the value is nullable.
 
@@ -942,7 +942,7 @@ def __init__(self, types: list[Type]):
             self._types = types
 
         @classmethod
-        def of(cls, *types: Type) -> "UnionType":
+        def of(cls, *types: Type) -> Types.UnionType:
             """
             Create a new UnionType with the given types.
 
@@ -995,7 +995,7 @@ def __init__(self, unparsed_type: str):
             self._unparsed_type = unparsed_type
 
         @classmethod
-        def of(cls, unparsed_type: str) -> "UnparsedType":
+        def of(cls, unparsed_type: str) -> Types.UnparsedType:
             """
             Creates a new unparsed_type with the given unparsed type.
 
@@ -1051,7 +1051,7 @@ def __init__(self, catalog_string: str):
             self._catalog_string = catalog_string
 
         @classmethod
-        def of(cls, catalog_string: str) -> "ExternalType":
+        def of(cls, catalog_string: str) -> Types.ExternalType:
             """
             Creates a new ExternalType with the given catalog string.
 
diff --git a/clients/client-python/gravitino/client/generic_model_catalog.py b/clients/client-python/gravitino/client/generic_model_catalog.py
index 6077b0b6429..ca6b5cd31fb 100644
--- a/clients/client-python/gravitino/client/generic_model_catalog.py
+++ b/clients/client-python/gravitino/client/generic_model_catalog.py
@@ -199,7 +199,7 @@ def list_model_versions(self, model_ident: NameIdentifier) -> List[int]:
 
         model_full_ident = self._model_full_identifier(model_ident)
         resp = self.rest_client.get(
-            self._format_model_version_request_path(model_full_ident),
+            f"{self._format_model_version_request_path(model_full_ident)}/versions",
             error_handler=MODEL_ERROR_HANDLER,
         )
         model_version_list_resp = ModelVersionListResponse.from_json(
diff --git a/clients/client-python/gravitino/client/gravitino_metalake.py b/clients/client-python/gravitino/client/gravitino_metalake.py
index 0f72bfc0749..be502aae705 100644
--- a/clients/client-python/gravitino/client/gravitino_metalake.py
+++ b/clients/client-python/gravitino/client/gravitino_metalake.py
@@ -128,7 +128,9 @@ def create_catalog(
         Args:
             name: The name of the catalog.
             catalog_type: The type of the catalog.
-            provider: The provider of the catalog.
+            provider: The provider of the catalog. This parameter can be None if the catalog
+            provides a managed implementation. Currently, only the model catalog supports None
+            provider. For the details, please refer to the Catalog.Type.
             comment: The comment of the catalog.
             properties: The properties of the catalog.
 
diff --git a/clients/client-python/tests/integration/integration_test_env.py b/clients/client-python/tests/integration/integration_test_env.py
index 9344ff93a26..308303e8a7d 100644
--- a/clients/client-python/tests/integration/integration_test_env.py
+++ b/clients/client-python/tests/integration/integration_test_env.py
@@ -67,7 +67,10 @@ class IntegrationTestEnv(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        if os.environ.get("START_EXTERNAL_GRAVITINO") is not None:
+        if (
+            os.environ.get("START_EXTERNAL_GRAVITINO") is not None
+            and os.environ.get("START_EXTERNAL_GRAVITINO").lower() == "true"
+        ):
             # Maybe Gravitino server already startup by Gradle test command or developer manual startup.
             if not check_gravitino_server_status():
                 logger.error("ERROR: Can't find online Gravitino server!")
@@ -112,7 +115,10 @@ def setUpClass(cls):
 
     @classmethod
     def tearDownClass(cls):
-        if os.environ.get("START_EXTERNAL_GRAVITINO") is not None:
+        if (
+            os.environ.get("START_EXTERNAL_GRAVITINO") is not None
+            and os.environ.get("START_EXTERNAL_GRAVITINO").lower() == "true"
+        ):
             return
 
         logger.info("Stop integration test environment...")
diff --git a/clients/client-python/tests/integration/test_model_catalog.py b/clients/client-python/tests/integration/test_model_catalog.py
new file mode 100644
index 00000000000..35ebfdc4726
--- /dev/null
+++ b/clients/client-python/tests/integration/test_model_catalog.py
@@ -0,0 +1,403 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from random import randint
+
+from gravitino import GravitinoAdminClient, GravitinoClient, Catalog, NameIdentifier
+from gravitino.exceptions.base import (
+    ModelAlreadyExistsException,
+    NoSuchSchemaException,
+    NoSuchModelException,
+    ModelVersionAliasesAlreadyExistException,
+    NoSuchModelVersionException,
+)
+from gravitino.namespace import Namespace
+from tests.integration.integration_test_env import IntegrationTestEnv
+
+
+class TestModelCatalog(IntegrationTestEnv):
+
+    _metalake_name: str = "model_it_metalake" + str(randint(0, 1000))
+    _catalog_name: str = "model_it_catalog" + str(randint(0, 1000))
+    _schema_name: str = "model_it_schema" + str(randint(0, 1000))
+
+    _gravitino_admin_client: GravitinoAdminClient = None
+    _gravitino_client: GravitinoClient = None
+    _catalog: Catalog = None
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+        cls._gravitino_admin_client = GravitinoAdminClient(uri="http://localhost:8090")
+        cls._gravitino_admin_client.create_metalake(
+            cls._metalake_name, comment="comment", properties={}
+        )
+
+        cls._gravitino_client = GravitinoClient(
+            uri="http://localhost:8090", metalake_name=cls._metalake_name
+        )
+        cls._catalog = cls._gravitino_client.create_catalog(
+            name=cls._catalog_name,
+            catalog_type=Catalog.Type.MODEL,
+            provider=None,
+            comment="comment",
+            properties={},
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._gravitino_client.drop_catalog(name=cls._catalog_name, force=True)
+        cls._gravitino_admin_client.drop_metalake(name=cls._metalake_name, force=True)
+
+        super().tearDownClass()
+
+    def setUp(self):
+        self._catalog.as_schemas().create_schema(self._schema_name, "comment", {})
+
+    def tearDown(self):
+        self._catalog.as_schemas().drop_schema(self._schema_name, True)
+
+    def test_register_get_model(self):
+        model_name = "model_it_model" + str(randint(0, 1000))
+        model_ident = NameIdentifier.of(self._schema_name, model_name)
+        comment = "comment"
+        properties = {"k1": "v1", "k2": "v2"}
+
+        model = self._catalog.as_model_catalog().register_model(
+            model_ident, comment, properties
+        )
+        self.assertEqual(model_name, model.name())
+        self.assertEqual(comment, model.comment())
+        self.assertEqual(0, model.latest_version())
+        self.assertEqual(properties, model.properties())
+
+        # Test register model without comment and properties
+        model = self._catalog.as_model_catalog().register_model(
+            NameIdentifier.of(
+                self._schema_name, model_name + "_no_comment_no_properties"
+            ),
+            comment=None,
+            properties=None,
+        )
+        self.assertEqual(model_name + "_no_comment_no_properties", model.name())
+        self.assertIsNone(model.comment())
+        self.assertEqual(0, model.latest_version())
+        self.assertEqual({}, model.properties())
+
+        ## Test register same name model again
+        with self.assertRaises(ModelAlreadyExistsException):
+            self._catalog.as_model_catalog().register_model(
+                model_ident, comment, properties
+            )
+
+        # Test register model in a non-existent schema
+        with self.assertRaises(NoSuchSchemaException):
+            self._catalog.as_model_catalog().register_model(
+                NameIdentifier.of("non_existent_schema", model_name),
+                comment,
+                properties,
+            )
+
+        # Test get model
+        model = self._catalog.as_model_catalog().get_model(model_ident)
+        self.assertEqual(model_name, model.name())
+        self.assertEqual(comment, model.comment())
+        self.assertEqual(0, model.latest_version())
+        self.assertEqual(properties, model.properties())
+
+        # Test get non-existent model
+        with self.assertRaises(NoSuchModelException):
+            self._catalog.as_model_catalog().get_model(
+                NameIdentifier.of(self._schema_name, "non_existent_model")
+            )
+
+        # Test get a model for non-existent schema
+        with self.assertRaises(NoSuchModelException):
+            self._catalog.as_model_catalog().get_model(
+                NameIdentifier.of("non_existent_schema", model_name)
+            )
+
+    def test_register_list_models(self):
+
+        model_name1 = "model_it_model1" + str(randint(0, 1000))
+        model_name2 = "model_it_model2" + str(randint(0, 1000))
+        model_ident1 = NameIdentifier.of(self._schema_name, model_name1)
+        model_ident2 = NameIdentifier.of(self._schema_name, model_name2)
+        comment = "comment"
+        properties = {"k1": "v1", "k2": "v2"}
+
+        self._catalog.as_model_catalog().register_model(
+            model_ident1, comment, properties
+        )
+        self._catalog.as_model_catalog().register_model(
+            model_ident2, comment, properties
+        )
+
+        models = self._catalog.as_model_catalog().list_models(
+            Namespace.of(self._schema_name)
+        )
+        self.assertEqual(2, len(models))
+        self.assertTrue(model_ident1 in models)
+        self.assertTrue(model_ident2 in models)
+
+        # Test delete and list models
+        self.assertTrue(self._catalog.as_model_catalog().delete_model(model_ident1))
+        models = self._catalog.as_model_catalog().list_models(
+            Namespace.of(self._schema_name)
+        )
+        self.assertEqual(1, len(models))
+        self.assertTrue(model_ident2 in models)
+
+        self.assertTrue(self._catalog.as_model_catalog().delete_model(model_ident2))
+        models = self._catalog.as_model_catalog().list_models(
+            Namespace.of(self._schema_name)
+        )
+        self.assertEqual(0, len(models))
+
+        # Test list models for non-existent schema
+        with self.assertRaises(NoSuchSchemaException):
+            self._catalog.as_model_catalog().list_models(
+                Namespace.of("non_existent_schema")
+            )
+
+    def test_register_delete_model(self):
+        model_name = "model_it_model" + str(randint(0, 1000))
+        model_ident = NameIdentifier.of(self._schema_name, model_name)
+        comment = "comment"
+        properties = {"k1": "v1", "k2": "v2"}
+
+        self._catalog.as_model_catalog().register_model(
+            model_ident, comment, properties
+        )
+        self.assertTrue(self._catalog.as_model_catalog().delete_model(model_ident))
+        # delete again will return False
+        self.assertFalse(self._catalog.as_model_catalog().delete_model(model_ident))
+
+        # Test delete model in non-existent schema
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model(
+                NameIdentifier.of("non_existent_schema", model_name)
+            )
+        )
+
+        # Test delete non-existent model
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model(
+                NameIdentifier.of(self._schema_name, "non_existent_model")
+            )
+        )
+
+    def test_link_get_model_version(self):
+        model_name = "model_it_model" + str(randint(0, 1000))
+        model_ident = NameIdentifier.of(self._schema_name, model_name)
+        self._catalog.as_model_catalog().register_model(model_ident, "comment", {})
+
+        # Test link model version
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident,
+            uri="uri",
+            aliases=["alias1", "alias2"],
+            comment="comment",
+            properties={"k1": "v1", "k2": "v2"},
+        )
+
+        # Test link model version to a non-existent model
+        with self.assertRaises(NoSuchModelException):
+            self._catalog.as_model_catalog().link_model_version(
+                NameIdentifier.of(self._schema_name, "non_existent_model"),
+                uri="uri",
+                aliases=["alias1", "alias2"],
+                comment="comment",
+                properties={"k1": "v1", "k2": "v2"},
+            )
+
+        # Test link model version with existing aliases
+        with self.assertRaises(ModelVersionAliasesAlreadyExistException):
+            self._catalog.as_model_catalog().link_model_version(
+                model_ident,
+                uri="uri",
+                aliases=["alias1", "alias2"],
+                comment="comment",
+                properties={"k1": "v1", "k2": "v2"},
+            )
+
+        model_version = self._catalog.as_model_catalog().get_model_version(
+            model_ident, 0
+        )
+        self.assertEqual(0, model_version.version())
+        self.assertEqual("uri", model_version.uri())
+        self.assertEqual(["alias1", "alias2"], model_version.aliases())
+        self.assertEqual("comment", model_version.comment())
+        self.assertEqual({"k1": "v1", "k2": "v2"}, model_version.properties())
+
+        model_version = self._catalog.as_model_catalog().get_model_version_by_alias(
+            model_ident, "alias1"
+        )
+        self.assertEqual(0, model_version.version())
+        self.assertEqual("uri", model_version.uri())
+
+        model_version = self._catalog.as_model_catalog().get_model_version_by_alias(
+            model_ident, "alias2"
+        )
+        self.assertEqual(0, model_version.version())
+        self.assertEqual("uri", model_version.uri())
+
+        # Test get model version from non-existent model
+        with self.assertRaises(NoSuchModelVersionException):
+            self._catalog.as_model_catalog().get_model_version(
+                NameIdentifier.of(self._schema_name, "non_existent_model"), 0
+            )
+
+        with self.assertRaises(NoSuchModelVersionException):
+            self._catalog.as_model_catalog().get_model_version_by_alias(
+                NameIdentifier.of(self._schema_name, "non_existent_model"), "alias1"
+            )
+
+        # Test get non-existent model version
+        with self.assertRaises(NoSuchModelVersionException):
+            self._catalog.as_model_catalog().get_model_version(model_ident, 1)
+
+        with self.assertRaises(NoSuchModelVersionException):
+            self._catalog.as_model_catalog().get_model_version_by_alias(
+                model_ident, "non_existent_alias"
+            )
+
+        # Test link model version with None aliases, comment and properties
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident, uri="uri", aliases=None, comment=None, properties=None
+        )
+        model_version = self._catalog.as_model_catalog().get_model_version(
+            model_ident, 1
+        )
+        self.assertEqual(1, model_version.version())
+        self.assertEqual("uri", model_version.uri())
+        self.assertEqual([], model_version.aliases())
+        self.assertIsNone(model_version.comment())
+        self.assertEqual({}, model_version.properties())
+
+    def test_link_list_model_versions(self):
+        model_name = "model_it_model" + str(randint(0, 1000))
+        model_ident = NameIdentifier.of(self._schema_name, model_name)
+        self._catalog.as_model_catalog().register_model(model_ident, "comment", {})
+
+        # Test link model versions
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident,
+            uri="uri1",
+            aliases=["alias1", "alias2"],
+            comment="comment",
+            properties={"k1": "v1", "k2": "v2"},
+        )
+
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident,
+            uri="uri2",
+            aliases=["alias3", "alias4"],
+            comment="comment",
+            properties={"k1": "v1", "k2": "v2"},
+        )
+
+        model_versions = self._catalog.as_model_catalog().list_model_versions(
+            model_ident
+        )
+        self.assertEqual(2, len(model_versions))
+        self.assertTrue(0 in model_versions)
+        self.assertTrue(1 in model_versions)
+
+        # Test delete model version
+        self.assertTrue(
+            self._catalog.as_model_catalog().delete_model_version(model_ident, 0)
+        )
+        model_versions = self._catalog.as_model_catalog().list_model_versions(
+            model_ident
+        )
+        self.assertEqual(1, len(model_versions))
+        self.assertTrue(1 in model_versions)
+
+        self.assertTrue(
+            self._catalog.as_model_catalog().delete_model_version(model_ident, 1)
+        )
+        model_versions = self._catalog.as_model_catalog().list_model_versions(
+            model_ident
+        )
+        self.assertEqual(0, len(model_versions))
+
+        # Test list model versions for non-existent model
+        with self.assertRaises(NoSuchModelException):
+            self._catalog.as_model_catalog().list_model_versions(
+                NameIdentifier.of(self._schema_name, "non_existent_model")
+            )
+
+    def test_link_delete_model_version(self):
+        model_name = "model_it_model" + str(randint(0, 1000))
+        model_ident = NameIdentifier.of(self._schema_name, model_name)
+        self._catalog.as_model_catalog().register_model(model_ident, "comment", {})
+
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident,
+            uri="uri",
+            aliases=["alias1"],
+            comment="comment",
+            properties={"k1": "v1", "k2": "v2"},
+        )
+
+        self.assertTrue(
+            self._catalog.as_model_catalog().delete_model_version(model_ident, 0)
+        )
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version(model_ident, 0)
+        )
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version_by_alias(
+                model_ident, "alias1"
+            )
+        )
+
+        self._catalog.as_model_catalog().link_model_version(
+            model_ident,
+            uri="uri",
+            aliases=["alias2"],
+            comment="comment",
+            properties={"k1": "v1", "k2": "v2"},
+        )
+
+        self.assertTrue(
+            self._catalog.as_model_catalog().delete_model_version_by_alias(
+                model_ident, "alias2"
+            )
+        )
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version_by_alias(
+                model_ident, "alias2"
+            )
+        )
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version(model_ident, 1)
+        )
+
+        # Test delete model version for non-existent model
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version(
+                NameIdentifier.of(self._schema_name, "non_existent_model"), 0
+            )
+        )
+
+        self.assertFalse(
+            self._catalog.as_model_catalog().delete_model_version_by_alias(
+                NameIdentifier.of(self._schema_name, "non_existent_model"), "alias1"
+            )
+        )
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/ModelVersionAliasSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/ModelVersionAliasSQLProviderFactory.java
index 726e3d0e2b7..c83e9deaa22 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/ModelVersionAliasSQLProviderFactory.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/ModelVersionAliasSQLProviderFactory.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
 import org.apache.gravitino.storage.relational.mapper.provider.base.ModelVersionAliasRelBaseSQLProvider;
+import org.apache.gravitino.storage.relational.mapper.provider.h2.ModelVersionAliasRelH2SQLProvider;
 import org.apache.gravitino.storage.relational.mapper.provider.postgresql.ModelVersionAliasRelPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.ModelVersionAliasRelPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
@@ -32,13 +33,11 @@ public class ModelVersionAliasSQLProviderFactory {
 
   static class ModelVersionAliasRelMySQLProvider extends ModelVersionAliasRelBaseSQLProvider {}
 
-  static class ModelVersionAliasRelH2Provider extends ModelVersionAliasRelBaseSQLProvider {}
-
   private static final Map<JDBCBackendType, ModelVersionAliasRelBaseSQLProvider>
       MODEL_VERSION_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new ModelVersionAliasRelMySQLProvider(),
-              JDBCBackendType.H2, new ModelVersionAliasRelH2Provider(),
+              JDBCBackendType.H2, new ModelVersionAliasRelH2SQLProvider(),
               JDBCBackendType.POSTGRESQL, new ModelVersionAliasRelPostgreSQLProvider());
 
   public static ModelVersionAliasRelBaseSQLProvider getProvider() {
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/ModelVersionAliasRelBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/ModelVersionAliasRelBaseSQLProvider.java
index 5354b888f33..abaaa5a8aed 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/ModelVersionAliasRelBaseSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/ModelVersionAliasRelBaseSQLProvider.java
@@ -100,13 +100,14 @@ public String softDeleteModelVersionAliasRelsByModelIdAndAlias(
       @Param("modelId") Long modelId, @Param("alias") String alias) {
     return "UPDATE "
         + ModelVersionAliasRelMapper.TABLE_NAME
-        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
-        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE model_id = #{modelId} AND model_version = ("
+        + " mvar JOIN ("
         + " SELECT model_version FROM "
         + ModelVersionAliasRelMapper.TABLE_NAME
         + " WHERE model_id = #{modelId} AND model_version_alias = #{alias} AND deleted_at = 0)"
-        + " AND deleted_at = 0";
+        + " subquery ON mvar.model_version = subquery.model_version"
+        + " SET mvar.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE mvar.model_id = #{modelId} AND mvar.deleted_at = 0";
   }
 
   public String softDeleteModelVersionAliasRelsBySchemaId(@Param("schemaId") Long schemaId) {
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/h2/ModelVersionAliasRelH2SQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/h2/ModelVersionAliasRelH2SQLProvider.java
new file mode 100644
index 00000000000..a9ddc01c1e2
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/h2/ModelVersionAliasRelH2SQLProvider.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.h2;
+
+import org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper;
+import org.apache.gravitino.storage.relational.mapper.provider.base.ModelVersionAliasRelBaseSQLProvider;
+import org.apache.ibatis.annotations.Param;
+
+public class ModelVersionAliasRelH2SQLProvider extends ModelVersionAliasRelBaseSQLProvider {
+
+  @Override
+  public String softDeleteModelVersionAliasRelsByModelIdAndAlias(
+      @Param("modelId") Long modelId, @Param("alias") String alias) {
+    return "UPDATE "
+        + ModelVersionAliasRelMapper.TABLE_NAME
+        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE model_id = #{modelId} AND model_version = ("
+        + " SELECT model_version FROM "
+        + ModelVersionAliasRelMapper.TABLE_NAME
+        + " WHERE model_id = #{modelId} AND model_version_alias = #{alias} AND deleted_at = 0)"
+        + " AND deleted_at = 0";
+  }
+}
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionAliasRelPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionAliasRelPostgreSQLProvider.java
index a37f0531258..da23bdca2d4 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionAliasRelPostgreSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionAliasRelPostgreSQLProvider.java
@@ -46,7 +46,7 @@ public String softDeleteModelVersionAliasRelsByModelIdAndVersion(
         + ModelVersionAliasRelMapper.TABLE_NAME
         + " SET deleted_at = floor(extract(epoch from((current_timestamp -"
         + " timestamp '1970-01-01 00:00:00')*1000)))"
-        + " WHERE model_id = #{modelId} AND model_version = #{version} AND deleted_at = 0";
+        + " WHERE model_id = #{modelId} AND model_version = #{modelVersion} AND deleted_at = 0";
   }
 
   @Override
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionMetaPostgreSQLProvider.java
index 09be14319bd..4183a53617c 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionMetaPostgreSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/ModelVersionMetaPostgreSQLProvider.java
@@ -47,7 +47,7 @@ public String softDeleteModelVersionMetaByModelIdAndVersion(
         + ModelVersionMetaMapper.TABLE_NAME
         + " SET deleted_at = floor(extract(epoch from((current_timestamp -"
         + " timestamp '1970-01-01 00:00:00')*1000)))"
-        + " WHERE model_id = #{modelId} AND version = #{version} AND deleted_at = 0";
+        + " WHERE model_id = #{modelId} AND version = #{modelVersion} AND deleted_at = 0";
   }
 
   @Override
diff --git a/docs/how-to-use-gvfs.md b/docs/how-to-use-gvfs.md
index 0dbfd867a3d..4f3515ea9c7 100644
--- a/docs/how-to-use-gvfs.md
+++ b/docs/how-to-use-gvfs.md
@@ -42,10 +42,7 @@ the path mapping and convert automatically.
 
 ### Prerequisites
 
-+ A Hadoop environment with HDFS running. GVFS has been tested against
-  Hadoop 3.1.0. It is recommended to use Hadoop 3.1.0 or later, but it should work with Hadoop 2.
-  x. Please create an [issue](https://www.github.com/apache/gravitino/issues) if you find any
-  compatibility issues.
++ A Hadoop environment with HDFS or other Hadoop Compatible File System (HCFS) implementations like S3, GCS, etc. GVFS has been tested against Hadoop 3.3.1. It is recommended to use Hadoop 3.3.1 or later, but it should work with Hadoop 2.x. Please create an [issue](https://www.github.com/apache/gravitino/issues) if you find any compatibility issues.
 
 ### Configuration
 
@@ -447,9 +444,7 @@ FileSystem fs = filesetPath.getFileSystem(conf);
 
 ### Prerequisites
 
-+ A Hadoop environment with HDFS running. Now we only supports Fileset on HDFS.
-  GVFS in Python has been tested against Hadoop 2.7.3. It is recommended to use Hadoop 2.7.3 or later,
-  it should work with Hadoop 3.x. Please create an [issue](https://www.github.com/apache/gravitino/issues)
++ A Hadoop environment with HDFS or other Hadoop Compatible File System (HCFS) implementations like S3, GCS, etc. GVFS has been tested against Hadoop 3.3.1. It is recommended to use Hadoop 3.3.1 or later, but it should work with Hadoop 2.x. Please create an [issue](https://www.github.com/apache/gravitino/issues)
   if you find any compatibility issues.
 + Python version >= 3.8. It has been tested GVFS works well with Python 3.8 and Python 3.9.
   Your Python version should be at least higher than Python 3.8.