diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestMiniClusters.java b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestMiniClusters.java
index 452c6cb0b6d7..82cb4f63e7da 100644
--- a/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestMiniClusters.java
+++ b/itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestMiniClusters.java
@@ -49,6 +49,7 @@
 import org.apache.hadoop.hive.llap.LlapItUtils;
 import org.apache.hadoop.hive.llap.daemon.MiniLlapCluster;
 import org.apache.hadoop.hive.llap.io.api.LlapProxy;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.exec.tez.TezSession;
 import org.apache.hadoop.hive.ql.lockmgr.zookeeper.CuratorFrameworkSingleton;
@@ -544,6 +545,7 @@ private void setFsRelatedProperties(HiveConf conf, boolean isLocalFs, FileSystem
     // Different paths if running locally vs a remote fileSystem. Ideally this difference should not
     // exist.
     Path warehousePath;
+    Path warehouseCatPath;
     Path jarPath;
     Path userInstallPath;
     if (isLocalFs) {
@@ -554,16 +556,19 @@ private void setFsRelatedProperties(HiveConf conf, boolean isLocalFs, FileSystem
       // Create a fake fs root for local fs
       Path localFsRoot = new Path(path, "localfs");
       warehousePath = new Path(localFsRoot, "warehouse");
+      warehouseCatPath = new Path(localFsRoot, "catalog");
       jarPath = new Path(localFsRoot, "jar");
       userInstallPath = new Path(localFsRoot, "user_install");
     } else {
       // TODO Why is this changed from the default in hive-conf?
       warehousePath = new Path(fsUriString, "/build/ql/test/data/warehouse/");
+      warehouseCatPath = new Path(fsUriString, "/build/ql/test/data/catalog/");
       jarPath = new Path(new Path(fsUriString, "/user"), "hive");
       userInstallPath = new Path(fsUriString, "/user");
     }
 
     warehousePath = fs.makeQualified(warehousePath);
+    warehouseCatPath = fs.makeQualified(warehouseCatPath);
     jarPath = fs.makeQualified(jarPath);
     userInstallPath = fs.makeQualified(userInstallPath);
 
@@ -571,6 +576,7 @@ private void setFsRelatedProperties(HiveConf conf, boolean isLocalFs, FileSystem
 
     // Remote dirs
     conf.setVar(ConfVars.METASTORE_WAREHOUSE, warehousePath.toString());
+    MetastoreConf.setVar(conf, MetastoreConf.ConfVars.WAREHOUSE_CATALOG, warehouseCatPath.toString());
     conf.setVar(ConfVars.HIVE_JAR_DIRECTORY, jarPath.toString());
     conf.setVar(ConfVars.HIVE_USER_INSTALL_DIR, userInstallPath.toString());
     // ConfVars.SCRATCH_DIR - {test.tmp.dir}/scratchdir
diff --git a/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g b/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g
index 0033aa3251c6..1c32d4c7d4ab 100644
--- a/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g
+++ b/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveParser.g
@@ -1132,10 +1132,10 @@ createCatalogStatement
     : KW_CREATE KW_CATALOG
         ifNotExists?
         name=identifier
-        catLocation
+        catLocation?
         catalogComment?
         (KW_PROPERTIES catprops=properties)?
-    -> ^(TOK_CREATECATALOG $name catLocation ifNotExists? catalogComment? $catprops?)
+    -> ^(TOK_CREATECATALOG $name catLocation? ifNotExists? catalogComment? $catprops?)
     ;
 
 catLocation
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogAnalyzer.java
index cc95fca3d5e5..71047787ba3f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogAnalyzer.java
@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.hive.ql.ddl.catalog.create;
 
+import com.google.common.base.Strings;
+import org.apache.hadoop.hive.metastore.CatalogUtil;
 import org.apache.hadoop.hive.metastore.api.Catalog;
 import org.apache.hadoop.hive.ql.QueryState;
 import org.apache.hadoop.hive.ql.ddl.DDLSemanticAnalyzerFactory;
@@ -29,6 +31,7 @@
 import org.apache.hadoop.hive.ql.parse.HiveParser;
 import org.apache.hadoop.hive.ql.parse.SemanticException;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -43,16 +46,19 @@ public CreateCatalogAnalyzer(QueryState queryState) throws SemanticException {
   @Override
   public void analyzeInternal(ASTNode root) throws SemanticException {
     String catalogName = unescapeIdentifier(root.getChild(0).getText());
-    String locationUrl = unescapeSQLString(root.getChild(1).getChild(0).getText());
-    outputs.add(toWriteEntity(locationUrl));
 
+    String locationUrl = null;
     boolean ifNotExists = false;
     String comment = null;
-    Map<String, String> props = null;
+    Map<String, String> props = new HashMap<>();
 
-    for (int i = 2; i < root.getChildCount(); i++) {
+    for (int i = 1; i < root.getChildCount(); i++) {
       ASTNode childNode = (ASTNode) root.getChild(i);
       switch (childNode.getToken().getType()) {
+        case HiveParser.TOK_CATALOGLOCATION:
+          locationUrl = unescapeSQLString(childNode.getChild(0).getText());
+          outputs.add(toWriteEntity(locationUrl));
+          break;
         case HiveParser.TOK_IFNOTEXISTS:
           ifNotExists = true;
           break;
@@ -67,10 +73,23 @@ public void analyzeInternal(ASTNode root) throws SemanticException {
       }
     }
 
+    // Set default type to hive if not specified
+    checkCatalogType(props);
+
     CreateCatalogDesc desc = new CreateCatalogDesc(catalogName, comment, locationUrl, ifNotExists, props);
     Catalog catalog = new Catalog(catalogName, locationUrl);
 
     rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(), desc)));
     outputs.add(new WriteEntity(catalog, WriteEntity.WriteType.DDL_NO_LOCK));
   }
+
+  private static void checkCatalogType(Map<String, String> props) throws SemanticException {
+    String catalogType = props.get(CatalogUtil.TYPE);
+    // Normalize and validate catalog type (fail fast on invalid input)
+    try {
+      props.put(CatalogUtil.TYPE, CatalogUtil.normalizeCatalogType(catalogType));
+    } catch (IllegalArgumentException e) {
+      throw new SemanticException(String.format("type '%s' is not valid", catalogType));
+    }
+  }
 }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogOperation.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogOperation.java
index d252c02a7ff0..13cbbf57ce08 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogOperation.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/catalog/create/CreateCatalogOperation.java
@@ -20,6 +20,7 @@
 
 import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
 import org.apache.hadoop.hive.metastore.api.Catalog;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.ql.ErrorMsg;
 import org.apache.hadoop.hive.ql.ddl.DDLOperation;
 import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
@@ -37,7 +38,11 @@ public CreateCatalogOperation(DDLOperationContext context, CreateCatalogDesc des
 
   @Override
   public int execute() throws Exception {
-    Catalog catalog = new Catalog(desc.getName(), desc.getLocationUri());
+    String catLocationUri = Optional.ofNullable(desc.getLocationUri())
+            .orElse(MetastoreConf.getVar(context.getConf(),
+                    MetastoreConf.ConfVars.WAREHOUSE_CATALOG) + "/" + desc.getName());
+
+    Catalog catalog = new Catalog(desc.getName(), catLocationUri);
     catalog.setDescription(desc.getComment());
     Optional.ofNullable(desc.getCatlogProperties())
             .ifPresent(catalog::setParameters);
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseAnalyzer.java
index ed26d2ec7fee..47a708d7b715 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseAnalyzer.java
@@ -19,13 +19,13 @@
 package org.apache.hadoop.hive.ql.ddl.database.create;
 
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hive.metastore.api.DataConnector;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.DatabaseType;
 import org.apache.hadoop.hive.metastore.api.PrincipalType;
-import org.apache.hadoop.hive.ql.ErrorMsg;
 import org.apache.hadoop.hive.ql.QueryState;
 import org.apache.hadoop.hive.ql.ddl.DDLUtils;
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.ql.ddl.DDLWork;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
 import org.apache.hadoop.hive.ql.parse.ASTNode;
 import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
 import org.apache.hadoop.hive.ql.parse.HiveParser;
@@ -51,10 +52,9 @@ public CreateDatabaseAnalyzer(QueryState queryState) throws SemanticException {
   @Override
   public void analyzeInternal(ASTNode root) throws SemanticException {
     Pair<String, String> catDbNamePair = DDLUtils.getCatDbNamePair((ASTNode) root.getChild(0));
-    String catalogName = catDbNamePair.getLeft();
-    if (catalogName != null && getCatalog(catalogName) == null) {
-      throw new SemanticException(ErrorMsg.CATALOG_NOT_EXISTS, catalogName);
-    }
+    String catalogName = Optional.ofNullable(catDbNamePair.getLeft())
+            .orElse(HiveUtils.getCurrentCatalogOrDefault(conf));
+
     String databaseName = catDbNamePair.getRight();
 
     boolean ifNotExists = false;
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseOperation.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseOperation.java
index e17a9419dd5d..270d7ede2057 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseOperation.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/database/create/CreateDatabaseOperation.java
@@ -31,6 +31,7 @@
 import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
 import org.apache.hadoop.hive.ql.session.SessionState;
 
 /**
@@ -56,12 +57,12 @@ public int execute() throws HiveException {
         if (desc.getManagedLocationUri() != null) {
           database.setManagedLocationUri(desc.getManagedLocationUri());
         }
-        makeLocationQualified(database); // TODO catalog. Add catalog prefix for db location. Depend on HIVE-29241.
+        makeLocationQualified(database);
         if (database.getLocationUri().equalsIgnoreCase(database.getManagedLocationUri())) {
           throw new HiveException("Managed and external locations for database cannot be the same");
         }
       } else if (desc.getDatabaseType() == DatabaseType.REMOTE) {
-        makeLocationQualified(database); // TODO catalog. Add catalog prefix for db location. Depend on HIVE-29241.
+        makeLocationQualified(database);
         database.setConnector_name(desc.getConnectorName());
         database.setRemote_dbname(desc.getRemoteDbName());
       } else { // should never be here
@@ -81,34 +82,62 @@ public int execute() throws HiveException {
   }
 
   private void makeLocationQualified(Database database) throws HiveException {
+    String catalogName = database.getCatalogName().toLowerCase();
+    String dbName = database.getName().toLowerCase();
+    boolean isDefaultCatalog = HiveUtils.isDefaultCatalog(catalogName, context.getConf());
+
+    // -------- External location --------
     if (database.isSetLocationUri()) {
       database.setLocationUri(Utilities.getQualifiedPath(context.getConf(), new Path(database.getLocationUri())));
     } else {
-      // Location is not set we utilize WAREHOUSE_EXTERNAL together with database name
-      String rootDir = MetastoreConf.getVar(context.getConf(), MetastoreConf.ConfVars.WAREHOUSE_EXTERNAL);
-      if (rootDir == null || rootDir.trim().isEmpty()) {
-        // Fallback plan
-        LOG.warn(String.format(
-            "%s is not set, falling back to %s. This could cause external tables to use to managed tablespace.",
-            MetastoreConf.ConfVars.WAREHOUSE_EXTERNAL.getVarname(), MetastoreConf.ConfVars.WAREHOUSE.getVarname()));
-        rootDir = MetastoreConf.getVar(context.getConf(), MetastoreConf.ConfVars.WAREHOUSE);
-      }
-      Path path = new Path(rootDir, database.getName().toLowerCase() + DATABASE_PATH_SUFFIX);
-      String qualifiedPath = Utilities.getQualifiedPath(context.getConf(), path);
-      database.setLocationUri(qualifiedPath);
+      String rootDir = getExternalRootDir(isDefaultCatalog);
+      Path path = buildDbPath(rootDir, catalogName, dbName, isDefaultCatalog);
+      database.setLocationUri(Utilities.getQualifiedPath(context.getConf(), path));
     }
 
+    // -------- Managed location --------
     if (database.isSetManagedLocationUri()) {
       database.setManagedLocationUri(Utilities.getQualifiedPath(context.getConf(),
           new Path(database.getManagedLocationUri())));
     } else {
-      // ManagedLocation is not set we utilize WAREHOUSE together with database name 
-      String rootDir = MetastoreConf.getVar(context.getConf(), MetastoreConf.ConfVars.WAREHOUSE);
-      Path path = new Path(rootDir, database.getName().toLowerCase() + DATABASE_PATH_SUFFIX);
+      String rootDir = MetastoreConf.getVar(
+              context.getConf(),
+              isDefaultCatalog
+                      ? MetastoreConf.ConfVars.WAREHOUSE
+                      : MetastoreConf.ConfVars.WAREHOUSE_CATALOG
+      );
+
+      Path path = buildDbPath(rootDir, catalogName, dbName, isDefaultCatalog);
       String qualifiedPath = Utilities.getQualifiedPath(context.getConf(), path);
       if (!qualifiedPath.equals(database.getLocationUri())) {
         database.setManagedLocationUri(qualifiedPath);
       }
     }
   }
+
+  private Path buildDbPath(String rootDir, String catalogName, String dbName, boolean isDefaultCatalog) {
+    return isDefaultCatalog
+            ? new Path(rootDir, dbName + DATABASE_PATH_SUFFIX)
+            : new Path(rootDir + "/" + catalogName, dbName + DATABASE_PATH_SUFFIX);
+  }
+
+  private String getExternalRootDir(boolean isDefaultCatalog) {
+    MetastoreConf.ConfVars externalVar = isDefaultCatalog
+            ? MetastoreConf.ConfVars.WAREHOUSE_EXTERNAL
+            : MetastoreConf.ConfVars.WAREHOUSE_CATALOG_EXTERNAL;
+
+    String rootDir = MetastoreConf.getVar(context.getConf(), externalVar);
+    if (rootDir != null && !rootDir.trim().isEmpty()) {
+      return rootDir;
+    }
+
+    MetastoreConf.ConfVars fallbackVar = isDefaultCatalog
+            ? MetastoreConf.ConfVars.WAREHOUSE
+            : MetastoreConf.ConfVars.WAREHOUSE_CATALOG;
+
+    LOG.warn("{} is not set, falling back to {}. This could cause external tables to use managed tablespace.",
+            externalVar.getVarname(), fallbackVar.getVarname());
+
+    return MetastoreConf.getVar(context.getConf(), fallbackVar);
+  }
 }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveUtils.java
index a75349bf5444..2621bad88395 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveUtils.java
@@ -553,4 +553,8 @@ public static String getCurrentCatalogOrDefault(Configuration conf) {
             .map(SessionState::getCurrentCatalog)
             .orElseGet(() -> getDefaultCatalog(conf));
   }
+
+  public static boolean isDefaultCatalog(String catName, Configuration conf) {
+    return catName == null || catName.isEmpty() || getDefaultCatalog(conf).equals(catName);
+  }
 }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
index 573e1ec66db8..d3c916802017 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
@@ -181,7 +181,7 @@ private Warehouse getWh() throws MetaException {
   }
 
   private boolean isDefaultCatalog(String catName) {
-    return catName == null || catName.isEmpty() || getDefaultCatalog(conf).equals(catName);
+    return HiveUtils.isDefaultCatalog(catName, conf);
   }
 
   /**
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java
index 8aa2ca2ce3a8..fd38199d5678 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java
@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.hive.common.repl.ReplConst;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.Warehouse;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
@@ -39,6 +40,7 @@
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.metadata.Hive;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
 import org.apache.hadoop.hive.ql.metadata.Partition;
 import org.apache.hadoop.hive.ql.metadata.Table;
 import org.apache.hadoop.hive.ql.parse.repl.DumpType;
@@ -388,14 +390,34 @@ public static void createDbExportDump(FileSystem fs, Path metadataPath, Database
 
   private static void updateIfCustomDbLocations(Database database, Configuration conf) throws SemanticException {
     try {
-      String whLocatoion = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.WAREHOUSE_EXTERNAL,
-              MetastoreConf.getVar(conf, MetastoreConf.ConfVars.WAREHOUSE));
-      Path dbDerivedLoc = new Path(whLocatoion, database.getName().toLowerCase() + DATABASE_PATH_SUFFIX);
+      String catName = database.getCatalogName();
+      String dbName = database.getName().toLowerCase();
+      boolean isDefaultCatalog = HiveUtils.isDefaultCatalog(catName, conf);
+
+      // external warehouse root
+      String whLocation = MetastoreConf.getVar(conf,
+              isDefaultCatalog ? MetastoreConf.ConfVars.WAREHOUSE_EXTERNAL : MetastoreConf.ConfVars.WAREHOUSE_CATALOG_EXTERNAL,
+              MetastoreConf.getVar(conf,
+                      isDefaultCatalog ? MetastoreConf.ConfVars.WAREHOUSE : MetastoreConf.ConfVars.WAREHOUSE_CATALOG));
+
+      if (!isDefaultCatalog) {
+        whLocation = new Path(whLocation, catName).toString();
+      }
+
+      Path dbDerivedLoc = new Path(whLocation, dbName + DATABASE_PATH_SUFFIX);
       String defaultDbLoc = Utilities.getQualifiedPath((HiveConf) conf, dbDerivedLoc);
       database.putToParameters(ReplConst.REPL_IS_CUSTOM_DB_LOC,
               Boolean.toString(!defaultDbLoc.equals(database.getLocationUri())));
-      String whManagedLocatoion = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.WAREHOUSE);
-      Path dbDerivedManagedLoc = new Path(whManagedLocatoion, database.getName().toLowerCase()
+
+      // managed warehouse root
+      String whManagedLocatoion = MetastoreConf.getVar(conf,
+              isDefaultCatalog ? MetastoreConf.ConfVars.WAREHOUSE
+                      : MetastoreConf.ConfVars.WAREHOUSE_CATALOG);
+
+      if (!isDefaultCatalog) {
+        whManagedLocatoion = new Path(whManagedLocatoion, catName).toString();
+      }
+      Path dbDerivedManagedLoc = new Path(whManagedLocatoion, dbName
               + DATABASE_PATH_SUFFIX);
       String defaultDbManagedLoc = Utilities.getQualifiedPath((HiveConf) conf, dbDerivedManagedLoc);
       database.getParameters().put(ReplConst.REPL_IS_CUSTOM_DB_MANAGEDLOC, Boolean.toString(
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/queryhistory/repository/AbstractRepository.java b/ql/src/java/org/apache/hadoop/hive/ql/queryhistory/repository/AbstractRepository.java
index cd4a22ea3e91..7c889c32a90b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/queryhistory/repository/AbstractRepository.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/queryhistory/repository/AbstractRepository.java
@@ -28,6 +28,7 @@
 import org.apache.hadoop.hive.ql.QueryState;
 import org.apache.hadoop.hive.ql.metadata.Hive;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
 import org.apache.hadoop.hive.ql.metadata.Table;
 import org.apache.hadoop.hive.ql.parse.PartitionTransform;
 import org.apache.hadoop.hive.ql.parse.TransformSpec;
@@ -40,6 +41,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
+
 public abstract class AbstractRepository implements QueryHistoryRepository {
   protected Logger LOG = LoggerFactory.getLogger(getClass());
   @VisibleForTesting
@@ -82,9 +85,12 @@ protected Database initDatabase(Hive hive) {
       db = hive.getDatabase(QUERY_HISTORY_DB_NAME);
       if (db == null) {
         LOG.warn("Database ({}) for query history table hasn't been found, auto-creating one", QUERY_HISTORY_DB_NAME);
-        String location = getDatabaseLocation(QUERY_HISTORY_DB_NAME);
+        // TODO catalog. The Hive Query History functionality is currently limited to the default catalog. Depend on HIVE-29561
         db = new Database(QUERY_HISTORY_DB_NAME, QUERY_HISTORY_DB_COMMENT,
-            location, null);
+            null, null);
+        db.setCatalogName(getDefaultCatalog(conf));
+        String location = getDatabaseLocation(db);
+        db.setLocationUri(location);
         hive.createDatabase(db, false);
       }
       return db;
@@ -93,8 +99,8 @@ protected Database initDatabase(Hive hive) {
     }
   }
 
-  private String getDatabaseLocation(String databaseName) throws Exception {
-    return warehouse.getDefaultExternalDatabasePath(databaseName).toUri().toString();
+  private String getDatabaseLocation(Database db) throws Exception {
+    return warehouse.getDefaultExternalDatabasePath(db).toUri().toString();
   }
 
   protected Table initTable(Hive hive, Database db) {
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java b/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java
index 254d60fb8844..efbde31be9cc 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java
@@ -132,6 +132,8 @@ public void authorizeDbLevelOperations(Privilege[] readRequiredPriv, Privilege[]
     Path root = null;
     try {
       initWh();
+      // TODO catalog. Need to determine auth root path based on catalog name.
+      //  If the catalog name is available, use `wh.getWhRoot(catName)` to obtain the auth path. Depend on HIVE-29562
       root = wh.getWhRoot();
       // When we have some path in outputs, we should check access on that path, usually happens when
       // we have HiveOperation.CREATEDATABASE query with some location
@@ -455,7 +457,7 @@ protected Path getDbLocation(Database db) throws HiveException {
       initWh();
       String location = db.getLocationUri();
       if (location == null) {
-        return wh.getDefaultDatabasePath(db.getName());
+        return wh.getDefaultDatabasePath(db);
       } else {
         return wh.getDnsPath(wh.getDatabasePath(db));
       }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/util/HiveStrictManagedMigration.java b/ql/src/java/org/apache/hadoop/hive/ql/util/HiveStrictManagedMigration.java
index ede6fc791344..714195236926 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/util/HiveStrictManagedMigration.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/util/HiveStrictManagedMigration.java
@@ -747,7 +747,7 @@ void processDatabase(String dbName, ForkJoinPool tablePool) {
       boolean modifyLocation = shouldModifyDatabaseLocation(dbObj);
 
       if (modifyLocation) {
-        Path newDefaultDbLocation = getDefaultDbPathManagedOrExternal(dbName);
+        Path newDefaultDbLocation = getDefaultDbPathManagedOrExternal(dbObj);
 
         LOG.info("Changing location of database {} to {}", dbName, newDefaultDbLocation);
         if (!runOptions.dryRun) {
@@ -794,7 +794,7 @@ void processDatabase(String dbName, ForkJoinPool tablePool) {
             "The migration must be run again for this database.", dbObj.getName());
         } else {
           if (!runOptions.dryRun) {
-            Path newDefaultDbLocation = getDefaultDbPathManagedOrExternal(dbName);
+            Path newDefaultDbLocation = getDefaultDbPathManagedOrExternal(dbObj);
             // dbObj after this call would have the new DB location.
             // Keep that in mind if anything below this requires the old DB path.
             hiveUpdater.get().updateDbLocation(dbObj, newDefaultDbLocation);
@@ -810,10 +810,10 @@ void processDatabase(String dbName, ForkJoinPool tablePool) {
     }
   }
 
-  private Path getDefaultDbPathManagedOrExternal(String dbName) throws MetaException {
+  private Path getDefaultDbPathManagedOrExternal(Database db) throws MetaException {
     return runOptions.shouldMoveExternal ?
-      wh.get().getDefaultExternalDatabasePath(dbName) :
-      wh.get().getDefaultDatabasePath(dbName);
+            wh.get().getDefaultExternalDatabasePath(db) :
+            wh.get().getDefaultDatabasePath(db);
   }
 
   public static boolean migrateTable(Table tableObj, TableType tableType, TableMigrationOption migrationOption,
@@ -878,7 +878,7 @@ boolean processTable(Database dbObj, String tableName, boolean modifyLocation) {
 
       if (shouldMoveTable && shouldModifyTableLocation(dbObj, tableObj)) {
         Path newTablePath = wh.get().getDnsPath(
-          new Path(getDefaultDbPathManagedOrExternal(dbName),
+          new Path(getDefaultDbPathManagedOrExternal(dbObj),
             MetaStoreUtils.encodeTableName(tableName.toLowerCase())));
         moveTableData(dbObj, tableObj, newTablePath);
         if (!runOptions.dryRun) {
@@ -910,7 +910,7 @@ boolean shouldModifyDatabaseLocation(Database dbObj) throws IOException, MetaExc
       // Check if the database location is in the default location based on the old warehouse root.
       // If so then change the database location to the default based on the current warehouse root.
       String dbLocation = dbObj.getLocationUri();
-      Path oldDefaultDbLocation = oldWh.get().getDefaultDatabasePath(dbName);
+      Path oldDefaultDbLocation = oldWh.get().getDefaultDatabasePath(dbObj);
       if (arePathsEqual(conf, dbLocation, oldDefaultDbLocation.toString())) {
         if (hasEquivalentEncryption(encryptionShim, oldDefaultDbLocation, targetPath)) {
           if (hasEquivalentErasureCodingPolicy(ecShim, oldDefaultDbLocation, targetPath)) {
@@ -965,7 +965,7 @@ boolean shouldModifyPartitionLocation(Database dbObj, Table tableObj, Partition
   }
 
   void createExternalDbDir(Database dbObj) throws IOException, MetaException {
-    Path externalTableDbPath = wh.get().getDefaultExternalDatabasePath(dbObj.getName());
+    Path externalTableDbPath = wh.get().getDefaultExternalDatabasePath(dbObj);
     FileSystem fs = getFS(externalTableDbPath, conf, fsOperationUser);
     if (!fs.exists(externalTableDbPath)) {
       String dbOwner = ownerName;
diff --git a/ql/src/test/queries/clientpositive/catalog.q b/ql/src/test/queries/clientpositive/catalog.q
index cfeb1e56780e..269588307b5f 100644
--- a/ql/src/test/queries/clientpositive/catalog.q
+++ b/ql/src/test/queries/clientpositive/catalog.q
@@ -7,8 +7,8 @@ dfs -mkdir -p hdfs:///tmp/test_cat;
 -- SORT_QUERY_RESULTS
 SHOW CATALOGS;
 
--- CREATE with comment
-CREATE CATALOG test_cat LOCATION 'hdfs:///tmp/test_cat' COMMENT 'Hive test catalog';
+-- CREATE with comment and default location
+CREATE CATALOG test_cat COMMENT 'Hive test catalog';
 
 -- DESCRIBE
 DESC CATALOG test_cat;
@@ -22,7 +22,7 @@ DROP CATALOG test_cat;
 SHOW CATALOGS;
 
 -- CREATE INE doesn't exist
-CREATE CATALOG IF NOT EXISTS test_cat LOCATION 'hdfs:///tmp/test_cat' COMMENT 'Hive test catalog' PROPERTIES('key1'='value1');;
+CREATE CATALOG IF NOT EXISTS test_cat LOCATION 'hdfs:///tmp/test_cat' COMMENT 'Hive test catalog' PROPERTIES('type'='hive');
 SHOW CATALOGS;
 
 -- DROP IE exists
@@ -33,7 +33,7 @@ SHOW CATALOGS;
 DROP CATALOG IF EXISTS test_cat;
 
 -- SHOW
-CREATE CATALOG test_cat LOCATION 'hdfs:///tmp/test_cat' COMMENT 'Hive test catalog';
+CREATE CATALOG test_cat LOCATION 'hdfs:///tmp/test_cat' COMMENT 'Hive test catalog' PROPERTIES('type'='hive');
 SHOW CATALOGS;
 
 -- SHOW pattern
diff --git a/ql/src/test/queries/clientpositive/catalog_database.q b/ql/src/test/queries/clientpositive/catalog_database.q
index e8b7b1332600..6598e0c12584 100644
--- a/ql/src/test/queries/clientpositive/catalog_database.q
+++ b/ql/src/test/queries/clientpositive/catalog_database.q
@@ -7,7 +7,7 @@ CREATE DATABASE testdb;
 -- The list of databases in the catalog 'hive' should only contain the default and the testdb.
 SHOW DATABASES;
 
--- CREATE a new catalog with comment
+-- CREATE a new native catalog with comment
 CREATE CATALOG testcat LOCATION '/tmp/testcat' COMMENT 'Hive test catalog';
 
 -- Check catalogs list
diff --git a/ql/src/test/results/clientpositive/llap/catalog.q.out b/ql/src/test/results/clientpositive/llap/catalog.q.out
index d86ab511706d..7e50701f9cbc 100644
--- a/ql/src/test/results/clientpositive/llap/catalog.q.out
+++ b/ql/src/test/results/clientpositive/llap/catalog.q.out
@@ -3,14 +3,12 @@ PREHOOK: type: SHOWCATALOGS
 POSTHOOK: query: SHOW CATALOGS
 POSTHOOK: type: SHOWCATALOGS
 hive
-#### A masked pattern was here ####
+PREHOOK: query: CREATE CATALOG test_cat COMMENT 'Hive test catalog'
 PREHOOK: type: CREATECATALOG
 PREHOOK: Output: catalog:test_cat
-PREHOOK: Output: hdfs://### HDFS PATH ###
-#### A masked pattern was here ####
+POSTHOOK: query: CREATE CATALOG test_cat COMMENT 'Hive test catalog'
 POSTHOOK: type: CREATECATALOG
 POSTHOOK: Output: catalog:test_cat
-POSTHOOK: Output: hdfs://### HDFS PATH ###
 PREHOOK: query: DESC CATALOG test_cat
 PREHOOK: type: DESCCATALOG
 PREHOOK: Input: catalog:test_cat
diff --git a/ql/src/test/results/clientpositive/llap/database_location.q.out b/ql/src/test/results/clientpositive/llap/database_location.q.out
index 49821dadb1b2..d769dfc63f06 100644
--- a/ql/src/test/results/clientpositive/llap/database_location.q.out
+++ b/ql/src/test/results/clientpositive/llap/database_location.q.out
@@ -160,6 +160,7 @@ STAGE DEPENDENCIES:
 STAGE PLANS:
   Stage: Stage-0
     Create Database
+      catalogName: hive
       database type:
 #### A masked pattern was here ####
       name: db3
diff --git a/ql/src/test/results/clientpositive/llap/db_ddl_explain.q.out b/ql/src/test/results/clientpositive/llap/db_ddl_explain.q.out
index 257d69751fde..c04bef3acbfe 100644
--- a/ql/src/test/results/clientpositive/llap/db_ddl_explain.q.out
+++ b/ql/src/test/results/clientpositive/llap/db_ddl_explain.q.out
@@ -10,6 +10,7 @@ STAGE DEPENDENCIES:
 STAGE PLANS:
   Stage: Stage-0
     Create Database
+      catalogName: hive
       database type:
       name: d
 
diff --git a/ql/src/test/results/clientpositive/tez/explainanalyze_3.q.out b/ql/src/test/results/clientpositive/tez/explainanalyze_3.q.out
index 980d6cc6c78a..b223677416be 100644
--- a/ql/src/test/results/clientpositive/tez/explainanalyze_3.q.out
+++ b/ql/src/test/results/clientpositive/tez/explainanalyze_3.q.out
@@ -93,7 +93,7 @@ POSTHOOK: type: CREATEDATABASE
 POSTHOOK: Output: database:newDB
 POSTHOOK: Output: hdfs://### HDFS PATH ###
 Stage-0
-  Create Database{"name:":"newDB"}
+  Create Database{"catalogName:":"hive","name:":"newDB"}
 
 #### A masked pattern was here ####
 PREHOOK: type: CREATEDATABASE
diff --git a/ql/src/test/results/clientpositive/tez/explainuser_3.q.out b/ql/src/test/results/clientpositive/tez/explainuser_3.q.out
index cc2adc2fc676..4923ee4a5e5d 100644
--- a/ql/src/test/results/clientpositive/tez/explainuser_3.q.out
+++ b/ql/src/test/results/clientpositive/tez/explainuser_3.q.out
@@ -117,7 +117,7 @@ POSTHOOK: type: CREATEDATABASE
 POSTHOOK: Output: database:newDB
 POSTHOOK: Output: hdfs://### HDFS PATH ###
 Stage-0
-  Create Database{"name:":"newDB"}
+  Create Database{"catalogName:":"hive","name:":"newDB"}
 
 #### A masked pattern was here ####
 PREHOOK: type: CREATEDATABASE
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/CatalogUtil.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/CatalogUtil.java
new file mode 100644
index 000000000000..75ebed67a241
--- /dev/null
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/CatalogUtil.java
@@ -0,0 +1,63 @@
+/*
+ * 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.hadoop.hive.metastore;
+
+import com.google.common.base.Strings;
+
+public class CatalogUtil {
+    public static final String TYPE = "type";
+
+    // Supported Catalog types
+    public enum CatalogType {
+        HIVE,
+        ICEBERG
+    }
+
+    /**
+     * Check if the given catalog type is valid.
+     * @param type catalog type (case-insensitive)
+     * @return true if valid, false otherwise
+     */
+    public static boolean isValidCatalogType(String type) {
+        try {
+            CatalogType.valueOf(type.toUpperCase().trim());
+            return true;
+        } catch (IllegalArgumentException e) {
+            return false;
+        }
+    }
+
+    /**
+     * Normalize catalog type to standard lowercase format.
+     * @param type catalog type (case-insensitive)
+     * @return normalized catalog type in lowercase
+     * @throws IllegalArgumentException if the catalog type is null, empty, or invalid
+     */
+    public static String normalizeCatalogType(String type) {
+        if (Strings.isNullOrEmpty(type)) {
+            // Set default type to hive if not specified
+            return CatalogType.HIVE.name();
+        }
+        if (!isValidCatalogType(type)) {
+            throw new IllegalArgumentException("Invalid catalog type: " + type);
+        }
+        return type.trim().toLowerCase();
+    }
+}
+
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/ReplChangeManager.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/ReplChangeManager.java
index a0680ca4234b..775ad9521f2c 100644
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/ReplChangeManager.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/ReplChangeManager.java
@@ -631,6 +631,8 @@ private static void setCmRootPermissions(Path cmroot) throws IOException{
     FileSystem cmFs = cmroot.getFileSystem(conf);
     cmFs.setPermission(cmroot, new FsPermission("770"));
     try {
+      // TODO catalog. The Repl function is currently only available for the default catalog path ConfVars.WAREHOUSE.
+      //  We may consider adding new created native catalog path (ConfVars.WAREHOUSE_CATALOG) later. Depend on HIVE-29278.
       FileStatus warehouseStatus = cmFs.getFileStatus(new Path(MetastoreConf.get(conf, ConfVars.WAREHOUSE.getVarname())));
       String warehouseOwner = warehouseStatus.getOwner();
       String warehouseGroup = warehouseStatus.getGroup();
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/Warehouse.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/Warehouse.java
index d32812e8c5b6..637031bf0932 100755
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/Warehouse.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/Warehouse.java
@@ -73,10 +73,14 @@ public class Warehouse {
   private static final String CAT_DB_TABLE_SEPARATOR = ".";
 
   private Path whRoot;
+  private Path whCatRoot;
   private Path whRootExternal;
+  private Path whCatRootExternal;
   private final Configuration conf;
   private final String whRootString;
+  private final String whCatRootString;
   private final String whRootExternalString;
+  private final String whCatRootExternalString;
   private final boolean isTenantBasedStorage;
 
   public static final Logger LOG = LoggerFactory.getLogger("hive.metastore.warehouse");
@@ -86,17 +90,26 @@ public class Warehouse {
 
   public Warehouse(Configuration conf) throws MetaException {
     this.conf = conf;
-    whRootString = MetastoreConf.getVar(conf, ConfVars.WAREHOUSE);
-    if (StringUtils.isBlank(whRootString)) {
-      throw new MetaException(ConfVars.WAREHOUSE.getVarname()
-          + " is not set in the config or blank");
-    }
+
+    whRootString = getRequiredVar(conf, ConfVars.WAREHOUSE);
+    whCatRootString = getRequiredVar(conf, ConfVars.WAREHOUSE_CATALOG);
+
     whRootExternalString = MetastoreConf.getVar(conf, ConfVars.WAREHOUSE_EXTERNAL);
+    whCatRootExternalString = MetastoreConf.getVar(conf, ConfVars.WAREHOUSE_CATALOG_EXTERNAL);
+
     cm = ReplChangeManager.getInstance(conf);
     storageAuthCheck = MetastoreConf.getBoolVar(conf, ConfVars.AUTHORIZATION_STORAGE_AUTH_CHECKS);
     isTenantBasedStorage = MetastoreConf.getBoolVar(conf, ConfVars.ALLOW_TENANT_BASED_STORAGE);
   }
 
+  private static String getRequiredVar(Configuration conf, ConfVars var) throws MetaException {
+    String value = MetastoreConf.getVar(conf, var);
+    if (StringUtils.isBlank(value)) {
+      throw new MetaException(var.getVarname() + " is not set in the config or blank");
+    }
+    return value;
+  }
+
   /**
    * Helper functions to convert IOException to MetaException
    */
@@ -156,27 +169,62 @@ public Path getDnsPath(Path path) throws MetaException {
    * This involves opening the FileSystem corresponding to the warehouse root
    * dir (but that should be ok given that this is only called during DDL
    * statements for non-external tables).
+   *
+   * @deprecated use {@link #getWhRoot(String)}
    */
   public Path getWhRoot() throws MetaException {
-    if (whRoot != null) {
-      return whRoot;
+    return getWhRoot(DEFAULT_CATALOG_NAME);
+  }
+
+  public Path getWhRoot(String catalogName) throws MetaException {
+    boolean isDefault = DEFAULT_CATALOG_NAME.equals(catalogName.trim());
+
+    Path rootDir = isDefault ? whRoot : whCatRoot;
+    if (rootDir == null) {
+      rootDir = getDnsPath(new Path(isDefault ? whRootString : whCatRootString));
+      if (isDefault) {
+        whRoot = rootDir;
+      } else {
+        whCatRoot = rootDir;
+      }
     }
-    whRoot = getDnsPath(new Path(whRootString));
-    return whRoot;
+
+    return isDefault ? rootDir : new Path(rootDir, catalogName);
   }
 
+
+
+  /**
+   * @deprecated use {@link #getWhRootExternal(String)}
+   */
   public Path getWhRootExternal() throws MetaException {
-    if (whRootExternal != null) {
-      return whRootExternal;
+    return getWhRootExternal(DEFAULT_CATALOG_NAME);
+  }
+
+  public Path getWhRootExternal(String catalogName) throws MetaException {
+    boolean isDefault = DEFAULT_CATALOG_NAME.equals(catalogName.trim());
+
+    Path rootExDir = isDefault ? whRootExternal : whCatRootExternal;
+    if (rootExDir != null) {
+      return rootExDir;
     }
-    if (!hasExternalWarehouseRoot()) {
-      whRootExternal = getWhRoot();
-    } else {
-      whRootExternal = getDnsPath(new Path(whRootExternalString));
+
+    if (isDefault) {
+      rootExDir = hasExternalWarehouseRoot()
+              ? getDnsPath(new Path(whRootExternalString))
+              : getWhRoot(catalogName);
+      whRootExternal = rootExDir;
+      return rootExDir;
     }
-    return whRootExternal;
+
+    rootExDir = hasExternalWarehouseRoot(catalogName)
+            ? getDnsPath(new Path(whCatRootExternalString, catalogName))
+            : getWhRoot(catalogName);
+    whCatRootExternal = rootExDir;
+    return rootExDir;
   }
 
+
   /**
    * Build the database path based on catalog name and database name.  This should only be used
    * when a database is being created or altered.  If you just want to find out the path a
@@ -191,16 +239,16 @@ public Path getWhRootExternal() throws MetaException {
    */
   public Path determineDatabasePath(Catalog cat, Database db) throws MetaException {
     if (db.getType() == DatabaseType.REMOTE) {
-      return getDefaultDatabasePath(db.getName(), true);
+      return getDefaultDatabasePath(db, true);
     }
     if (db.isSetLocationUri()) {
       return getDnsPath(new Path(db.getLocationUri()));
     }
-    if (cat == null || cat.getName().equalsIgnoreCase(DEFAULT_CATALOG_NAME)) {
+    if (cat.getName().equalsIgnoreCase(DEFAULT_CATALOG_NAME)) {
       if (db.getName().equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
-        return getWhRootExternal();
+        return getWhRootExternal(cat.getName());
       } else {
-        return new Path(getWhRootExternal(), dbDirFromDbName(db));
+        return new Path(getWhRootExternal(cat.getName()), dbDirFromDbName(db));
       }
     } else {
       return new Path(getDnsPath(new Path(cat.getLocationUri())), dbDirFromDbName(db));
@@ -225,7 +273,7 @@ public Path getDatabasePath(Database db) throws MetaException {
     }
     if (db.getCatalogName().equalsIgnoreCase(DEFAULT_CATALOG_NAME) &&
         db.getName().equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
-      return getWhRoot();
+      return getWhRoot(DEFAULT_CATALOG_NAME);
     }
     // this is for backward-compatibility where certain DBs do not have managedLocationUri set
     return new Path(db.getLocationUri());
@@ -241,9 +289,9 @@ public Path getDatabasePath(Database db) throws MetaException {
    */
   public Path getDatabaseExternalPath(Database db) throws MetaException {
       Path dbPath = new Path(db.getLocationUri());
-      if (FileUtils.isSubdirectory(getWhRoot().toString(), dbPath.toString() + Path.SEPARATOR)) {
+      if (FileUtils.isSubdirectory(getWhRoot(db.getCatalogName()).toString(), dbPath.toString() + Path.SEPARATOR)) {
         // db metadata incorrect, find new location based on external warehouse root
-        dbPath = getDefaultExternalDatabasePath(db.getName());
+        dbPath = getDefaultExternalDatabasePath(db);
       }
       return getDnsPath(dbPath);
   }
@@ -264,12 +312,15 @@ public Path getDatabaseManagedPath(Database db) throws MetaException {
       return new Path(db.getLocationUri());
     }
     if (db.getName().equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
-      return getWhRoot();
+      return getWhRoot(db.getCatalogName());
     }
 
-    return new Path(getWhRoot(), db.getName().toLowerCase() + DATABASE_WAREHOUSE_SUFFIX);
+    return new Path(getWhRoot(db.getCatalogName()), db.getName().toLowerCase() + DATABASE_WAREHOUSE_SUFFIX);
   }
 
+  /**
+   * @deprecated use {@link #getDefaultDatabasePath(Database)}
+   */
   public Path getDefaultDatabasePath(String dbName) throws MetaException {
     // TODO CAT - I am fairly certain that most calls to this are in error.  This should only be
     // used when the database location is unset, which should never happen except when a
@@ -279,27 +330,56 @@ public Path getDefaultDatabasePath(String dbName) throws MetaException {
     return getDefaultDatabasePath(dbName, false);
   }
 
+  public Path getDefaultDatabasePath(Database db) throws MetaException {
+    return getDefaultDatabasePath(db, false);
+  }
+
+  /**
+   * @deprecated use {@link #getDefaultExternalDatabasePath(Database)}
+   */
   public Path getDefaultExternalDatabasePath(String dbName) throws MetaException {
     return getDefaultDatabasePath(dbName, true);
   }
 
+  public Path getDefaultExternalDatabasePath(Database db) throws MetaException {
+    return getDefaultDatabasePath(db, true);
+  }
+
+  /**
+   * @deprecated use {@link #getDefaultDatabasePath(Database, boolean)}
+   */
   // should only be used to determine paths before the creation of databases
   public Path getDefaultDatabasePath(String dbName, boolean inExternalWH) throws MetaException {
-    if (inExternalWH) {
-      if (dbName.equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
-        return getWhRootExternal();
-      }
-      return new Path(getWhRootExternal(), dbName.toLowerCase() + DATABASE_WAREHOUSE_SUFFIX);
-    } else {
-      if (dbName.equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
-        return getWhRoot();
-      }
-      return new Path(getWhRoot(), dbName.toLowerCase() + DATABASE_WAREHOUSE_SUFFIX);
+    Database db = new Database();
+    db.setName(dbName);
+    db.setCatalogName(DEFAULT_CATALOG_NAME);
+    return getDefaultDatabasePath(db, inExternalWH);
+  }
+
+  public Path getDefaultDatabasePath(Database db, boolean inExternalWH) throws MetaException {
+    String catalogName = db.getCatalogName() != null ? db.getCatalogName() : DEFAULT_CATALOG_NAME;
+    String dbName = db.getName();
+
+    Path whRoot = inExternalWH ? getWhRootExternal(catalogName) : getWhRoot(catalogName);
+
+    if (dbName.equalsIgnoreCase(DEFAULT_DATABASE_NAME)) {
+      return whRoot;
     }
+
+    return new Path(whRoot, dbName.toLowerCase() + DATABASE_WAREHOUSE_SUFFIX);
   }
 
+  /**
+   * @deprecated use {@link #hasExternalWarehouseRoot(String)}
+   */
   private boolean hasExternalWarehouseRoot() {
-    return !StringUtils.isBlank(whRootExternalString);
+    return hasExternalWarehouseRoot(DEFAULT_CATALOG_NAME);
+  }
+
+  private boolean hasExternalWarehouseRoot(String catalogName) {
+    return catalogName.equalsIgnoreCase(DEFAULT_CATALOG_NAME)
+            ? !StringUtils.isBlank(whRootExternalString)
+            : !StringUtils.isBlank(whCatRootExternalString);
   }
 
   /**
@@ -319,10 +399,10 @@ public Path getDefaultTablePath(Database db, String tableName, boolean isExterna
     Path dbPath = null;
     if (isExternal) {
       dbPath = new Path(db.getLocationUri());
-      if (FileUtils.isSubdirectory(getWhRoot().toString(), dbPath.toString()) ||
-          getWhRoot().equals(dbPath)) {
+      if (FileUtils.isSubdirectory(getWhRoot(db.getCatalogName()).toString(), dbPath.toString()) ||
+          getWhRoot(db.getCatalogName()).equals(dbPath)) {
         // db metadata incorrect, find new location based on external warehouse root
-        dbPath = getDefaultExternalDatabasePath(db.getName());
+        dbPath = getDefaultExternalDatabasePath(db);
       }
     } else {
       dbPath = getDatabaseManagedPath(db);
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
index 10015f74837c..1d9d3024876b 100644
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
@@ -1776,6 +1776,15 @@ public enum ConfVars {
         "hive.metastore.warehouse.external.dir", "",
         "Default location for external tables created in the warehouse. " +
         "If not set or null, then the normal warehouse location will be used as the default location."),
+    WAREHOUSE_CATALOG("metastore.warehouse.catalog.dir",
+            "hive.metastore.warehouse.catalog.dir", "/user/hive/catalog/warehouse",
+            "Base location for databases in non-default catalogs. If the name of the catalog to which the db belongs is testcat, " +
+                    "then the default path prefix for the db is /user/hive/catalog/warehouse/testcat."),
+    WAREHOUSE_CATALOG_EXTERNAL("metastore.warehouse.catalog.external.dir",
+            "hive.metastore.warehouse.catalog.external.dir", "",
+            "Base location for external tables created in the warehouse of non-default catalogs. " +
+                    "If not set or null, then the normal warehouse location (MetastoreConf.WAREHOUSE_CATALOG) " +
+                    "will be used as the default location."),
     WM_DEFAULT_POOL_SIZE("metastore.wm.default.pool.size",
         "hive.metastore.wm.default.pool.size", 4,
         "The size of a default pool to create when creating an empty resource plan;\n" +
diff --git a/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java b/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java
index d21f239f3416..f41c2c087c58 100644
--- a/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java
+++ b/standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogFactory.java
@@ -79,6 +79,10 @@ private Catalog createCatalog() {
     if (configUri != null && !configUri.isEmpty()) {
       properties.put("uri", configUri);
     }
+    // TODO catalog. Currently, `configWarehouse` is determined by the path of the default catalog. We need to consider whether,
+    //  in the case of a non-default catalog, `configWarehouse` should be determined by the specific catalog path instead.
+    //  Need to consider adding new created native/internal catalog warehouse(MetastoreConf.ConfVars.WAREHOUSE_CATALOG) later.
+    //  Create HIVE-28879 to track this issue.
     final String configWarehouse = MetastoreConf.getVar(configuration, MetastoreConf.ConfVars.WAREHOUSE);
     if (configWarehouse != null) {
       properties.put("warehouse", configWarehouse);
diff --git a/standalone-metastore/metastore-server/pom.xml b/standalone-metastore/metastore-server/pom.xml
index 64926eaba1c9..0ac0f61880dd 100644
--- a/standalone-metastore/metastore-server/pom.xml
+++ b/standalone-metastore/metastore-server/pom.xml
@@ -704,6 +704,8 @@
             <test.tmp.dir>${test.tmp.dir}</test.tmp.dir>
             <metastore.warehouse.dir>${test.warehouse.scheme}${test.warehouse.dir}</metastore.warehouse.dir>
             <hive.metastore.warehouse.external.dir>${test.warehouse.scheme}${test.warehouse.external.dir}</hive.metastore.warehouse.external.dir>
+            <metastore.warehouse.catalog.dir>${test.warehouse.scheme}${test.warehouse.dir}</metastore.warehouse.catalog.dir>
+            <hive.metastore.warehouse.catalog.external.dir>${test.warehouse.scheme}${test.warehouse.external.dir}</hive.metastore.warehouse.catalog.external.dir>
           </systemPropertyVariables>
           <additionalClasspathElements>
             <additionalClasspathElement>${log4j.conf.dir}</additionalClasspathElement>
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
index 842b9390af3e..a4c2b9870fd6 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandler.java
@@ -188,12 +188,12 @@ public static void createDefaultCatalog(RawStore ms, Warehouse wh) throws MetaEx
         // One time update issue.  When the new 'hive' catalog is created in an upgrade the
         // script does not know the location of the warehouse.  So we need to update it.
         LOG.info("Setting location of default catalog, as it hasn't been done after upgrade");
-        defaultCat.setLocationUri(wh.getWhRoot().toString());
+        defaultCat.setLocationUri(wh.getWhRoot(DEFAULT_CATALOG_NAME).toString());
         ms.alterCatalog(defaultCat.getName(), defaultCat);
       }
 
     } catch (NoSuchObjectException e) {
-      Catalog cat = new Catalog(DEFAULT_CATALOG_NAME, wh.getWhRoot().toString());
+      Catalog cat = new Catalog(DEFAULT_CATALOG_NAME, wh.getWhRoot(DEFAULT_CATALOG_NAME).toString());
       long time = System.currentTimeMillis() / 1000;
       cat.setCreateTime((int) time);
       cat.setDescription(Warehouse.DEFAULT_CATALOG_COMMENT);
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java
index 235d1cfef550..3693256e978c 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java
@@ -825,7 +825,7 @@ public Database transformDatabase(Database db, List<String> processorCapabilitie
           LOG.debug("Processor has atleast one of ACID write capabilities, setting current locationUri " + db.getLocationUri() + " as managedLocationUri");
           db.setManagedLocationUri(new Path(db.getLocationUri()).toString());
         }
-        Path extWhLocation = hmsHandler.getWh().getDefaultExternalDatabasePath(db.getName());
+        Path extWhLocation = hmsHandler.getWh().getDefaultExternalDatabasePath(db);
         LOG.info("Database's location is a managed location, setting to a new default path based on external warehouse path:" + extWhLocation.toString());
         db.setLocationUri(extWhLocation.toString());
       }
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
index 447d1a6a2c21..aee3337a8882 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
@@ -694,6 +694,9 @@ private void rollbackTransactionToSavePoint(String savePoint) {
   @Override
   public void createCatalog(Catalog cat) throws MetaException {
     LOG.debug("Creating catalog {}", cat);
+
+    ensureCatalogType(cat);
+
     boolean committed = false;
     MCatalog mCat = catToMCat(cat);
     try {
@@ -705,6 +708,25 @@ public void createCatalog(Catalog cat) throws MetaException {
     }
   }
 
+  /**
+   * Ensure catalog has a valid type, default to NATIVE if not specified.
+   * Also normalizes the type to lowercase (e.g., "NATIVE" -> "native").
+   */
+  private void ensureCatalogType(Catalog cat) throws MetaException {
+    Map<String, String> parameters = cat.getParameters();
+    if (parameters == null) {
+      parameters = new HashMap<>();
+      cat.setParameters(parameters);
+    }
+
+    String catalogType = parameters.get("type");
+    try {
+      parameters.put("type", CatalogUtil.normalizeCatalogType(catalogType));
+    } catch (IllegalArgumentException e) {
+      throw new MetaException("Invalid catalog type: " + catalogType);
+    }
+  }
+
   @Override
   public void alterCatalog(String catName, Catalog cat)
       throws MetaException, InvalidOperationException {
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/BaseHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/BaseHandler.java
index b91af16a35cf..f4fa6dd44a48 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/BaseHandler.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/BaseHandler.java
@@ -703,11 +703,11 @@ private void createDefaultDB_core(RawStore ms) throws MetaException, InvalidObje
       ms.getDatabase(DEFAULT_CATALOG_NAME, DEFAULT_DATABASE_NAME);
     } catch (NoSuchObjectException e) {
       LOG.info("Started creating a default database with name: {}", DEFAULT_DATABASE_NAME);
-      Database db = new Database(DEFAULT_DATABASE_NAME, DEFAULT_DATABASE_COMMENT,
-          wh.getDefaultDatabasePath(DEFAULT_DATABASE_NAME, true).toString(), null);
+      Database db = new Database(DEFAULT_DATABASE_NAME, DEFAULT_DATABASE_COMMENT, null, null);
+      db.setCatalogName(DEFAULT_CATALOG_NAME);
+      db.setLocationUri(wh.getDefaultDatabasePath(db, true).toString());
       db.setOwnerName(PUBLIC);
       db.setOwnerType(PrincipalType.ROLE);
-      db.setCatalogName(DEFAULT_CATALOG_NAME);
       long time = System.currentTimeMillis() / 1000;
       db.setCreateTime((int) time);
       db.setType(DatabaseType.NATIVE);
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/CreateDatabaseHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/CreateDatabaseHandler.java
index 523cd335bb27..215cadcb4f2f 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/CreateDatabaseHandler.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/CreateDatabaseHandler.java
@@ -225,8 +225,8 @@ protected void beforeExecute() throws TException, IOException {
 
     String passedInURI = db.getLocationUri();
     String passedInManagedURI = db.getManagedLocationUri();
-    Path defaultDbExtPath = wh.getDefaultDatabasePath(db.getName(), true);
-    Path defaultDbMgdPath = wh.getDefaultDatabasePath(db.getName(), false);
+    Path defaultDbExtPath = wh.getDefaultDatabasePath(db, true);
+    Path defaultDbMgdPath = wh.getDefaultDatabasePath(db, false);
     Path dbExtPath = (passedInURI != null) ?
         wh.getDnsPath(new Path(passedInURI)) : wh.determineDatabasePath(cat, db);
     Path dbMgdPath = (passedInManagedURI != null) ? wh.getDnsPath(new Path(passedInManagedURI)) : null;
diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/MetaStoreTestUtils.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/MetaStoreTestUtils.java
index 61fa385aaa3d..bf12a42513d3 100644
--- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/MetaStoreTestUtils.java
+++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/MetaStoreTestUtils.java
@@ -164,15 +164,18 @@ public static int startMetaStoreWithRetry(HadoopThriftAuthBridge bridge, Configu
                                             boolean createTransactionalTables) throws Exception {
     Exception metaStoreException = null;
     String warehouseDir = MetastoreConf.getVar(conf, ConfVars.WAREHOUSE);
+    String warehouseCatDir = MetastoreConf.getVar(conf, ConfVars.WAREHOUSE_CATALOG);
 
     for (int tryCount = 0; tryCount < MetaStoreTestUtils.RETRY_COUNT; tryCount++) {
       try {
         int metaStorePort = findFreePort();
         if (!keepWarehousePath) {
           // Setting metastore instance specific warehouse directory, postfixing with port
-          Path postfixedWarehouseDir = new Path(warehouseDir, String.valueOf(metaStorePort));
-          MetastoreConf.setVar(conf, ConfVars.WAREHOUSE, postfixedWarehouseDir.toString());
-          warehouseDir = postfixedWarehouseDir.toString();
+          warehouseDir = new Path(warehouseDir, String.valueOf(metaStorePort)).toString();
+          MetastoreConf.setVar(conf, ConfVars.WAREHOUSE, warehouseDir);
+
+          warehouseCatDir = new Path(warehouseCatDir, String.valueOf(metaStorePort)).toString();
+          MetastoreConf.setVar(conf, ConfVars.WAREHOUSE_CATALOG, warehouseCatDir);
         }
 
         String jdbcUrl = MetastoreConf.getVar(conf, ConfVars.CONNECT_URL_KEY);