diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
index a4fcbabf..318ff9b 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
@@ -270,8 +270,8 @@
             activeNotificationHandler.suspend(metadataProvider);
             try {
                 IMetadataLockManager lockManager = appCtx.getMetadataLockManager();
-                lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataverseName,
-                        datasetName);
+                lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), database,
+                        dataverseName, datasetName);
                 RebalanceUtil.rebalance(dataverseName, datasetName, targetNodes, metadataProvider, hcc,
                         NoOpDatasetRebalanceCallback.INSTANCE, force);
             } finally {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
index 3fd339e..9b87fc9 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveEntityEventsListener.java
@@ -718,7 +718,8 @@
         IMetadataLockManager lockManager = metadataProvider.getApplicationContext().getMetadataLockManager();
         DataverseName dataverseName = entityId.getDataverseName();
         String entityName = entityId.getEntityName();
-        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), dataverseName, entityName);
+        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), entityId.getDatabaseName(), dataverseName,
+                entityName);
         acquireSuspendDatasetsLocks(metadataProvider, lockManager, targetDataset);
     }
 
@@ -729,8 +730,8 @@
                 // DDL operation already acquired the proper lock for the operation
                 continue;
             }
-            lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataset.getDataverseName(),
-                    dataset.getDatasetName());
+            lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataset.getDatabaseName(),
+                    dataset.getDataverseName(), dataset.getDatasetName());
         }
     }
 
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
index 34a54d1..f63cc9f 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java
@@ -191,12 +191,14 @@
     protected void acquireRecoveryLocks(IMetadataLockManager lockManager, IMetadataLockUtil lockUtil)
             throws AlgebricksException {
         EntityId entityId = listener.getEntityId();
-        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), entityId.getDataverseName(),
-                entityId.getEntityName());
+        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), entityId.getDatabaseName(),
+                entityId.getDataverseName(), entityId.getEntityName());
         for (Dataset dataset : listener.getDatasets()) {
-            lockManager.acquireDataverseReadLock(metadataProvider.getLocks(), dataset.getDataverseName());
-            lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataset.getDataverseName(),
-                    dataset.getDatasetName());
+            //TODO(DB): read lock on database
+            lockManager.acquireDataverseReadLock(metadataProvider.getLocks(), dataset.getDatabaseName(),
+                    dataset.getDataverseName());
+            lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataset.getDatabaseName(),
+                    dataset.getDataverseName(), dataset.getDatasetName());
         }
     }
 
@@ -207,11 +209,11 @@
     protected void acquirePostRecoveryLocks(IMetadataLockManager lockManager, IMetadataLockUtil lockUtil)
             throws AlgebricksException {
         EntityId entityId = listener.getEntityId();
-        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), entityId.getDataverseName(),
-                entityId.getEntityName());
+        lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), entityId.getDatabaseName(),
+                entityId.getDataverseName(), entityId.getEntityName());
         for (Dataset dataset : listener.getDatasets()) {
-            lockUtil.modifyDatasetBegin(lockManager, metadataProvider.getLocks(), dataset.getDataverseName(),
-                    dataset.getDatasetName());
+            lockUtil.modifyDatasetBegin(lockManager, metadataProvider.getLocks(), dataset.getDatabaseName(),
+                    dataset.getDataverseName(), dataset.getDatasetName());
         }
     }
 
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 17988cf..062be19 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -578,8 +578,10 @@
             throws Exception {
         DataverseDecl dvd = (DataverseDecl) stmt;
         DataverseName dvName = dvd.getDataverseName();
+        String database = MetadataUtil.resolveDatabase(null, dvName);
         metadataProvider.validateDataverseName(dvName, dvd.getSourceLocation());
-        lockManager.acquireDataverseReadLock(metadataProvider.getLocks(), dvName);
+        //TODO(DB): read lock on database
+        lockManager.acquireDataverseReadLock(metadataProvider.getLocks(), database, dvName);
         try {
             return doUseDataverseStatement(metadataProvider, dvd);
         } finally {
@@ -621,11 +623,12 @@
             IRequestParameters requestParameters) throws Exception {
         CreateDataverseStatement stmtCreateDataverse = (CreateDataverseStatement) stmt;
         DataverseName dvName = stmtCreateDataverse.getDataverseName();
+        String database = MetadataUtil.resolveDatabase(null, dvName);
         metadataProvider.validateDataverseName(dvName, stmtCreateDataverse.getSourceLocation());
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createDataverseBegin(lockManager, metadataProvider.getLocks(), dvName);
+        lockUtil.createDataverseBegin(lockManager, metadataProvider.getLocks(), database, dvName);
         try {
             doCreateDataverseStatement(metadataProvider, stmtCreateDataverse, requestParameters);
         } finally {
@@ -733,10 +736,13 @@
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName,
-                itemTypeDataverseName, itemTypeName, itemTypeAnonymous, metaItemTypeDataverseName, metaItemTypeName,
-                metaItemTypeAnonymous, nodegroupName, compactionPolicy, defaultCompactionPolicy, dd.getDatasetType(),
-                dd.getDatasetDetailsDecl());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
+        String itemTypeDatabase = MetadataUtil.resolveDatabase(null, itemTypeDataverseName);
+        String metaItemTypeDatabase = MetadataUtil.resolveDatabase(null, metaItemTypeDataverseName);
+        lockUtil.createDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName,
+                itemTypeDatabase, itemTypeDataverseName, itemTypeName, itemTypeAnonymous, metaItemTypeDatabase,
+                metaItemTypeDataverseName, metaItemTypeName, metaItemTypeAnonymous, nodegroupName, compactionPolicy,
+                defaultCompactionPolicy, dd.getDatasetType(), dd.getDatasetDetailsDecl());
         try {
             doCreateDatasetStatement(metadataProvider, dd, dataverseName, datasetName, itemTypeDataverseName,
                     itemTypeExpr, itemTypeName, metaItemTypeExpr, metaItemTypeDataverseName, metaItemTypeName, hcc,
@@ -1118,10 +1124,11 @@
         metadataProvider.validateDatabaseObjectName(stmtCreateIndex.getDataverseName(), indexName,
                 stmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtCreateIndex.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createIndexBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName,
+        lockUtil.createIndexBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName,
                 fullTextConfigName);
         try {
             doCreateIndex(metadataProvider, stmtCreateIndex, dataverseName, datasetName, hcc, requestParameters);
@@ -1473,11 +1480,12 @@
         metadataProvider.validateDatabaseObjectName(stmtCreateFilter.getDataverseName(), fullTextFilterName,
                 stmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtCreateFilter.getDataverseName());
-
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createFullTextFilterBegin(lockManager, metadataProvider.getLocks(), dataverseName, fullTextFilterName);
+        lockUtil.createFullTextFilterBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                fullTextFilterName);
         try {
             doCreateFullTextFilter(metadataProvider, stmtCreateFilter, dataverseName);
         } finally {
@@ -1540,13 +1548,14 @@
         metadataProvider.validateDatabaseObjectName(stmtCreateConfig.getDataverseName(), configName,
                 stmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtCreateConfig.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         ImmutableList<String> filterNames = stmtCreateConfig.getFilterNames();
 
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createFullTextConfigBegin(lockManager, metadataProvider.getLocks(), dataverseName, configName,
-                filterNames);
+        lockUtil.createFullTextConfigBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                configName, filterNames);
         try {
             doCreateFullTextConfig(metadataProvider, stmtCreateConfig, dataverseName, configName, filterNames);
         } finally {
@@ -1788,7 +1797,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.createTypeBegin(lockManager, metadataProvider.getLocks(), dataverseName, typeName);
+        lockUtil.createTypeBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, typeName);
         try {
             Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, database, dataverseName);
             if (dv == null) {
@@ -1834,6 +1843,7 @@
         DataverseDropStatement stmtDropDataverse = (DataverseDropStatement) stmt;
         SourceLocation sourceLoc = stmtDropDataverse.getSourceLocation();
         DataverseName dataverseName = stmtDropDataverse.getDataverseName();
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         metadataProvider.validateDataverseName(dataverseName, sourceLoc);
         if (dataverseName.equals(MetadataBuiltinEntities.DEFAULT_DATAVERSE_NAME)
                 || dataverseName.equals(MetadataConstants.METADATA_DATAVERSE_NAME)) {
@@ -1843,7 +1853,7 @@
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropDataverseBegin(lockManager, metadataProvider.getLocks(), dataverseName);
+        lockUtil.dropDataverseBegin(lockManager, metadataProvider.getLocks(), database, dataverseName);
         try {
             doDropDataverse(stmtDropDataverse, metadataProvider, hcc, requestParameters);
         } finally {
@@ -2029,10 +2039,11 @@
         String datasetName = stmtDelete.getDatasetName().getValue();
         metadataProvider.validateDatabaseObjectName(stmtDelete.getDataverseName(), datasetName, sourceLoc);
         DataverseName dataverseName = getActiveDataverseName(stmtDelete.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.dropDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName);
         try {
             doDropDataset(dataverseName, datasetName, metadataProvider, stmtDelete.getIfExists(), hcc,
                     requestParameters, true, sourceLoc);
@@ -2138,11 +2149,12 @@
         metadataProvider.validateDatabaseObjectName(stmtIndexDrop.getDataverseName(),
                 stmtIndexDrop.getIndexName().getValue(), stmtIndexDrop.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtIndexDrop.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String datasetName = stmtIndexDrop.getDatasetName().getValue();
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropIndexBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.dropIndexBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName);
         try {
             doDropIndex(metadataProvider, stmtIndexDrop, dataverseName, datasetName, hcc, requestParameters);
         } finally {
@@ -2243,12 +2255,14 @@
             IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {
         FullTextFilterDropStatement stmtFilterDrop = (FullTextFilterDropStatement) stmt;
         DataverseName dataverseName = getActiveDataverseName(stmtFilterDrop.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String fullTextFilterName = stmtFilterDrop.getFilterName();
 
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropFullTextFilterBegin(lockManager, metadataProvider.getLocks(), dataverseName, fullTextFilterName);
+        lockUtil.dropFullTextFilterBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                fullTextFilterName);
         try {
             doDropFullTextFilter(metadataProvider, stmtFilterDrop, dataverseName, fullTextFilterName);
         } finally {
@@ -2287,12 +2301,13 @@
             throws AlgebricksException, RemoteException {
         FullTextConfigDropStatement stmtConfigDrop = (FullTextConfigDropStatement) stmt;
         DataverseName dataverseName = getActiveDataverseName(stmtConfigDrop.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String configName = stmtConfigDrop.getConfigName();
 
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropFullTextConfigBegin(lockManager, metadataProvider.getLocks(), dataverseName, configName);
+        lockUtil.dropFullTextConfigBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, configName);
         try {
             doDropFullTextConfig(metadataProvider, stmtConfigDrop, hcc, requestParameters);
         } finally {
@@ -2348,7 +2363,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.dropTypeBegin(lockManager, metadataProvider.getLocks(), dataverseName, typeName);
+        lockUtil.dropTypeBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, typeName);
         try {
             // Check if the dataverse exists
             Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, database, dataverseName);
@@ -2417,6 +2432,7 @@
         String viewName = cvs.getViewName();
         metadataProvider.validateDatabaseObjectName(cvs.getDataverseName(), viewName, stmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(cvs.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
 
         DataverseName viewItemTypeDataverseName;
         String viewItemTypeName;
@@ -2432,13 +2448,14 @@
             viewItemTypeName = MetadataBuiltinEntities.ANY_OBJECT_DATATYPE.getDatatypeName();
             viewItemTypeAnonymous = false;
         }
+        String itemTypeDatabase = MetadataUtil.resolveDatabase(null, viewItemTypeDataverseName);
 
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, viewName,
-                viewItemTypeDataverseName, viewItemTypeName, viewItemTypeAnonymous, null, null, false, null, null, true,
-                DatasetType.VIEW, null);
+        lockUtil.createDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, viewName,
+                itemTypeDatabase, viewItemTypeDataverseName, viewItemTypeName, viewItemTypeAnonymous, null, null, null,
+                false, null, null, true, DatasetType.VIEW, null);
         try {
             doCreateView(metadataProvider, cvs, dataverseName, viewName, viewItemTypeDataverseName, viewItemTypeName,
                     stmtRewriter, requestParameters);
@@ -2636,10 +2653,11 @@
         String viewName = stmtDrop.getViewName().getValue();
         metadataProvider.validateDatabaseObjectName(stmtDrop.getDataverseName(), viewName, sourceLoc);
         DataverseName dataverseName = getActiveDataverseName(stmtDrop.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, viewName);
+        lockUtil.dropDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, viewName);
         try {
             doDropView(metadataProvider, stmtDrop, dataverseName, viewName);
         } finally {
@@ -2712,6 +2730,7 @@
         metadataProvider.validateDatabaseObjectName(signature.getDataverseName(), signature.getName(),
                 stmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(signature.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         signature.setDataverseName(dataverseName);
         DataverseName libraryDataverseName = null;
         String libraryName = cfs.getLibraryName();
@@ -2721,12 +2740,12 @@
                 libraryDataverseName = dataverseName;
             }
         }
-
+        String libraryDatabase = MetadataUtil.resolveDatabase(null, libraryDataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createFunctionBegin(lockManager, metadataProvider.getLocks(), dataverseName, signature.getName(),
-                libraryDataverseName, libraryName);
+        lockUtil.createFunctionBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                signature.getName(), libraryDatabase, libraryDataverseName, libraryName);
         try {
             doCreateFunction(metadataProvider, cfs, signature, stmtRewriter, requestParameters);
         } finally {
@@ -2832,7 +2851,7 @@
                 if (libraryDataverseName == null) {
                     libraryDataverseName = dataverseName;
                 }
-                String libraryDatabase = MetadataUtil.resolveDatabase(null, dataverseName);
+                String libraryDatabase = MetadataUtil.resolveDatabase(null, libraryDataverseName);
                 String libraryName = cfs.getLibraryName();
                 Library library = MetadataManager.INSTANCE.getLibrary(mdTxnCtx, libraryDatabase, libraryDataverseName,
                         libraryName);
@@ -2983,11 +3002,13 @@
         metadataProvider.validateDatabaseObjectName(signature.getDataverseName(), signature.getName(),
                 stmtDropFunction.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(signature.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         signature.setDataverseName(dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropFunctionBegin(lockManager, metadataProvider.getLocks(), dataverseName, signature.getName());
+        lockUtil.dropFunctionBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                signature.getName());
         try {
             doDropFunction(metadataProvider, stmtDropFunction, signature, requestParameters);
         } finally {
@@ -3043,16 +3064,18 @@
         String adapterName = cas.getAdapterName();
         metadataProvider.validateDatabaseObjectName(cas.getDataverseName(), adapterName, cas.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(cas.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         DataverseName libraryDataverseName = cas.getLibraryDataverseName();
         if (libraryDataverseName == null) {
             libraryDataverseName = dataverseName;
         }
+        String libraryDatabase = MetadataUtil.resolveDatabase(null, libraryDataverseName);
         String libraryName = cas.getLibraryName();
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createAdapterBegin(lockManager, metadataProvider.getLocks(), dataverseName, adapterName,
-                libraryDataverseName, libraryName);
+        lockUtil.createAdapterBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, adapterName,
+                libraryDatabase, libraryDataverseName, libraryName);
         try {
             doCreateAdapter(metadataProvider, cas);
         } finally {
@@ -3126,10 +3149,11 @@
         String adapterName = stmtDropAdapter.getAdapterName();
         metadataProvider.validateDatabaseObjectName(stmtDropAdapter.getDataverseName(), adapterName, sourceLoc);
         DataverseName dataverseName = getActiveDataverseName(stmtDropAdapter.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropAdapterBegin(lockManager, metadataProvider.getLocks(), dataverseName, adapterName);
+        lockUtil.dropAdapterBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, adapterName);
         try {
             doDropAdapter(metadataProvider, stmtDropAdapter, dataverseName, adapterName);
         } finally {
@@ -3179,12 +3203,13 @@
         metadataProvider.validateDatabaseObjectName(cls.getDataverseName(), cls.getLibraryName(),
                 cls.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(cls.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String libraryName = cls.getLibraryName();
         String libraryHash = cls.getHash();
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createLibraryBegin(lockManager, metadataProvider.getLocks(), dataverseName, libraryName);
+        lockUtil.createLibraryBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, libraryName);
         try {
             doCreateLibrary(metadataProvider, dataverseName, libraryName, libraryHash, cls, hcc, requestParameters);
         } finally {
@@ -3313,10 +3338,11 @@
         metadataProvider.validateDatabaseObjectName(stmtDropLibrary.getDataverseName(), libraryName,
                 stmtDropLibrary.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtDropLibrary.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropLibraryBegin(lockManager, metadataProvider.getLocks(), dataverseName, libraryName);
+        lockUtil.dropLibraryBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, libraryName);
         try {
             doDropLibrary(metadataProvider, stmtDropLibrary, dataverseName, libraryName, hcc, requestParameters);
         } finally {
@@ -3407,6 +3433,7 @@
         metadataProvider.validateDatabaseObjectName(css.getDataverseName(), css.getSynonymName(),
                 css.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(css.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String synonymName = css.getSynonymName();
         DataverseName objectDataverseName =
                 css.getObjectDataverseName() != null ? css.getObjectDataverseName() : dataverseName;
@@ -3414,7 +3441,7 @@
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createSynonymBegin(lockManager, metadataProvider.getLocks(), dataverseName, synonymName);
+        lockUtil.createSynonymBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, synonymName);
         try {
             doCreateSynonym(metadataProvider, css, dataverseName, synonymName, objectDataverseName, objectName);
         } finally {
@@ -3462,10 +3489,11 @@
         metadataProvider.validateDatabaseObjectName(stmtSynDrop.getDataverseName(), synonymName,
                 stmtSynDrop.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtSynDrop.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.dropSynonymBegin(lockManager, metadataProvider.getLocks(), dataverseName, synonymName);
+        lockUtil.dropSynonymBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, synonymName);
         try {
             doDropSynonym(metadataProvider, stmtSynDrop, dataverseName, synonymName);
         } finally {
@@ -3503,10 +3531,11 @@
         metadataProvider.validateDatabaseObjectName(loadStmt.getDataverseName(), datasetName,
                 loadStmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(loadStmt.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         boolean bActiveTxn = true;
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.modifyDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.modifyDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName);
         try {
             Map<String, String> properties = loadStmt.getProperties();
             ExternalDataUtils.normalize(properties);
@@ -3549,7 +3578,8 @@
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         boolean bActiveTxn = true;
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                datasetName);
         JobId jobId = null;
         boolean atomic = false;
         try {
@@ -3636,11 +3666,13 @@
         metadataProvider.validateDatabaseObjectName(stmtInsertUpsert.getDataverseName(), datasetName,
                 stmtInsertUpsert.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(stmtInsertUpsert.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         final IMetadataLocker locker = new IMetadataLocker() {
             @Override
             public void lock() throws AlgebricksException {
                 compilationLock.readLock().lock();
-                lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+                lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                        datasetName);
             }
 
             @Override
@@ -3679,9 +3711,10 @@
                 if (jobSpec == null) {
                     return jobSpec;
                 }
-                String database = MetadataUtil.resolveDatabase(null, ((InsertStatement) stmt).getDataverseName());
-                Dataset ds = metadataProvider.findDataset(database, ((InsertStatement) stmt).getDataverseName(),
-                        ((InsertStatement) stmt).getDatasetName());
+                String insertStmtDatabase =
+                        MetadataUtil.resolveDatabase(null, ((InsertStatement) stmt).getDataverseName());
+                Dataset ds = metadataProvider.findDataset(insertStmtDatabase,
+                        ((InsertStatement) stmt).getDataverseName(), ((InsertStatement) stmt).getDatasetName());
                 atomic = ds.isAtomic();
                 if (atomic) {
                     int numParticipatingNodes = appCtx.getNodeJobTracker()
@@ -3729,7 +3762,8 @@
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         boolean bActiveTxn = true;
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.insertDeleteUpsertBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                datasetName);
         boolean atomic = false;
         JobId jobId = null;
         try {
@@ -3856,7 +3890,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.createFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, feedName);
+        lockUtil.createFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, feedName);
         try {
             Feed feed = MetadataManager.INSTANCE.getFeed(metadataProvider.getMetadataTxnContext(), database,
                     dataverseName, feedName);
@@ -3897,7 +3931,7 @@
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.createFeedPolicyBegin(lockManager, metadataProvider.getLocks(), dataverseName, policyName);
+        lockUtil.createFeedPolicyBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, policyName);
         try {
             mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
             metadataProvider.setMetadataTxnContext(mdTxnCtx);
@@ -3965,7 +3999,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.dropFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, feedName);
+        lockUtil.dropFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, feedName);
         try {
             Feed feed = MetadataManager.INSTANCE.getFeed(mdTxnCtx, database, dataverseName, feedName);
             if (feed == null) {
@@ -4022,7 +4056,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.dropFeedPolicyBegin(lockManager, metadataProvider.getLocks(), dataverseName, policyName);
+        lockUtil.dropFeedPolicyBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, policyName);
         try {
             FeedPolicyEntity feedPolicy =
                     MetadataManager.INSTANCE.getFeedPolicy(mdTxnCtx, database, dataverseName, policyName);
@@ -4056,7 +4090,7 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         boolean committed = false;
-        lockUtil.startFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, feedName);
+        lockUtil.startFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, feedName);
         try {
             metadataProvider.setMetadataTxnContext(mdTxnCtx);
             // Runtime handler
@@ -4071,8 +4105,8 @@
             }
             for (FeedConnection feedConnection : feedConnections) {
                 // what if the dataset is in a different dataverse
-                lockManager.acquireDatasetReadLock(metadataProvider.getLocks(), feedConnection.getDataverseName(),
-                        feedConnection.getDatasetName());
+                lockManager.acquireDatasetReadLock(metadataProvider.getLocks(), feedConnection.getDatabaseName(),
+                        feedConnection.getDataverseName(), feedConnection.getDatasetName());
             }
             ActiveNotificationHandler activeEventHandler =
                     (ActiveNotificationHandler) appCtx.getActiveNotificationHandler();
@@ -4120,7 +4154,7 @@
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.stopFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, feedName);
+        lockUtil.stopFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, feedName);
         try {
             listener.stop(metadataProvider);
         } finally {
@@ -4148,7 +4182,8 @@
         ActiveNotificationHandler activeEventHandler =
                 (ActiveNotificationHandler) appCtx.getActiveNotificationHandler();
         // Transaction handling
-        lockUtil.connectFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName, feedName);
+        lockUtil.connectFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName,
+                feedName);
         try {
             // validation
             Dataset dataset =
@@ -4203,7 +4238,8 @@
         }
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
-        lockUtil.disconnectFeedBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName, feedName);
+        lockUtil.disconnectFeedBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName,
+                feedName);
         try {
             ActiveNotificationHandler activeEventHandler =
                     (ActiveNotificationHandler) appCtx.getActiveNotificationHandler();
@@ -4247,11 +4283,12 @@
         metadataProvider.validateDatabaseObjectName(analyzeStatement.getDataverseName(),
                 analyzeStatement.getDatasetName(), analyzeStatement.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(analyzeStatement.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String datasetName = analyzeStatement.getDatasetName();
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.analyzeDatasetBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.analyzeDatasetBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName);
         try {
             doAnalyzeDataset(metadataProvider, analyzeStatement, dataverseName, datasetName, hcc, requestParameters);
         } finally {
@@ -4471,11 +4508,13 @@
         metadataProvider.validateDatabaseObjectName(analyzeDropStmt.getDataverseName(),
                 analyzeDropStmt.getDatasetName(), analyzeDropStmt.getSourceLocation());
         DataverseName dataverseName = getActiveDataverseName(analyzeDropStmt.getDataverseName());
+        String database = MetadataUtil.resolveDatabase(null, dataverseName);
         String datasetName = analyzeDropStmt.getDatasetName();
         if (isCompileOnly()) {
             return;
         }
-        lockUtil.analyzeDatasetDropBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.analyzeDatasetDropBegin(lockManager, metadataProvider.getLocks(), database, dataverseName,
+                datasetName);
         try {
             doAnalyzeDatasetDrop(metadataProvider, analyzeDropStmt, dataverseName, datasetName, hcc, requestParams);
         } finally {
@@ -4633,7 +4672,7 @@
         boolean bActiveTxn = true;
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
         List<JobSpecification> jobsToExecute = new ArrayList<>();
-        lockUtil.compactBegin(lockManager, metadataProvider.getLocks(), dataverseName, datasetName);
+        lockUtil.compactBegin(lockManager, metadataProvider.getLocks(), database, dataverseName, datasetName);
         try {
             Dataset ds = metadataProvider.findDataset(database, dataverseName, datasetName);
             if (ds == null) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
index db76734..acbc9e3 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
@@ -263,8 +263,8 @@
                 (ActiveNotificationHandler) appCtx.getActiveNotificationHandler();
         IMetadataLockManager lockManager = appCtx.getMetadataLockManager();
         LOGGER.debug("attempting to acquire dataset {} upgrade lock", source.getDatasetName());
-        lockManager.upgradeDatasetLockToWrite(metadataProvider.getLocks(), source.getDataverseName(),
-                source.getDatasetName());
+        lockManager.upgradeDatasetLockToWrite(metadataProvider.getLocks(), source.getDatabaseName(),
+                source.getDataverseName(), source.getDatasetName());
         LOGGER.debug("acquired dataset {} upgrade lock", source.getDatasetName());
         LOGGER.info("Updating dataset {} node group from {} to {}", source.getDatasetName(), source.getNodeGroupName(),
                 target.getNodeGroupName());
@@ -280,8 +280,8 @@
             MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
             LOGGER.info("dataset {} node group updated to {}", target.getDatasetName(), target.getNodeGroupName());
         } finally {
-            lockManager.downgradeDatasetLockToExclusiveModify(metadataProvider.getLocks(), target.getDataverseName(),
-                    target.getDatasetName());
+            lockManager.downgradeDatasetLockToExclusiveModify(metadataProvider.getLocks(), target.getDatabaseName(),
+                    target.getDataverseName(), target.getDatasetName());
         }
     }
 
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
index 301eebd..62d061e 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
@@ -187,8 +187,8 @@
      * @param targetNodes
      * @throws Exception
      */
-    public static void rebalanceDataset(AsterixHyracksIntegrationUtil integrationUtil, DataverseName dataverseName,
-            String datasetName, String[] targetNodes) throws Exception {
+    public static void rebalanceDataset(AsterixHyracksIntegrationUtil integrationUtil, String database,
+            DataverseName dataverseName, String datasetName, String[] targetNodes) throws Exception {
         ICcApplicationContext ccAppCtx =
                 (ICcApplicationContext) integrationUtil.getClusterControllerService().getApplicationContext();
         MetadataProvider metadataProvider = MetadataProvider.create(ccAppCtx, null);
@@ -198,8 +198,8 @@
             activeNotificationHandler.suspend(metadataProvider);
             try {
                 IMetadataLockManager lockManager = ccAppCtx.getMetadataLockManager();
-                lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), dataverseName,
-                        datasetName);
+                lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), database,
+                        dataverseName, datasetName);
                 RebalanceUtil.rebalance(dataverseName, datasetName, new LinkedHashSet<>(Arrays.asList(targetNodes)),
                         metadataProvider, ccAppCtx.getHcc(), NoOpDatasetRebalanceCallback.INSTANCE, false);
             } finally {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
index f87bfe2..fb0e629 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestEventsListener.java
@@ -114,7 +114,8 @@
         step(onStart);
         try {
             metadataProvider.getApplicationContext().getMetadataLockManager().acquireDatasetReadLock(
-                    metadataProvider.getLocks(), MetadataBuiltinEntities.DEFAULT_DATAVERSE_NAME, "type");
+                    metadataProvider.getLocks(), MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDatabaseName(),
+                    MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName(), "type");
         } catch (AlgebricksException e) {
             throw HyracksDataException.create(e);
         }
@@ -208,8 +209,11 @@
         try {
             IMetadataLockManager lockManager = metadataProvider.getApplicationContext().getMetadataLockManager();
             LockList locks = metadataProvider.getLocks();
-            lockManager.acquireDataverseReadLock(locks, entityId.getDataverseName());
-            lockManager.acquireActiveEntityWriteLock(locks, entityId.getDataverseName(), entityId.getEntityName());
+            String entityDatabase = entityId.getDatabaseName();
+            lockManager.acquireDatabaseReadLock(locks, entityDatabase);
+            lockManager.acquireDataverseReadLock(locks, entityDatabase, entityId.getDataverseName());
+            lockManager.acquireActiveEntityWriteLock(locks, entityDatabase, entityId.getDataverseName(),
+                    entityId.getEntityName());
             // persist entity
         } catch (Throwable th) {
             // This failure puts the system in a bad state.
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestUserActor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestUserActor.java
index 2d64f79..919e0e4 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestUserActor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestUserActor.java
@@ -52,13 +52,15 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 try {
-                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), dataverseName, entityName);
+                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), database, dataverseName,
+                            entityName);
                     Collection<Dataset> datasets = actionListener.getDatasets();
                     for (Dataset dataset : datasets) {
-                        lockUtil.modifyDatasetBegin(lockManager, mdProvider.getLocks(), dataset.getDataverseName(),
-                                dataset.getDatasetName());
+                        lockUtil.modifyDatasetBegin(lockManager, mdProvider.getLocks(), dataset.getDatabaseName(),
+                                dataset.getDataverseName(), dataset.getDatasetName());
                     }
                     actionListener.start(mdProvider);
                 } finally {
@@ -75,13 +77,15 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 try {
-                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), dataverseName, entityName);
+                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), database, dataverseName,
+                            entityName);
                     Collection<Dataset> datasets = actionListener.getDatasets();
                     for (Dataset dataset : datasets) {
-                        lockUtil.modifyDatasetBegin(lockManager, mdProvider.getLocks(), dataset.getDataverseName(),
-                                dataset.getDatasetName());
+                        lockUtil.modifyDatasetBegin(lockManager, mdProvider.getLocks(), dataset.getDatabaseName(),
+                                dataset.getDataverseName(), dataset.getDatasetName());
                     }
                     actionListener.stop(mdProvider);
                 } finally {
@@ -98,13 +102,15 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 Collection<Dataset> datasets = actionListener.getDatasets();
                 try {
-                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), dataverseName, entityName);
+                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), database, dataverseName,
+                            entityName);
                     for (Dataset dataset : datasets) {
                         lockManager.acquireDatasetExclusiveModificationLock(mdProvider.getLocks(),
-                                dataset.getDataverseName(), dataset.getDatasetName());
+                                dataset.getDatabaseName(), dataset.getDataverseName(), dataset.getDatasetName());
                     }
                     actionListener.suspend(mdProvider);
                 } catch (Exception e) {
@@ -123,15 +129,17 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 try {
-                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), dataverseName, entityName);
+                    lockManager.acquireActiveEntityWriteLock(mdProvider.getLocks(), database, dataverseName,
+                            entityName);
                     Collection<Dataset> datasets = actionListener.getDatasets();
                     for (Dataset dataset : datasets) {
-                        lockManager.upgradeDatasetLockToWrite(mdProvider.getLocks(), dataset.getDataverseName(),
-                                dataset.getDatasetName());
-                        lockManager.downgradeDatasetLockToExclusiveModify(mdProvider.getLocks(),
+                        lockManager.upgradeDatasetLockToWrite(mdProvider.getLocks(), dataset.getDatabaseName(),
                                 dataset.getDataverseName(), dataset.getDatasetName());
+                        lockManager.downgradeDatasetLockToExclusiveModify(mdProvider.getLocks(),
+                                dataset.getDatabaseName(), dataset.getDataverseName(), dataset.getDatasetName());
                     }
                     actionListener.resume(mdProvider);
                 } finally {
@@ -148,11 +156,13 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName entityDataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 try {
-                    lockManager.acquireActiveEntityReadLock(mdProvider.getLocks(), entityDataverseName, entityName);
-                    lockManager.acquireDatasetWriteLock(mdProvider.getLocks(), dataset.getDataverseName(),
-                            dataset.getDatasetName());
+                    lockManager.acquireActiveEntityReadLock(mdProvider.getLocks(), database, entityDataverseName,
+                            entityName);
+                    lockManager.acquireDatasetWriteLock(mdProvider.getLocks(), dataset.getDatabaseName(),
+                            dataset.getDataverseName(), dataset.getDatasetName());
                     List<Dataset> datasets = clusterController.getAllDatasets();
                     if (datasets.contains(dataset)) {
                         throw new HyracksDataException("Dataset " + dataset + " already exists");
@@ -173,11 +183,13 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName entityDataverseName = actionListener.getEntityId().getDataverseName();
+                String database = actionListener.getEntityId().getDatabaseName();
                 String entityName = actionListener.getEntityId().getEntityName();
                 try {
-                    lockManager.acquireActiveEntityReadLock(mdProvider.getLocks(), entityDataverseName, entityName); // we have to first read lock all active entities before deleting a dataset
-                    lockManager.acquireDatasetWriteLock(mdProvider.getLocks(), dataset.getDataverseName(),
-                            dataset.getDatasetName());
+                    lockManager.acquireActiveEntityReadLock(mdProvider.getLocks(), database, entityDataverseName,
+                            entityName); // we have to first read lock all active entities before deleting a dataset
+                    lockManager.acquireDatasetWriteLock(mdProvider.getLocks(), dataset.getDatabaseName(),
+                            dataset.getDataverseName(), dataset.getDatasetName());
                     List<Dataset> datasets = clusterController.getAllDatasets();
                     if (!datasets.contains(dataset)) {
                         throw new HyracksDataException("Dataset " + dataset + " does not exist");
@@ -198,9 +210,11 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = dataset.getDataverseName();
+                String database = dataset.getDatabaseName();
                 String datasetName = dataset.getDatasetName();
                 try {
-                    lockUtil.createIndexBegin(lockManager, mdProvider.getLocks(), dataverseName, datasetName, null);
+                    lockUtil.createIndexBegin(lockManager, mdProvider.getLocks(), database, dataverseName, datasetName,
+                            null);
                     if (actionListener.isActive()) {
                         throw new RuntimeDataException(ErrorCode.CANNOT_ADD_INDEX_TO_DATASET_CONNECTED_TO_ACTIVE_ENTITY,
                                 DatasetUtil.getFullyQualifiedDisplayName(dataverseName, datasetName) + ".index",
@@ -220,9 +234,10 @@
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
                 DataverseName dataverseName = dataset.getDataverseName();
+                String database = dataset.getDatabaseName();
                 String datasetName = dataset.getDatasetName();
                 try {
-                    lockUtil.dropIndexBegin(lockManager, mdProvider.getLocks(), dataverseName, datasetName);
+                    lockUtil.dropIndexBegin(lockManager, mdProvider.getLocks(), database, dataverseName, datasetName);
                     if (actionListener.isActive()) {
                         throw new RuntimeDataException(
                                 ErrorCode.CANNOT_REMOVE_INDEX_FROM_DATASET_CONNECTED_TO_ACTIVE_ENTITY,
@@ -242,11 +257,13 @@
         Action action = new Action() {
             @Override
             protected void doExecute(MetadataProvider mdProvider) throws Exception {
+                String database = dataset.getDatabaseName();
                 DataverseName dataverseName = dataset.getDataverseName();
                 String datasetName = dataset.getDatasetName();
                 try {
-                    lockManager.acquireDataverseReadLock(mdProvider.getLocks(), dataverseName);
-                    lockManager.acquireDatasetReadLock(mdProvider.getLocks(), dataverseName, datasetName);
+                    lockManager.acquireDatabaseReadLock(mdProvider.getLocks(), database);
+                    lockManager.acquireDataverseReadLock(mdProvider.getLocks(), database, dataverseName);
+                    lockManager.acquireDatasetReadLock(mdProvider.getLocks(), database, dataverseName, datasetName);
                     if (!semaphore.tryAcquire()) {
                         semaphore.acquire();
                     }
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/txn/RecoveryManagerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/txn/RecoveryManagerTest.java
index 4f8df3d..eb3a438 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/txn/RecoveryManagerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/txn/RecoveryManagerTest.java
@@ -90,8 +90,9 @@
         TestDataUtil.upsertData(datasetName, 10);
         final long countBeforeRebalance = TestDataUtil.getDatasetCount(datasetName);
         // rebalance dataset to single nc
-        TestDataUtil.rebalanceDataset(integrationUtil, MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName(),
-                datasetName, new String[] { "asterix_nc2" });
+        TestDataUtil.rebalanceDataset(integrationUtil, MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDatabaseName(),
+                MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName(), datasetName,
+                new String[] { "asterix_nc2" });
         // check data after rebalance
         final long countAfterRebalance = TestDataUtil.getDatasetCount(datasetName);
         Assert.assertEquals(countBeforeRebalance, countAfterRebalance);
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IMetadataLockManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IMetadataLockManager.java
index 79f13bf..ec34205 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IMetadataLockManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IMetadataLockManager.java
@@ -24,88 +24,106 @@
 
 public interface IMetadataLockManager {
 
+    void acquireDatabaseReadLock(LockList locks, String database) throws AlgebricksException;
+
+    void acquireDatabaseWriteLock(LockList locks, String database) throws AlgebricksException;
+
     /**
      * Acquire read lock on the dataverse
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDataverseReadLock(LockList locks, DataverseName dataverseName) throws AlgebricksException;
+    void acquireDataverseReadLock(LockList locks, String database, DataverseName dataverseName)
+            throws AlgebricksException;
 
     /**
      * Acquire write lock on the dataverse
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDataverseWriteLock(LockList locks, DataverseName dataverseName) throws AlgebricksException;
+    void acquireDataverseWriteLock(LockList locks, String database, DataverseName dataverseName)
+            throws AlgebricksException;
 
     /**
      * Acquire read lock on the dataset (for queries)
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDatasetReadLock(LockList locks, DataverseName dataverseName, String datasetName)
+    void acquireDatasetReadLock(LockList locks, String database, DataverseName dataverseName, String datasetName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the dataset (for dataset create, dataset drop, and index drop)
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDatasetWriteLock(LockList locks, DataverseName dataverseName, String datasetName)
+    void acquireDatasetWriteLock(LockList locks, String database, DataverseName dataverseName, String datasetName)
             throws AlgebricksException;
 
     /**
      * Acquire modify lock on the dataset (for inserts, upserts, deletes) Mutually exclusive with create index lock
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDatasetModifyLock(LockList locks, DataverseName dataverseName, String datasetName)
+    void acquireDatasetModifyLock(LockList locks, String database, DataverseName dataverseName, String datasetName)
             throws AlgebricksException;
 
     /**
      * Acquire create index lock on the dataset (for index creation) Mutually exclusive with modify lock
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDatasetCreateIndexLock(LockList locks, DataverseName dataverseName, String datasetName)
+    void acquireDatasetCreateIndexLock(LockList locks, String database, DataverseName dataverseName, String datasetName)
             throws AlgebricksException;
 
     /**
@@ -113,168 +131,187 @@
      * exclusive with modify locks and index build lock
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDatasetExclusiveModificationLock(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException;
+    void acquireDatasetExclusiveModificationLock(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException;
 
     /**
      * Acquire read lock on the function
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param functionName
-     *            the name of the function in the given dataverse
+     *         the name of the function in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFunctionReadLock(LockList locks, DataverseName dataverseName, String functionName)
+    void acquireFunctionReadLock(LockList locks, String database, DataverseName dataverseName, String functionName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the function
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param functionName
-     *            the name of the function in the given dataverse
+     *         the name of the function in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFunctionWriteLock(LockList locks, DataverseName dataverseName, String functionName)
+    void acquireFunctionWriteLock(LockList locks, String database, DataverseName dataverseName, String functionName)
             throws AlgebricksException;
 
     /**
      * Acquire read lock on the full-text config
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param fullTextConfigName
-     *            the name of the full-text config in the given dataverse
+     *         the name of the full-text config in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFullTextConfigReadLock(LockList locks, DataverseName dataverseName, String fullTextConfigName)
-            throws AlgebricksException;
+    void acquireFullTextConfigReadLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextConfigName) throws AlgebricksException;
 
     /**
      * Acquire write lock on the full-text config
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param fullTextConfigName
-     *            the name of the full-text config in the given dataverse
+     *         the name of the full-text config in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFullTextConfigWriteLock(LockList locks, DataverseName dataverseName, String fullTextConfigName)
-            throws AlgebricksException;
+    void acquireFullTextConfigWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextConfigName) throws AlgebricksException;
 
     /**
      * Acquire read lock on the full-text filter
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param fullTextFilterName
-     *            the name of the full-text filter in the given dataverse
+     *         the name of the full-text filter in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFullTextFilterReadLock(LockList locks, DataverseName dataverseName, String fullTextFilterName)
-            throws AlgebricksException;
+    void acquireFullTextFilterReadLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextFilterName) throws AlgebricksException;
 
     /**
      * Acquire write lock on the full-text filter
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param fullTextFilterName
-     *            the name of the full-text filter in the given dataverse
+     *         the name of the full-text filter in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFullTextFilterWriteLock(LockList locks, DataverseName dataverseName, String fullTextFilterName)
-            throws AlgebricksException;
+    void acquireFullTextFilterWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextFilterName) throws AlgebricksException;
 
     /**
      * Acquire read lock on the library
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param libraryName
-     *            the name of the library in the given dataverse
+     *         the name of the library in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-
-    void acquireLibraryReadLock(LockList locks, DataverseName dataverseName, String libraryName)
+    void acquireLibraryReadLock(LockList locks, String database, DataverseName dataverseName, String libraryName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the library
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param libraryName
-     *            the name of the library in the given dataverse
+     *         the name of the library in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireLibraryWriteLock(LockList locks, DataverseName dataverseName, String libraryName)
+    void acquireLibraryWriteLock(LockList locks, String database, DataverseName dataverseName, String libraryName)
             throws AlgebricksException;
 
     /**
      * Acquire read lock on the adapter
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param adapterName
-     *            the name of the adapter in the given dataverse
+     *         the name of the adapter in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-
-    void acquireAdapterReadLock(LockList locks, DataverseName dataverseName, String adapterName)
+    void acquireAdapterReadLock(LockList locks, String database, DataverseName dataverseName, String adapterName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the adapter
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param adapterName
-     *            the name of the adapter in the given dataverse
+     *         the name of the adapter in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-
-    void acquireAdapterWriteLock(LockList locks, DataverseName dataverseName, String adapterName)
+    void acquireAdapterWriteLock(LockList locks, String database, DataverseName dataverseName, String adapterName)
             throws AlgebricksException;
 
     /**
@@ -305,60 +342,68 @@
      * Acquire read lock on the active entity
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param entityName
-     *            the name of the active entity in the given dataverse
+     *         the name of the active entity in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireActiveEntityReadLock(LockList locks, DataverseName dataverseName, String entityName)
+    void acquireActiveEntityReadLock(LockList locks, String database, DataverseName dataverseName, String entityName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the active entity
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param entityName
-     *            the name of the active entity in the given dataverse
+     *         the name of the active entity in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireActiveEntityWriteLock(LockList locks, DataverseName dataverseName, String entityName)
+    void acquireActiveEntityWriteLock(LockList locks, String database, DataverseName dataverseName, String entityName)
             throws AlgebricksException;
 
     /**
      * Acquire read lock on the feed policy
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param feedPolicyName
-     *            the name of the feed policy in the given dataverse
+     *         the name of the feed policy in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFeedPolicyWriteLock(LockList locks, DataverseName dataverseName, String feedPolicyName)
+    void acquireFeedPolicyWriteLock(LockList locks, String database, DataverseName dataverseName, String feedPolicyName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the feed policy
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param feedPolicyName
-     *            the name of the feed policy in the given dataverse
+     *         the name of the feed policy in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireFeedPolicyReadLock(LockList locks, DataverseName dataverseName, String feedPolicyName)
+    void acquireFeedPolicyReadLock(LockList locks, String database, DataverseName dataverseName, String feedPolicyName)
             throws AlgebricksException;
 
     /**
@@ -389,123 +434,139 @@
      * Acquire read lock on the data type
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datatypeName
-     *            the name of the data type in the given dataverse
+     *         the name of the data type in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDataTypeReadLock(LockList locks, DataverseName dataverseName, String datatypeName)
+    void acquireDataTypeReadLock(LockList locks, String database, DataverseName dataverseName, String datatypeName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the data type
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datatypeName
-     *            the name of the data type in the given dataverse
+     *         the name of the data type in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireDataTypeWriteLock(LockList locks, DataverseName dataverseName, String datatypeName)
+    void acquireDataTypeWriteLock(LockList locks, String database, DataverseName dataverseName, String datatypeName)
             throws AlgebricksException;
 
     /**
      * Acquire read lock on the synonym
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param synonymName
-     *            the name of the synonym in the given dataverse
+     *         the name of the synonym in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireSynonymReadLock(LockList locks, DataverseName dataverseName, String synonymName)
+    void acquireSynonymReadLock(LockList locks, String database, DataverseName dataverseName, String synonymName)
             throws AlgebricksException;
 
     /**
      * Acquire write lock on the synonym
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param synonymName
-     *            the name of the synonym in the given dataverse
+     *         the name of the synonym in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireSynonymWriteLock(LockList locks, DataverseName dataverseName, String synonymName)
+    void acquireSynonymWriteLock(LockList locks, String database, DataverseName dataverseName, String synonymName)
             throws AlgebricksException;
 
     /**
      * Acquire read lock on the extension entity
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
      * @param extension
-     *            the extension key
+     *         the extension key
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param extensionEntityName
-     *            the name of the extension entity in the given dataverse
+     *         the name of the extension entity in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireExtensionEntityReadLock(LockList locks, String extension, DataverseName dataverseName,
+    void acquireExtensionEntityReadLock(LockList locks, String extension, String database, DataverseName dataverseName,
             String extensionEntityName) throws AlgebricksException;
 
     /**
      * Acquire write lock on the extension entity
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
      * @param extension
-     *            the extension key
+     *         the extension key
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param extensionEntityName
-     *            the name of the extension entity in the given dataverse
+     *         the name of the extension entity in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be acquired
+     *         if lock couldn't be acquired
      */
-    void acquireExtensionEntityWriteLock(LockList locks, String extension, DataverseName dataverseName,
+    void acquireExtensionEntityWriteLock(LockList locks, String extension, String database, DataverseName dataverseName,
             String extensionEntityName) throws AlgebricksException;
 
     /**
      * Upgrade a previously acquired exclusive modification lock on the dataset to a write lock
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be upgraded
+     *         if lock couldn't be upgraded
      */
-    void upgradeDatasetLockToWrite(LockList locks, DataverseName dataverseName, String datasetName)
+    void upgradeDatasetLockToWrite(LockList locks, String database, DataverseName dataverseName, String datasetName)
             throws AlgebricksException;
 
     /**
      * Downgrade an upgraded dataset write lock to an exclusive modification lock
      *
      * @param locks
-     *            the lock list to add the new lock to
+     *         the lock list to add the new lock to
+     * @param database
+     *         the database name
      * @param dataverseName
-     *            the dataverse name
+     *         the dataverse name
      * @param datasetName
-     *            the name of the dataset in the given dataverse
+     *         the name of the dataset in the given dataverse
      * @throws AlgebricksException
-     *             if lock couldn't be downgraded
+     *         if lock couldn't be downgraded
      */
-    void downgradeDatasetLockToExclusiveModify(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException;
+    void downgradeDatasetLockToExclusiveModify(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException;
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/metadata/IMetadataLockUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/metadata/IMetadataLockUtil.java
index d830868..795e40f 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/metadata/IMetadataLockUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/metadata/IMetadataLockUtil.java
@@ -27,130 +27,143 @@
 
 public interface IMetadataLockUtil {
 
+    // Database helpers
+
+    void createDatabaseBegin(IMetadataLockManager lockManager, LockList locks, String database)
+            throws AlgebricksException;
+
+    void dropDatabaseBegin(IMetadataLockManager lockManager, LockList locks, String database)
+            throws AlgebricksException;
+
     // Dataverse helpers
 
-    void createDataverseBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName)
-            throws AlgebricksException;
+    void createDataverseBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName) throws AlgebricksException;
 
-    void dropDataverseBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName)
-            throws AlgebricksException;
+    void dropDataverseBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName) throws AlgebricksException;
 
     // Dataset helpers
 
-    void createDatasetBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName, DataverseName itemTypeDataverseName, String itemTypeName, boolean itemTypeAnonymous,
-            DataverseName metaItemTypeDataverseName, String metaItemTypeName, boolean metaItemTypeAnonymous,
-            String nodeGroupName, String compactionPolicyName, boolean isDefaultCompactionPolicy,
-            DatasetConfig.DatasetType datasetType, Object datasetDetails) throws AlgebricksException;
-
-    void dropDatasetBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException;
-
-    void modifyDatasetBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException;
-
-    void refreshDatasetBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException;
-
-    void analyzeDatasetBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException;
-
-    void analyzeDatasetDropBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException;
-
-    void compactBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String datasetName)
+    void createDatasetBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String itemTypeDatabase,
+            DataverseName itemTypeDataverseName, String itemTypeName, boolean itemTypeAnonymous,
+            String metaItemTypeDatabase, DataverseName metaItemTypeDataverseName, String metaItemTypeName,
+            boolean metaItemTypeAnonymous, String nodeGroupName, String compactionPolicyName,
+            boolean isDefaultCompactionPolicy, DatasetConfig.DatasetType datasetType, Object datasetDetails)
             throws AlgebricksException;
 
-    void insertDeleteUpsertBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
+    void dropDatasetBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
+    void modifyDatasetBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
+    void refreshDatasetBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
+    void analyzeDatasetBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
+    void analyzeDatasetDropBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
+    void compactBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
             String datasetName) throws AlgebricksException;
 
+    void insertDeleteUpsertBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException;
+
     // Index helpers
 
-    void createIndexBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName, String fullTextConfigName) throws AlgebricksException;
+    void createIndexBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String fullTextConfigName) throws AlgebricksException;
 
-    void dropIndexBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
+    void dropIndexBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
             String datasetName) throws AlgebricksException;
 
     // Type helpers
 
-    void createTypeBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String typeName)
-            throws AlgebricksException;
+    void createTypeBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String typeName) throws AlgebricksException;
 
-    void dropTypeBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String typeName)
-            throws AlgebricksException;
+    void dropTypeBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String typeName) throws AlgebricksException;
 
     // Library helpers
 
-    void createLibraryBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String libraryName) throws AlgebricksException;
+    void createLibraryBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String libraryName) throws AlgebricksException;
 
-    void dropLibraryBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String libraryName) throws AlgebricksException;
+    void dropLibraryBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String libraryName) throws AlgebricksException;
 
     // Function helpers
 
-    void createFunctionBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String functionName, DataverseName libraryDataverseName, String libraryName) throws AlgebricksException;
+    void createFunctionBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String functionName, String libraryDatabase,
+            DataverseName libraryDataverseName, String libraryName) throws AlgebricksException;
 
-    void dropFunctionBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String functionName) throws AlgebricksException;
+    void dropFunctionBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String functionName) throws AlgebricksException;
 
     // Full-text filter helpers
 
-    void createFullTextFilterBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String fullTextFilterName) throws AlgebricksException;
+    void createFullTextFilterBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String fullTextFilterName) throws AlgebricksException;
 
-    void dropFullTextFilterBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String fullTextFilterName) throws AlgebricksException;
+    void dropFullTextFilterBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String fullTextFilterName) throws AlgebricksException;
 
     // Full-text config helpers
 
-    void createFullTextConfigBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String fullTextConfigName, ImmutableList<String> fullTextFilterNames) throws AlgebricksException;
+    void createFullTextConfigBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String fullTextConfigName, ImmutableList<String> fullTextFilterNames)
+            throws AlgebricksException;
 
-    void dropFullTextConfigBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String fullTextFilterName) throws AlgebricksException;
+    void dropFullTextConfigBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String fullTextFilterName) throws AlgebricksException;
 
     // Adapter helpers
 
-    void createAdapterBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String adapterName, DataverseName libraryDataverseName, String libraryName) throws AlgebricksException;
+    void createAdapterBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String adapterName, String libraryDatabase, DataverseName libraryDataverseName,
+            String libraryName) throws AlgebricksException;
 
-    void dropAdapterBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String adapterName) throws AlgebricksException;
+    void dropAdapterBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String adapterName) throws AlgebricksException;
 
     // Synonym helpers
 
-    void createSynonymBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String synonymName) throws AlgebricksException;
+    void createSynonymBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String synonymName) throws AlgebricksException;
 
-    void dropSynonymBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String synonymName) throws AlgebricksException;
+    void dropSynonymBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String synonymName) throws AlgebricksException;
 
     // Feed helpers
 
-    void createFeedPolicyBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String policyName) throws AlgebricksException;
+    void createFeedPolicyBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String policyName) throws AlgebricksException;
 
-    void dropFeedPolicyBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String policyName) throws AlgebricksException;
+    void dropFeedPolicyBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String policyName) throws AlgebricksException;
 
-    void createFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String feedName)
-            throws AlgebricksException;
+    void createFeedBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String feedName) throws AlgebricksException;
 
-    void dropFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String feedName)
-            throws AlgebricksException;
+    void dropFeedBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String feedName) throws AlgebricksException;
 
-    void startFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String feedName)
-            throws AlgebricksException;
+    void startFeedBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String feedName) throws AlgebricksException;
 
-    void stopFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName, String feedName)
-            throws AlgebricksException;
+    void stopFeedBegin(IMetadataLockManager lockManager, LockList locks, String database, DataverseName dataverseName,
+            String feedName) throws AlgebricksException;
 
-    void connectFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName, String feedName) throws AlgebricksException;
+    void connectFeedBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String feedName) throws AlgebricksException;
 
-    void disconnectFeedBegin(IMetadataLockManager lockManager, LockList locks, DataverseName dataverseName,
-            String datasetName, String feedName) throws AlgebricksException;
+    void disconnectFeedBegin(IMetadataLockManager lockManager, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String feedName) throws AlgebricksException;
 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index 3a9f6c1..a1ba9c2 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -364,8 +364,9 @@
         } else if (dbName == null || dvName == null) {
             return null;
         }
-        appCtx.getMetadataLockManager().acquireDataverseReadLock(locks, dvName);
-        appCtx.getMetadataLockManager().acquireDatasetReadLock(locks, dvName, datasetName);
+        //TODO(DB): read lock on database
+        appCtx.getMetadataLockManager().acquireDataverseReadLock(locks, dbName, dvName);
+        appCtx.getMetadataLockManager().acquireDatasetReadLock(locks, dbName, dvName, datasetName);
         return MetadataManagerUtil.findDataset(mdTxnCtx, dbName, dvName, datasetName, includingViews);
     }
 
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockKey.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockKey.java
index 5e44e32..48b6ca5 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockKey.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockKey.java
@@ -30,6 +30,7 @@
         ACTIVE,
         DATASET,
         DATATYPE,
+        DATABASE,
         DATAVERSE,
         EXTENSION,
         FEED_POLICY,
@@ -47,17 +48,20 @@
 
     private final String entityKindExtension;
 
+    private final String database;
+
     private final DataverseName dataverseName;
 
     private final String entityName;
 
-    private MetadataLockKey(EntityKind entityKind, String entityKindExtension, DataverseName dataverseName,
-            String entityName) {
-        if (entityKind == null || (dataverseName == null && entityName == null)) {
+    private MetadataLockKey(EntityKind entityKind, String entityKindExtension, String database,
+            DataverseName dataverseName, String entityName) {
+        if (entityKind == null || (database == null && dataverseName == null && entityName == null)) {
             throw new NullPointerException();
         }
         this.entityKind = entityKind;
         this.entityKindExtension = entityKindExtension;
+        this.database = database;
         this.dataverseName = dataverseName;
         this.entityName = entityName;
     }
@@ -72,12 +76,13 @@
         }
         MetadataLockKey that = (MetadataLockKey) o;
         return entityKind == that.entityKind && Objects.equals(entityKindExtension, that.entityKindExtension)
-                && Objects.equals(dataverseName, that.dataverseName) && Objects.equals(entityName, that.entityName);
+                && Objects.equals(database, that.database) && Objects.equals(dataverseName, that.dataverseName)
+                && Objects.equals(entityName, that.entityName);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(entityKind, entityKindExtension, dataverseName, entityName);
+        return Objects.hash(entityKind, entityKindExtension, database, dataverseName, entityName);
     }
 
     @Override
@@ -94,60 +99,67 @@
         return sb.toString();
     }
 
-    static MetadataLockKey createDataverseLockKey(DataverseName dataverseName) {
-        return new MetadataLockKey(EntityKind.DATAVERSE, null, dataverseName, null);
+    static MetadataLockKey createDatabaseLockKey(String database) {
+        return new MetadataLockKey(EntityKind.DATABASE, null, database, null, null);
     }
 
-    static MetadataLockKey createDatasetLockKey(DataverseName dataverseName, String datasetName) {
-        return new MetadataLockKey(EntityKind.DATASET, null, dataverseName, datasetName);
+    static MetadataLockKey createDataverseLockKey(String database, DataverseName dataverseName) {
+        return new MetadataLockKey(EntityKind.DATAVERSE, null, database, dataverseName, null);
     }
 
-    static MetadataLockKey createDataTypeLockKey(DataverseName dataverseName, String datatypeName) {
-        return new MetadataLockKey(EntityKind.DATATYPE, null, dataverseName, datatypeName);
+    static MetadataLockKey createDatasetLockKey(String database, DataverseName dataverseName, String datasetName) {
+        return new MetadataLockKey(EntityKind.DATASET, null, database, dataverseName, datasetName);
     }
 
-    static MetadataLockKey createFunctionLockKey(DataverseName dataverseName, String functionName) {
-        return new MetadataLockKey(EntityKind.FUNCTION, null, dataverseName, functionName);
+    static MetadataLockKey createDataTypeLockKey(String database, DataverseName dataverseName, String datatypeName) {
+        return new MetadataLockKey(EntityKind.DATATYPE, null, database, dataverseName, datatypeName);
     }
 
-    static MetadataLockKey createFullTextConfigLockKey(DataverseName dataverseName, String fullTextConfigName) {
-        return new MetadataLockKey(EntityKind.FULL_TEXT_CONFIG, null, dataverseName, fullTextConfigName);
+    static MetadataLockKey createFunctionLockKey(String database, DataverseName dataverseName, String functionName) {
+        return new MetadataLockKey(EntityKind.FUNCTION, null, database, dataverseName, functionName);
     }
 
-    static MetadataLockKey createFullTextFilterLockKey(DataverseName dataverseName, String fullTextFilterName) {
-        return new MetadataLockKey(EntityKind.FULL_TEXT_FILTER, null, dataverseName, fullTextFilterName);
+    static MetadataLockKey createFullTextConfigLockKey(String database, DataverseName dataverseName,
+            String fullTextConfigName) {
+        return new MetadataLockKey(EntityKind.FULL_TEXT_CONFIG, null, database, dataverseName, fullTextConfigName);
     }
 
-    static MetadataLockKey createLibraryLockKey(DataverseName dataverseName, String libraryName) {
-        return new MetadataLockKey(EntityKind.LIBRARY, null, dataverseName, libraryName);
+    static MetadataLockKey createFullTextFilterLockKey(String database, DataverseName dataverseName,
+            String fullTextFilterName) {
+        return new MetadataLockKey(EntityKind.FULL_TEXT_FILTER, null, database, dataverseName, fullTextFilterName);
     }
 
-    static MetadataLockKey createAdapterLockKey(DataverseName dataverseName, String adapterName) {
-        return new MetadataLockKey(EntityKind.ADAPTER, null, dataverseName, adapterName);
+    static MetadataLockKey createLibraryLockKey(String database, DataverseName dataverseName, String libraryName) {
+        return new MetadataLockKey(EntityKind.LIBRARY, null, database, dataverseName, libraryName);
     }
 
-    static MetadataLockKey createActiveEntityLockKey(DataverseName dataverseName, String entityName) {
-        return new MetadataLockKey(EntityKind.ACTIVE, null, dataverseName, entityName);
+    static MetadataLockKey createAdapterLockKey(String database, DataverseName dataverseName, String adapterName) {
+        return new MetadataLockKey(EntityKind.ADAPTER, null, database, dataverseName, adapterName);
     }
 
-    static MetadataLockKey createFeedPolicyLockKey(DataverseName dataverseName, String feedPolicyName) {
-        return new MetadataLockKey(EntityKind.FEED_POLICY, null, dataverseName, feedPolicyName);
+    static MetadataLockKey createActiveEntityLockKey(String database, DataverseName dataverseName, String entityName) {
+        return new MetadataLockKey(EntityKind.ACTIVE, null, database, dataverseName, entityName);
     }
 
-    static MetadataLockKey createSynonymLockKey(DataverseName dataverseName, String synonymName) {
-        return new MetadataLockKey(EntityKind.SYNONYM, null, dataverseName, synonymName);
+    static MetadataLockKey createFeedPolicyLockKey(String database, DataverseName dataverseName,
+            String feedPolicyName) {
+        return new MetadataLockKey(EntityKind.FEED_POLICY, null, database, dataverseName, feedPolicyName);
     }
 
-    static MetadataLockKey createExtensionEntityLockKey(String extension, DataverseName dataverseName,
+    static MetadataLockKey createSynonymLockKey(String database, DataverseName dataverseName, String synonymName) {
+        return new MetadataLockKey(EntityKind.SYNONYM, null, database, dataverseName, synonymName);
+    }
+
+    static MetadataLockKey createExtensionEntityLockKey(String extension, String database, DataverseName dataverseName,
             String entityName) {
-        return new MetadataLockKey(EntityKind.EXTENSION, extension, dataverseName, entityName);
+        return new MetadataLockKey(EntityKind.EXTENSION, extension, database, dataverseName, entityName);
     }
 
     static MetadataLockKey createNodeGroupLockKey(String nodeGroupName) {
-        return new MetadataLockKey(EntityKind.NODE_GROUP, null, null, nodeGroupName);
+        return new MetadataLockKey(EntityKind.NODE_GROUP, null, null, null, nodeGroupName);
     }
 
     static MetadataLockKey createMergePolicyLockKey(String mergePolicyName) {
-        return new MetadataLockKey(EntityKind.MERGE_POLICY, null, null, mergePolicyName);
+        return new MetadataLockKey(EntityKind.MERGE_POLICY, null, null, null, mergePolicyName);
     }
 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
index 48508e1..e3c9aa2 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/MetadataLockManager.java
@@ -40,135 +40,151 @@
     }
 
     @Override
-    public void acquireDataverseReadLock(LockList locks, DataverseName dataverseName) throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDataverseLockKey(dataverseName);
+    public void acquireDatabaseReadLock(LockList locks, String database) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatabaseLockKey(database);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireDataverseWriteLock(LockList locks, DataverseName dataverseName) throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDataverseLockKey(dataverseName);
+    public void acquireDatabaseWriteLock(LockList locks, String database) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatabaseLockKey(database);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireDatasetReadLock(LockList locks, DataverseName dataverseName, String datasetName)
+    public void acquireDataverseReadLock(LockList locks, String database, DataverseName dataverseName)
             throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+        MetadataLockKey key = MetadataLockKey.createDataverseLockKey(database, dataverseName);
+        IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
+        locks.add(IMetadataLock.Mode.READ, lock);
+    }
+
+    @Override
+    public void acquireDataverseWriteLock(LockList locks, String database, DataverseName dataverseName)
+            throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDataverseLockKey(database, dataverseName);
+        IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
+        locks.add(IMetadataLock.Mode.WRITE, lock);
+    }
+
+    @Override
+    public void acquireDatasetReadLock(LockList locks, String database, DataverseName dataverseName, String datasetName)
+            throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireDatasetWriteLock(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void acquireDatasetWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireDatasetModifyLock(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void acquireDatasetModifyLock(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.MODIFY, lock);
     }
 
     @Override
-    public void acquireDatasetCreateIndexLock(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void acquireDatasetCreateIndexLock(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.INDEX_BUILD, lock);
     }
 
     @Override
-    public void acquireDatasetExclusiveModificationLock(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void acquireDatasetExclusiveModificationLock(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.EXCLUSIVE_MODIFY, lock);
     }
 
     @Override
-    public void acquireFunctionReadLock(LockList locks, DataverseName dataverseName, String synonymName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFunctionLockKey(dataverseName, synonymName);
+    public void acquireFunctionReadLock(LockList locks, String database, DataverseName dataverseName,
+            String synonymName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFunctionLockKey(database, dataverseName, synonymName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireFunctionWriteLock(LockList locks, DataverseName dataverseName, String synonymName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFunctionLockKey(dataverseName, synonymName);
+    public void acquireFunctionWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String synonymName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFunctionLockKey(database, dataverseName, synonymName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireFullTextConfigReadLock(LockList locks, DataverseName dataverseName, String fullTextConfigName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFullTextConfigLockKey(dataverseName, fullTextConfigName);
+    public void acquireFullTextConfigReadLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextConfigName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFullTextConfigLockKey(database, dataverseName, fullTextConfigName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireFullTextConfigWriteLock(LockList locks, DataverseName dataverseName, String fullTextConfigName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFullTextConfigLockKey(dataverseName, fullTextConfigName);
+    public void acquireFullTextConfigWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextConfigName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFullTextConfigLockKey(database, dataverseName, fullTextConfigName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireFullTextFilterReadLock(LockList locks, DataverseName dataverseName, String fullTextFilterName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFullTextFilterLockKey(dataverseName, fullTextFilterName);
+    public void acquireFullTextFilterReadLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextFilterName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFullTextFilterLockKey(database, dataverseName, fullTextFilterName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireFullTextFilterWriteLock(LockList locks, DataverseName dataverseName, String fullTextFilterName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFullTextFilterLockKey(dataverseName, fullTextFilterName);
+    public void acquireFullTextFilterWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String fullTextFilterName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFullTextFilterLockKey(database, dataverseName, fullTextFilterName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireLibraryReadLock(LockList locks, DataverseName dataverseName, String libraryName)
+    public void acquireLibraryReadLock(LockList locks, String database, DataverseName dataverseName, String libraryName)
             throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createLibraryLockKey(dataverseName, libraryName);
+        MetadataLockKey key = MetadataLockKey.createLibraryLockKey(database, dataverseName, libraryName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireLibraryWriteLock(LockList locks, DataverseName dataverseName, String libraryName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createLibraryLockKey(dataverseName, libraryName);
+    public void acquireLibraryWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String libraryName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createLibraryLockKey(database, dataverseName, libraryName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireAdapterReadLock(LockList locks, DataverseName dataverseName, String adapterName)
+    public void acquireAdapterReadLock(LockList locks, String database, DataverseName dataverseName, String adapterName)
             throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createAdapterLockKey(dataverseName, adapterName);
+        MetadataLockKey key = MetadataLockKey.createAdapterLockKey(database, dataverseName, adapterName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireAdapterWriteLock(LockList locks, DataverseName dataverseName, String adapterName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createAdapterLockKey(dataverseName, adapterName);
+    public void acquireAdapterWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String adapterName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createAdapterLockKey(database, dataverseName, adapterName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
@@ -188,33 +204,33 @@
     }
 
     @Override
-    public void acquireActiveEntityReadLock(LockList locks, DataverseName dataverseName, String entityName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createActiveEntityLockKey(dataverseName, entityName);
+    public void acquireActiveEntityReadLock(LockList locks, String database, DataverseName dataverseName,
+            String entityName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createActiveEntityLockKey(database, dataverseName, entityName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireActiveEntityWriteLock(LockList locks, DataverseName dataverseName, String entityName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createActiveEntityLockKey(dataverseName, entityName);
+    public void acquireActiveEntityWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String entityName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createActiveEntityLockKey(database, dataverseName, entityName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireFeedPolicyWriteLock(LockList locks, DataverseName dataverseName, String feedPolicyName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFeedPolicyLockKey(dataverseName, feedPolicyName);
+    public void acquireFeedPolicyWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String feedPolicyName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFeedPolicyLockKey(database, dataverseName, feedPolicyName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireFeedPolicyReadLock(LockList locks, DataverseName dataverseName, String feedPolicyName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createFeedPolicyLockKey(dataverseName, feedPolicyName);
+    public void acquireFeedPolicyReadLock(LockList locks, String database, DataverseName dataverseName,
+            String feedPolicyName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createFeedPolicyLockKey(database, dataverseName, feedPolicyName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
@@ -234,65 +250,67 @@
     }
 
     @Override
-    public void acquireDataTypeReadLock(LockList locks, DataverseName dataverseName, String datatypeName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDataTypeLockKey(dataverseName, datatypeName);
+    public void acquireDataTypeReadLock(LockList locks, String database, DataverseName dataverseName,
+            String datatypeName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDataTypeLockKey(database, dataverseName, datatypeName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireDataTypeWriteLock(LockList locks, DataverseName dataverseName, String datatypeName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDataTypeLockKey(dataverseName, datatypeName);
+    public void acquireDataTypeWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String datatypeName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDataTypeLockKey(database, dataverseName, datatypeName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireSynonymReadLock(LockList locks, DataverseName dataverseName, String synonymName)
+    public void acquireSynonymReadLock(LockList locks, String database, DataverseName dataverseName, String synonymName)
             throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createSynonymLockKey(dataverseName, synonymName);
+        MetadataLockKey key = MetadataLockKey.createSynonymLockKey(database, dataverseName, synonymName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireSynonymWriteLock(LockList locks, DataverseName dataverseName, String synonymName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createSynonymLockKey(dataverseName, synonymName);
+    public void acquireSynonymWriteLock(LockList locks, String database, DataverseName dataverseName,
+            String synonymName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createSynonymLockKey(database, dataverseName, synonymName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void acquireExtensionEntityReadLock(LockList locks, String extension, DataverseName dataverseName,
-            String entityName) throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createExtensionEntityLockKey(extension, dataverseName, entityName);
+    public void acquireExtensionEntityReadLock(LockList locks, String extension, String database,
+            DataverseName dataverseName, String entityName) throws AlgebricksException {
+        MetadataLockKey key =
+                MetadataLockKey.createExtensionEntityLockKey(extension, database, dataverseName, entityName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.READ, lock);
     }
 
     @Override
-    public void acquireExtensionEntityWriteLock(LockList locks, String extension, DataverseName dataverseName,
-            String entityName) throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createExtensionEntityLockKey(extension, dataverseName, entityName);
+    public void acquireExtensionEntityWriteLock(LockList locks, String extension, String database,
+            DataverseName dataverseName, String entityName) throws AlgebricksException {
+        MetadataLockKey key =
+                MetadataLockKey.createExtensionEntityLockKey(extension, database, dataverseName, entityName);
         IMetadataLock lock = mdlocks.computeIfAbsent(key, LOCK_FUNCTION);
         locks.add(IMetadataLock.Mode.WRITE, lock);
     }
 
     @Override
-    public void upgradeDatasetLockToWrite(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void upgradeDatasetLockToWrite(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.upgrade(IMetadataLock.Mode.UPGRADED_WRITE, lock);
     }
 
     @Override
-    public void downgradeDatasetLockToExclusiveModify(LockList locks, DataverseName dataverseName, String datasetName)
-            throws AlgebricksException {
-        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(dataverseName, datasetName);
+    public void downgradeDatasetLockToExclusiveModify(LockList locks, String database, DataverseName dataverseName,
+            String datasetName) throws AlgebricksException {
+        MetadataLockKey key = MetadataLockKey.createDatasetLockKey(database, dataverseName, datasetName);
         DatasetLock lock = (DatasetLock) mdlocks.computeIfAbsent(key, DATASET_LOCK_FUNCTION);
         locks.downgrade(IMetadataLock.Mode.EXCLUSIVE_MODIFY, lock);
     }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/MetadataLockUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/MetadataLockUtil.java
index 8ee8f11..6f44f4c 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/MetadataLockUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/MetadataLockUtil.java
@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 
 import org.apache.asterix.common.api.IMetadataLockManager;
 import org.apache.asterix.common.config.DatasetConfig;
@@ -35,55 +36,75 @@
 public class MetadataLockUtil implements IMetadataLockUtil {
 
     @Override
-    public void createDataverseBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName)
+    public void createDatabaseBegin(IMetadataLockManager lockMgr, LockList locks, String database)
             throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
+        //TODO(DB): write lock?
+        lockMgr.acquireDatabaseReadLock(locks, database);
     }
 
     @Override
-    public void dropDataverseBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName)
+    public void dropDatabaseBegin(IMetadataLockManager lockMgr, LockList locks, String database)
             throws AlgebricksException {
-        lockMgr.acquireDataverseWriteLock(locks, dataverseName);
+        lockMgr.acquireDatabaseWriteLock(locks, database);
     }
 
     @Override
-    public void createDatasetBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName, DataverseName itemTypeDataverseName, String itemTypeName, boolean itemTypeAnonymous,
-            DataverseName metaItemTypeDataverseName, String metaItemTypeName, boolean metaItemTypeAnonymous,
-            String nodeGroupName, String compactionPolicyName, boolean isDefaultCompactionPolicy,
-            DatasetConfig.DatasetType datasetType, Object datasetDetails) throws AlgebricksException {
-        createDatasetBeginPre(lockMgr, locks, dataverseName, itemTypeDataverseName, itemTypeName, itemTypeAnonymous,
-                metaItemTypeDataverseName, metaItemTypeName, metaItemTypeAnonymous, nodeGroupName, compactionPolicyName,
-                isDefaultCompactionPolicy);
-        lockMgr.acquireDatasetWriteLock(locks, dataverseName, datasetName);
+    public void createDataverseBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseWriteLock(locks, database, dataverseName);
     }
 
-    protected final void createDatasetBeginPre(IMetadataLockManager lockMgr, LockList locks,
-            DataverseName dataverseName, DataverseName itemTypeDataverseName, String itemTypeName,
-            boolean itemTypeAnonymous, DataverseName metaItemTypeDataverseName, String metaItemTypeName,
+    @Override
+    public void dropDataverseBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseWriteLock(locks, database, dataverseName);
+    }
+
+    @Override
+    public void createDatasetBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String itemTypeDatabase,
+            DataverseName itemTypeDataverseName, String itemTypeName, boolean itemTypeAnonymous,
+            String metaItemTypeDatabase, DataverseName metaItemTypeDataverseName, String metaItemTypeName,
             boolean metaItemTypeAnonymous, String nodeGroupName, String compactionPolicyName,
-            boolean isDefaultCompactionPolicy) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        if (itemTypeDataverseName != null && !dataverseName.equals(itemTypeDataverseName)) {
-            lockMgr.acquireDataverseReadLock(locks, itemTypeDataverseName);
-        }
-        if (metaItemTypeDataverseName != null && !metaItemTypeDataverseName.equals(dataverseName)
-                && !metaItemTypeDataverseName.equals(itemTypeDataverseName)) {
-            lockMgr.acquireDataverseReadLock(locks, metaItemTypeDataverseName);
-        }
+            boolean isDefaultCompactionPolicy, DatasetConfig.DatasetType datasetType, Object datasetDetails)
+            throws AlgebricksException {
+        createDatasetBeginPre(lockMgr, locks, database, dataverseName, itemTypeDatabase, itemTypeDataverseName,
+                itemTypeName, itemTypeAnonymous, metaItemTypeDatabase, metaItemTypeDataverseName, metaItemTypeName,
+                metaItemTypeAnonymous, nodeGroupName, compactionPolicyName, isDefaultCompactionPolicy);
+        lockMgr.acquireDatasetWriteLock(locks, database, dataverseName, datasetName);
+    }
+
+    protected final void createDatasetBeginPre(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String itemTypeDatabase, DataverseName itemTypeDataverseName,
+            String itemTypeName, boolean itemTypeAnonymous, String metaItemTypeDatabase,
+            DataverseName metaItemTypeDataverseName, String metaItemTypeName, boolean metaItemTypeAnonymous,
+            String nodeGroupName, String compactionPolicyName, boolean isDefaultCompactionPolicy)
+            throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockIfDifferentNamespace(lockMgr, locks, database, dataverseName, itemTypeDatabase, itemTypeDataverseName);
+        lockIfDifferentNamespace(lockMgr, locks, database, dataverseName, itemTypeDatabase, itemTypeDataverseName,
+                metaItemTypeDatabase, metaItemTypeDataverseName);
+
         if (itemTypeAnonymous) {
             // the datatype will be created
-            lockMgr.acquireDataTypeWriteLock(locks, itemTypeDataverseName, itemTypeName);
+            lockMgr.acquireDataTypeWriteLock(locks, itemTypeDatabase, itemTypeDataverseName, itemTypeName);
         } else {
-            lockMgr.acquireDataTypeReadLock(locks, itemTypeDataverseName, itemTypeName);
+            lockMgr.acquireDataTypeReadLock(locks, itemTypeDatabase, itemTypeDataverseName, itemTypeName);
         }
-        if (metaItemTypeDataverseName != null && !metaItemTypeDataverseName.equals(itemTypeDataverseName)
+        if (metaItemTypeDatabase != null && metaItemTypeDataverseName != null && !sameNamespace(metaItemTypeDatabase,
+                metaItemTypeDataverseName, itemTypeDatabase, itemTypeDataverseName)
                 && !metaItemTypeName.equals(itemTypeName)) {
+            //TODO(DB): why check the type name?
             if (metaItemTypeAnonymous) {
                 // the datatype will be created
-                lockMgr.acquireDataTypeWriteLock(locks, metaItemTypeDataverseName, metaItemTypeName);
+                lockMgr.acquireDataTypeWriteLock(locks, metaItemTypeDatabase, metaItemTypeDataverseName,
+                        metaItemTypeName);
             } else {
-                lockMgr.acquireDataTypeReadLock(locks, metaItemTypeDataverseName, metaItemTypeName);
+                lockMgr.acquireDataTypeReadLock(locks, metaItemTypeDatabase, metaItemTypeDataverseName,
+                        metaItemTypeName);
             }
         }
         if (nodeGroupName != null) {
@@ -95,96 +116,108 @@
     }
 
     @Override
-    public void createIndexBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName, String fullTextConfigName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetCreateIndexLock(locks, dataverseName, datasetName);
+    public void createIndexBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String fullTextConfigName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetCreateIndexLock(locks, database, dataverseName, datasetName);
         if (!Strings.isNullOrEmpty(fullTextConfigName)) {
-            lockMgr.acquireFullTextConfigReadLock(locks, dataverseName, fullTextConfigName);
+            lockMgr.acquireFullTextConfigReadLock(locks, database, dataverseName, fullTextConfigName);
         }
     }
 
     @Override
-    public void dropIndexBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetWriteLock(locks, dataverseName, datasetName);
+    public void dropIndexBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetWriteLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void createTypeBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String typeName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDataTypeWriteLock(locks, dataverseName, typeName);
+    public void createTypeBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String typeName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDataTypeWriteLock(locks, database, dataverseName, typeName);
     }
 
     @Override
-    public void dropDatasetBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetWriteLock(locks, dataverseName, datasetName);
+    public void dropDatasetBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetWriteLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void dropTypeBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String typeName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDataTypeWriteLock(locks, dataverseName, typeName);
+    public void dropTypeBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String typeName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDataTypeWriteLock(locks, database, dataverseName, typeName);
     }
 
     @Override
-    public void createLibraryBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String libraryName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireLibraryWriteLock(locks, dataverseName, libraryName);
+    public void createLibraryBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String libraryName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireLibraryWriteLock(locks, database, dataverseName, libraryName);
     }
 
     @Override
-    public void dropLibraryBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String libraryName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireLibraryWriteLock(locks, dataverseName, libraryName);
+    public void dropLibraryBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String libraryName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireLibraryWriteLock(locks, database, dataverseName, libraryName);
     }
 
     @Override
-    public void createFunctionBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String functionName, DataverseName libraryDataverseName, String libraryName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFunctionWriteLock(locks, dataverseName, functionName);
+    public void createFunctionBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String functionName, String libraryDatabase,
+            DataverseName libraryDataverseName, String libraryName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFunctionWriteLock(locks, database, dataverseName, functionName);
         if (libraryName != null) {
-            if (!dataverseName.equals(libraryDataverseName)) {
-                lockMgr.acquireDataverseReadLock(locks, libraryDataverseName);
-            }
-            lockMgr.acquireLibraryReadLock(locks, libraryDataverseName, libraryName);
+            lockIfDifferentNamespace(lockMgr, locks, database, dataverseName, libraryDatabase, libraryDataverseName);
+            lockMgr.acquireLibraryReadLock(locks, libraryDatabase, libraryDataverseName, libraryName);
         }
     }
 
     @Override
-    public void dropFunctionBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String functionName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFunctionWriteLock(locks, dataverseName, functionName);
+    public void dropFunctionBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String functionName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFunctionWriteLock(locks, database, dataverseName, functionName);
     }
 
     @Override
-    public void createFullTextFilterBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String fullTextFilterName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFullTextFilterWriteLock(locks, dataverseName, fullTextFilterName);
+    public void createFullTextFilterBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String fullTextFilterName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFullTextFilterWriteLock(locks, database, dataverseName, fullTextFilterName);
     }
 
     @Override
-    public void dropFullTextFilterBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String fullTextFilterName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFullTextFilterWriteLock(locks, dataverseName, fullTextFilterName);
+    public void dropFullTextFilterBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String fullTextFilterName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFullTextFilterWriteLock(locks, database, dataverseName, fullTextFilterName);
     }
 
     @Override
-    public void createFullTextConfigBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String fullTextConfigName, ImmutableList<String> fullTextFilterNames) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFullTextConfigWriteLock(locks, dataverseName, fullTextConfigName);
+    public void createFullTextConfigBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String fullTextConfigName, ImmutableList<String> fullTextFilterNames)
+            throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFullTextConfigWriteLock(locks, database, dataverseName, fullTextConfigName);
 
         // We should avoid sorting the original list, and the original list is immutable and cannot be sorted anyway
         List<String> fullTextFilterNamesMutable = new ArrayList<>(fullTextFilterNames);
@@ -192,149 +225,202 @@
         // sort the filters to guarantee locks are always fetched in the same order to avoid dead lock between filters
         Collections.sort(fullTextFilterNamesMutable);
         for (String filterName : fullTextFilterNamesMutable) {
-            lockMgr.acquireFullTextFilterReadLock(locks, dataverseName, filterName);
+            lockMgr.acquireFullTextFilterReadLock(locks, database, dataverseName, filterName);
         }
     }
 
     @Override
-    public void dropFullTextConfigBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String configName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFullTextConfigWriteLock(locks, dataverseName, configName);
+    public void dropFullTextConfigBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String configName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFullTextConfigWriteLock(locks, database, dataverseName, configName);
     }
 
     @Override
-    public void createAdapterBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String adapterName, DataverseName libraryDataverseName, String libraryName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireAdapterWriteLock(locks, dataverseName, adapterName);
+    public void createAdapterBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String adapterName, String libraryDatabase, DataverseName libraryDataverseName,
+            String libraryName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireAdapterWriteLock(locks, database, dataverseName, adapterName);
         if (libraryName != null) {
-            if (!dataverseName.equals(libraryDataverseName)) {
-                lockMgr.acquireDataverseReadLock(locks, libraryDataverseName);
-            }
-            lockMgr.acquireLibraryReadLock(locks, libraryDataverseName, libraryName);
+            lockIfDifferentNamespace(lockMgr, locks, database, dataverseName, libraryDatabase, libraryDataverseName);
+            lockMgr.acquireLibraryReadLock(locks, libraryDatabase, libraryDataverseName, libraryName);
         }
     }
 
     @Override
-    public void dropAdapterBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String adapterName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireAdapterWriteLock(locks, dataverseName, adapterName);
+    public void dropAdapterBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String adapterName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireAdapterWriteLock(locks, database, dataverseName, adapterName);
     }
 
     @Override
-    public void createSynonymBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String synonymName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireSynonymWriteLock(locks, dataverseName, synonymName);
+    public void createSynonymBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String synonymName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireSynonymWriteLock(locks, database, dataverseName, synonymName);
     }
 
     @Override
-    public void dropSynonymBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String synonymName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireSynonymWriteLock(locks, dataverseName, synonymName);
+    public void dropSynonymBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String synonymName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireSynonymWriteLock(locks, database, dataverseName, synonymName);
     }
 
     @Override
-    public void modifyDatasetBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetModifyLock(locks, dataverseName, datasetName);
+    public void modifyDatasetBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetModifyLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void insertDeleteUpsertBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetExclusiveModificationLock(locks, dataverseName, datasetName);
+    public void insertDeleteUpsertBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetExclusiveModificationLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void dropFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String feedName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityWriteLock(locks, dataverseName, feedName);
+    public void dropFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String feedName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityWriteLock(locks, database, dataverseName, feedName);
     }
 
     @Override
-    public void dropFeedPolicyBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String policyName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityWriteLock(locks, dataverseName, policyName);
+    public void dropFeedPolicyBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String policyName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityWriteLock(locks, database, dataverseName, policyName);
     }
 
     @Override
-    public void startFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String feedName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityReadLock(locks, dataverseName, feedName);
+    public void startFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String feedName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityReadLock(locks, database, dataverseName, feedName);
     }
 
     @Override
-    public void stopFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String feedName) throws AlgebricksException {
+    public void stopFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String feedName) throws AlgebricksException {
         // TODO: dataset lock?
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityReadLock(locks, dataverseName, feedName);
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityReadLock(locks, database, dataverseName, feedName);
     }
 
     @Override
-    public void createFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String feedName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityWriteLock(locks, dataverseName, feedName);
+    public void createFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String feedName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityWriteLock(locks, database, dataverseName, feedName);
     }
 
     @Override
-    public void connectFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName, String feedName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityReadLock(locks, dataverseName, feedName);
-        lockMgr.acquireDatasetReadLock(locks, dataverseName, datasetName);
+    public void connectFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String feedName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityReadLock(locks, database, dataverseName, feedName);
+        lockMgr.acquireDatasetReadLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void createFeedPolicyBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String policyName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireFeedPolicyWriteLock(locks, dataverseName, policyName);
+    public void createFeedPolicyBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String policyName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireFeedPolicyWriteLock(locks, database, dataverseName, policyName);
     }
 
     @Override
-    public void disconnectFeedBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName, String feedName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireActiveEntityReadLock(locks, dataverseName, feedName);
-        lockMgr.acquireDatasetReadLock(locks, dataverseName, datasetName);
+    public void disconnectFeedBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName, String feedName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireActiveEntityReadLock(locks, database, dataverseName, feedName);
+        lockMgr.acquireDatasetReadLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void analyzeDatasetBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
+    public void analyzeDatasetBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetCreateIndexLock(locks, database, dataverseName, datasetName);
+    }
+
+    @Override
+    public void analyzeDatasetDropBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetWriteLock(locks, database, dataverseName, datasetName);
+    }
+
+    @Override
+    public void compactBegin(IMetadataLockManager lockMgr, LockList locks, String database, DataverseName dataverseName,
             String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetCreateIndexLock(locks, dataverseName, datasetName);
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetReadLock(locks, database, dataverseName, datasetName);
     }
 
     @Override
-    public void analyzeDatasetDropBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetWriteLock(locks, dataverseName, datasetName);
+    public void refreshDatasetBegin(IMetadataLockManager lockMgr, LockList locks, String database,
+            DataverseName dataverseName, String datasetName) throws AlgebricksException {
+        lockMgr.acquireDatabaseReadLock(locks, database);
+        lockMgr.acquireDataverseReadLock(locks, database, dataverseName);
+        lockMgr.acquireDatasetExclusiveModificationLock(locks, database, dataverseName, datasetName);
     }
 
-    @Override
-    public void compactBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetReadLock(locks, dataverseName, datasetName);
+    private static void lockIfDifferentNamespace(IMetadataLockManager lockMgr, LockList locks, String lockedDatabase,
+            DataverseName lockedDataverse, String toBeLockedDatabase, DataverseName toBeLockedDataverse)
+            throws AlgebricksException {
+        if (toBeLockedDatabase != null && toBeLockedDataverse != null) {
+            if (!Objects.equals(lockedDatabase, toBeLockedDatabase)) {
+                lockMgr.acquireDatabaseReadLock(locks, toBeLockedDatabase);
+                lockMgr.acquireDataverseReadLock(locks, toBeLockedDatabase, toBeLockedDataverse);
+            } else if (!Objects.equals(lockedDataverse, toBeLockedDataverse)) {
+                lockMgr.acquireDataverseReadLock(locks, toBeLockedDatabase, toBeLockedDataverse);
+            }
+        }
     }
 
-    @Override
-    public void refreshDatasetBegin(IMetadataLockManager lockMgr, LockList locks, DataverseName dataverseName,
-            String datasetName) throws AlgebricksException {
-        lockMgr.acquireDataverseReadLock(locks, dataverseName);
-        lockMgr.acquireDatasetExclusiveModificationLock(locks, dataverseName, datasetName);
+    private static void lockIfDifferentNamespace(IMetadataLockManager lockMgr, LockList locks, String lockedDatabase1,
+            DataverseName lockedDataverse1, String lockedDatabase2, DataverseName lockedDataverse2,
+            String toBeLockedDatabase, DataverseName toBeLockedDataverse) throws AlgebricksException {
+        if (toBeLockedDatabase != null && toBeLockedDataverse != null) {
+            if (!Objects.equals(lockedDatabase1, toBeLockedDatabase)) {
+                if (!Objects.equals(lockedDatabase2, toBeLockedDatabase)) {
+                    lockMgr.acquireDatabaseReadLock(locks, toBeLockedDatabase);
+                    lockMgr.acquireDataverseReadLock(locks, toBeLockedDatabase, toBeLockedDataverse);
+                } else if (!Objects.equals(lockedDataverse2, toBeLockedDataverse)) {
+                    lockMgr.acquireDataverseReadLock(locks, toBeLockedDatabase, toBeLockedDataverse);
+                }
+            } else if (!Objects.equals(lockedDataverse1, toBeLockedDataverse)) {
+                lockMgr.acquireDataverseReadLock(locks, toBeLockedDatabase, toBeLockedDataverse);
+            }
+        }
+    }
+
+    private static boolean sameNamespace(String database1, DataverseName dataverse1, String database2,
+            DataverseName dataverse2) {
+        return database1.equals(database2) && dataverse1.equals(dataverse2);
     }
 }
diff --git a/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java
index 9fbc1c2..cf95705 100644
--- a/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java
+++ b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java
@@ -24,6 +24,7 @@
 
 import org.apache.asterix.common.metadata.DataverseName;
 import org.apache.asterix.common.metadata.LockList;
+import org.apache.asterix.common.metadata.MetadataUtil;
 import org.apache.hyracks.api.util.SingleThreadEventProcessor;
 import org.junit.Assert;
 import org.junit.Test;
@@ -33,6 +34,7 @@
 @RunWith(Parameterized.class)
 public class MetadataLockManagerTest {
 
+    //TODO(DB): adapt test for database
     static final int REPREAT_TEST_COUNT = 3;
 
     @Parameterized.Parameters
@@ -50,12 +52,14 @@
         }
 
         private final Statement statement;
+        private final String database;
         private final DataverseName dataverseName;
         private final String datasetName;
         private boolean done;
         private int step = 0;
 
-        public Request(Statement statement, DataverseName dataverseName, String datasetName) {
+        public Request(Statement statement, String database, DataverseName dataverseName, String datasetName) {
+            this.database = database;
             this.statement = statement;
             this.dataverseName = dataverseName;
             this.datasetName = datasetName;
@@ -66,6 +70,10 @@
             return statement;
         }
 
+        String database() {
+            return database;
+        }
+
         DataverseName dataverse() {
             return dataverseName;
         }
@@ -122,28 +130,33 @@
                 step.acquire();
                 switch (req.statement()) {
                     case INDEX:
-                        lockManager.acquireDatasetCreateIndexLock(locks, req.dataverse(), req.dataset());
+                        lockManager.acquireDatasetCreateIndexLock(locks, req.database(), req.dataverse(),
+                                req.dataset());
                         break;
                     case MODIFY:
-                        lockManager.acquireDatasetModifyLock(locks, req.dataverse(), req.dataset());
+                        lockManager.acquireDatasetModifyLock(locks, req.database(), req.dataverse(), req.dataset());
                         break;
                     case EXCLUSIVE_MODIFY:
-                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataverse(), req.dataset());
+                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.database(), req.dataverse(),
+                                req.dataset());
                         break;
                     case EXCLUSIVE_MODIFY_UPGRADE:
-                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataverse(), req.dataset());
+                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.database(), req.dataverse(),
+                                req.dataset());
                         req.step();
                         step.acquire();
-                        lockManager.upgradeDatasetLockToWrite(locks, req.dataverse(), req.dataset());
+                        lockManager.upgradeDatasetLockToWrite(locks, req.database(), req.dataverse(), req.dataset());
                         break;
                     case EXCLUSIVE_MODIFY_UPGRADE_DOWNGRADE:
-                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataverse(), req.dataset());
+                        lockManager.acquireDatasetExclusiveModificationLock(locks, req.database(), req.dataverse(),
+                                req.dataset());
                         req.step();
                         step.acquire();
-                        lockManager.upgradeDatasetLockToWrite(locks, req.dataverse(), req.dataset());
+                        lockManager.upgradeDatasetLockToWrite(locks, req.database(), req.dataverse(), req.dataset());
                         req.step();
                         step.acquire();
-                        lockManager.downgradeDatasetLockToExclusiveModify(locks, req.dataverse(), req.dataset());
+                        lockManager.downgradeDatasetLockToExclusiveModify(locks, req.database(), req.dataverse(),
+                                req.dataset());
                         break;
                     default:
                         break;
@@ -163,13 +176,14 @@
     public void testDatasetLockMultipleIndexBuildsSingleModifier() throws Exception {
         MetadataLockManager lockManager = new MetadataLockManager();
         DataverseName dataverseName = DataverseName.createSinglePartName("Dataverse");
+        String database = MetadataUtil.databaseFor(dataverseName);
         String datasetName = "Dataset";
         User till = new User("till", lockManager);
-        Request tReq = new Request(Request.Statement.INDEX, dataverseName, datasetName);
+        Request tReq = new Request(Request.Statement.INDEX, database, dataverseName, datasetName);
         User dmitry = new User("dmitry", lockManager);
-        Request dReq = new Request(Request.Statement.INDEX, dataverseName, datasetName);
+        Request dReq = new Request(Request.Statement.INDEX, database, dataverseName, datasetName);
         User mike = new User("mike", lockManager);
-        Request mReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        Request mReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         // Till builds an index
         till.add(tReq);
         // Dmitry builds an index
@@ -211,13 +225,14 @@
     public void testDatasetLockMultipleModifiersSingleIndexBuilder() throws Exception {
         MetadataLockManager lockManager = new MetadataLockManager();
         DataverseName dataverseName = DataverseName.createSinglePartName("Dataverse");
+        String database = MetadataUtil.databaseFor(dataverseName);
         String datasetName = "Dataset";
         User till = new User("till", lockManager);
-        Request tReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        Request tReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         User dmitry = new User("dmitry", lockManager);
-        Request dReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        Request dReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         User mike = new User("mike", lockManager);
-        Request mReq = new Request(Request.Statement.INDEX, dataverseName, datasetName);
+        Request mReq = new Request(Request.Statement.INDEX, database, dataverseName, datasetName);
         // Till modifies
         till.add(tReq);
         // Dmitry modifies
@@ -259,13 +274,14 @@
     public void testDatasetLockMultipleModifiersSingleExclusiveModifier() throws Exception {
         MetadataLockManager lockManager = new MetadataLockManager();
         DataverseName dataverseName = DataverseName.createSinglePartName("Dataverse");
+        String database = MetadataUtil.databaseFor(dataverseName);
         String datasetName = "Dataset";
         User till = new User("till", lockManager);
-        Request tReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        Request tReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         User dmitry = new User("dmitry", lockManager);
-        Request dReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        Request dReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         User mike = new User("mike", lockManager);
-        Request mReq = new Request(Request.Statement.EXCLUSIVE_MODIFY, dataverseName, datasetName);
+        Request mReq = new Request(Request.Statement.EXCLUSIVE_MODIFY, database, dataverseName, datasetName);
         // Till starts
         till.add(tReq);
         till.step();
@@ -290,7 +306,7 @@
         // Ensure that Mike got the lock
         mReq.await(1);
         // Till submits another request
-        tReq = new Request(Request.Statement.MODIFY, dataverseName, datasetName);
+        tReq = new Request(Request.Statement.MODIFY, database, dataverseName, datasetName);
         till.add(tReq);
         till.step();
         // Ensure that Till didn't get the lock