diff --git a/src/common/exception/src/exception_code.rs b/src/common/exception/src/exception_code.rs
index cacef1591d08..c6b84a7cad5b 100644
--- a/src/common/exception/src/exception_code.rs
+++ b/src/common/exception/src/exception_code.rs
@@ -385,6 +385,11 @@ build_exceptions! {
     // dictionary
     DictionaryAlreadyExists(3113),
     UnknownDictionary(3114),
+    UnknownDictionaryId(3115),
+    UnsupportedDictionaryOption(3116),
+    UnsupportedDictionarySource(3117),
+    MissingDictionaryOption(3118),
+    WrongDictionaryFieldExpr(3119),
 }
 
 // Storage errors [3001, 4000].
diff --git a/src/meta/app/src/app_error.rs b/src/meta/app/src/app_error.rs
index 9b7c52885e30..2ff5ffd5571a 100644
--- a/src/meta/app/src/app_error.rs
+++ b/src/meta/app/src/app_error.rs
@@ -1749,7 +1749,7 @@ impl AppErrorMessage for SequenceError {
 // dictionary
 impl AppErrorMessage for DictionaryAlreadyExists {
     fn message(&self) -> String {
-        format!("dictionary '{}' already exists", self.dictionary_name)
+        format!("Dictionary '{}' already exists", self.dictionary_name)
     }
 }
 
diff --git a/src/query/catalog/src/catalog/interface.rs b/src/query/catalog/src/catalog/interface.rs
index f006dc36588c..f1e7512c6f19 100644
--- a/src/query/catalog/src/catalog/interface.rs
+++ b/src/query/catalog/src/catalog/interface.rs
@@ -19,11 +19,14 @@ use std::sync::Arc;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -37,6 +40,8 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryIdentity;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -53,6 +58,7 @@ use databend_common_meta_app::schema::DroppedId;
 use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GcDroppedTableResp;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -62,6 +68,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
@@ -85,6 +92,8 @@ use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableByIdReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
@@ -296,6 +305,36 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
         }
     }
 
+    // Check a db.dictionary is exists or not.
+    #[async_backtrace::framed]
+    async fn exists_dictionary(
+        &self,
+        tenant: &Tenant,
+        db_name: &str,
+        dict_name: &str,
+    ) -> Result<bool> {
+        let db_id = self
+            .get_database(tenant, db_name)
+            .await?
+            .get_db_info()
+            .database_id
+            .db_id;
+        let req = TenantDictionaryIdent::new(
+            tenant,
+            DictionaryIdentity::new(db_id, dict_name.to_string()),
+        );
+        match self.get_dictionary(req).await {
+            Ok(_) => Ok(true),
+            Err(err) => {
+                if err.code() == ErrorCode::UNKNOWN_DICTIONARY {
+                    Ok(false)
+                } else {
+                    Err(err)
+                }
+            }
+        }
+    }
+
     async fn upsert_table_option(
         &self,
         tenant: &Tenant,
@@ -433,4 +472,24 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
     ) -> Result<GetSequenceNextValueReply>;
 
     async fn drop_sequence(&self, req: DropSequenceReq) -> Result<DropSequenceReply>;
+
+    /// Dictionary
+    async fn create_dictionary(&self, req: CreateDictionaryReq) -> Result<CreateDictionaryReply>;
+
+    async fn update_dictionary(&self, req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply>;
+
+    async fn drop_dictionary(
+        &self,
+        dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>>;
+
+    async fn get_dictionary(
+        &self,
+        req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>>;
+
+    async fn list_dictionaries(
+        &self,
+        req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>>;
 }
diff --git a/src/query/catalog/src/catalog/session_catalog.rs b/src/query/catalog/src/catalog/session_catalog.rs
index 7814464e1091..913e16b70fce 100644
--- a/src/query/catalog/src/catalog/session_catalog.rs
+++ b/src/query/catalog/src/catalog/session_catalog.rs
@@ -17,11 +17,14 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -35,6 +38,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -51,6 +55,7 @@ use databend_common_meta_app::schema::DroppedId;
 use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GcDroppedTableResp;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -60,6 +65,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
@@ -83,6 +89,8 @@ use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableByIdReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
@@ -501,4 +509,34 @@ impl Catalog for SessionCatalog {
     async fn drop_sequence(&self, req: DropSequenceReq) -> Result<DropSequenceReply> {
         self.inner.drop_sequence(req).await
     }
+
+    /// Dictionary
+    async fn create_dictionary(&self, req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        self.inner.create_dictionary(req).await
+    }
+
+    async fn update_dictionary(&self, req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        self.inner.update_dictionary(req).await
+    }
+
+    async fn drop_dictionary(
+        &self,
+        dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        self.inner.drop_dictionary(dict_ident).await
+    }
+
+    async fn get_dictionary(
+        &self,
+        req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        self.inner.get_dictionary(req).await
+    }
+
+    async fn list_dictionaries(
+        &self,
+        req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        self.inner.list_dictionaries(req).await
+    }
 }
diff --git a/src/query/service/src/catalogs/default/database_catalog.rs b/src/query/service/src/catalogs/default/database_catalog.rs
index d0efd0d8d3f1..7add4960253e 100644
--- a/src/query/service/src/catalogs/default/database_catalog.rs
+++ b/src/query/service/src/catalogs/default/database_catalog.rs
@@ -25,11 +25,14 @@ use databend_common_catalog::table_function::TableFunction;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -43,6 +46,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -59,6 +63,7 @@ use databend_common_meta_app::schema::DroppedId;
 use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GcDroppedTableResp;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -68,6 +73,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
@@ -91,6 +97,8 @@ use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableByIdReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
@@ -732,4 +740,39 @@ impl Catalog for DatabaseCatalog {
     async fn drop_sequence(&self, req: DropSequenceReq) -> Result<DropSequenceReply> {
         self.mutable_catalog.drop_sequence(req).await
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        self.mutable_catalog.create_dictionary(req).await
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        self.mutable_catalog.update_dictionary(req).await
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        self.mutable_catalog.drop_dictionary(dict_ident).await
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        self.mutable_catalog.get_dictionary(req).await
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        self.mutable_catalog.list_dictionaries(req).await
+    }
 }
diff --git a/src/query/service/src/catalogs/default/immutable_catalog.rs b/src/query/service/src/catalogs/default/immutable_catalog.rs
index 544d4b6b10cf..cfc15e0872d2 100644
--- a/src/query/service/src/catalogs/default/immutable_catalog.rs
+++ b/src/query/service/src/catalogs/default/immutable_catalog.rs
@@ -21,11 +21,14 @@ use databend_common_catalog::catalog::Catalog;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -39,6 +42,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -52,6 +56,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -61,6 +66,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -83,6 +89,8 @@ use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableByIdReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateVirtualColumnReply;
@@ -511,4 +519,39 @@ impl Catalog for ImmutableCatalog {
     async fn drop_sequence(&self, _req: DropSequenceReq) -> Result<DropSequenceReply> {
         unimplemented!()
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        _req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        unimplemented!()
+    }
 }
diff --git a/src/query/service/src/catalogs/default/mutable_catalog.rs b/src/query/service/src/catalogs/default/mutable_catalog.rs
index dc4024451ed2..1d0a1f7c440e 100644
--- a/src/query/service/src/catalogs/default/mutable_catalog.rs
+++ b/src/query/service/src/catalogs/default/mutable_catalog.rs
@@ -25,11 +25,14 @@ use databend_common_exception::Result;
 use databend_common_meta_api::SchemaApi;
 use databend_common_meta_api::SequenceApi;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -48,6 +51,7 @@ use databend_common_meta_app::schema::DatabaseInfo;
 use databend_common_meta_app::schema::DatabaseMeta;
 use databend_common_meta_app::schema::DatabaseType;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -65,6 +69,7 @@ use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GcDroppedTableResp;
 use databend_common_meta_app::schema::GetDatabaseReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -75,6 +80,7 @@ use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
 use databend_common_meta_app::schema::ListDatabaseReq;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
@@ -98,6 +104,8 @@ use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableByIdReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
@@ -654,4 +662,41 @@ impl Catalog for MutableCatalog {
     async fn drop_sequence(&self, req: DropSequenceReq) -> Result<DropSequenceReply> {
         Ok(self.ctx.meta.drop_sequence(req).await?)
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        Ok(self.ctx.meta.create_dictionary(req).await?)
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        Ok(self.ctx.meta.update_dictionary(req).await?)
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        let reply = self.ctx.meta.drop_dictionary(dict_ident.clone()).await?;
+        Ok(reply)
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        let reply = self.ctx.meta.get_dictionary(req.clone()).await?;
+        Ok(reply)
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        Ok(self.ctx.meta.list_dictionaries(req).await?)
+    }
 }
diff --git a/src/query/service/src/catalogs/share/share_catalog.rs b/src/query/service/src/catalogs/share/share_catalog.rs
index b0ab102efc5e..2416b2b60639 100644
--- a/src/query/service/src/catalogs/share/share_catalog.rs
+++ b/src/query/service/src/catalogs/share/share_catalog.rs
@@ -30,12 +30,15 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_api::ShareApi;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CatalogOption;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -52,6 +55,7 @@ use databend_common_meta_app::schema::DatabaseInfo;
 use databend_common_meta_app::schema::DatabaseMeta;
 use databend_common_meta_app::schema::DatabaseType;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -65,6 +69,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -74,6 +79,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -97,6 +103,8 @@ use databend_common_meta_app::schema::UndropDatabaseReply;
 use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateVirtualColumnReply;
@@ -740,4 +748,39 @@ impl Catalog for ShareCatalog {
     async fn drop_sequence(&self, _req: DropSequenceReq) -> Result<DropSequenceReply> {
         unimplemented!()
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        _req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        unimplemented!()
+    }
 }
diff --git a/src/query/service/src/interpreters/access/privilege_access.rs b/src/query/service/src/interpreters/access/privilege_access.rs
index cbaa6b92defc..11d6725b5745 100644
--- a/src/query/service/src/interpreters/access/privilege_access.rs
+++ b/src/query/service/src/interpreters/access/privilege_access.rs
@@ -999,6 +999,13 @@ impl AccessChecker for PrivilegeAccess {
             Plan::AnalyzeTable(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Super, false, false).await?
             }
+            // Dictionary
+            Plan::ShowCreateDictionary(_)
+            | Plan::CreateDictionary(_)
+            | Plan::DropDictionary(_) => {
+                self.validate_access(&GrantObject::Global, UserPrivilegeType::Super, false, false)
+                    .await?;
+            }
             // Others.
             Plan::Insert(plan) => {
                 let target_table_privileges = if plan.overwrite {
diff --git a/src/query/service/src/interpreters/interpreter_dictionary_create.rs b/src/query/service/src/interpreters/interpreter_dictionary_create.rs
new file mode 100644
index 000000000000..881cf3d42906
--- /dev/null
+++ b/src/query/service/src/interpreters/interpreter_dictionary_create.rs
@@ -0,0 +1,94 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_ast::ast::CreateOption;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
+use databend_common_meta_app::schema::CreateDictionaryReq;
+use databend_common_meta_app::schema::DictionaryIdentity;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
+use databend_common_sql::plans::CreateDictionaryPlan;
+
+use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+#[derive(Clone, Debug)]
+pub struct CreateDictionaryInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: CreateDictionaryPlan,
+}
+
+impl CreateDictionaryInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: CreateDictionaryPlan) -> Result<Self> {
+        Ok(CreateDictionaryInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for CreateDictionaryInterpreter {
+    fn name(&self) -> &str {
+        "CreateDictionaryInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let tenant = &self.plan.tenant;
+        let catalog = self.ctx.get_catalog(&self.plan.catalog).await?;
+
+        let dictionary_meta = self.plan.meta.clone();
+        let dict_ident =
+            DictionaryIdentity::new(self.plan.database_id, self.plan.dictionary.clone());
+        let dictionary_ident = TenantDictionaryIdent::new(tenant, dict_ident);
+        let req = CreateDictionaryReq {
+            dictionary_ident: dictionary_ident.clone(),
+            dictionary_meta: dictionary_meta.clone(),
+        };
+
+        let reply = catalog.create_dictionary(req).await;
+        if let Err(e) = reply {
+            if e.code() == ErrorCode::DICTIONARY_ALREADY_EXISTS {
+                match self.plan.create_option {
+                    CreateOption::Create => {
+                        return Err(ErrorCode::DictionaryAlreadyExists(format!(
+                            "Dictionary {} already exists",
+                            self.plan.dictionary,
+                        )));
+                    }
+                    CreateOption::CreateIfNotExists => {
+                        return Ok(PipelineBuildResult::create());
+                    }
+                    CreateOption::CreateOrReplace => {
+                        let req = UpdateDictionaryReq {
+                            dictionary_meta: dictionary_meta.clone(),
+                            dictionary_ident: dictionary_ident.clone(),
+                        };
+                        let _reply = catalog.update_dictionary(req).await?;
+                        return Ok(PipelineBuildResult::create());
+                    }
+                }
+            }
+            return Err(e);
+        }
+        Ok(PipelineBuildResult::create())
+    }
+}
diff --git a/src/query/service/src/interpreters/interpreter_dictionary_drop.rs b/src/query/service/src/interpreters/interpreter_dictionary_drop.rs
new file mode 100644
index 000000000000..2e7b426c1ee9
--- /dev/null
+++ b/src/query/service/src/interpreters/interpreter_dictionary_drop.rs
@@ -0,0 +1,70 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
+use databend_common_meta_app::schema::DictionaryIdentity;
+use databend_common_sql::plans::DropDictionaryPlan;
+
+use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+pub struct DropDictionaryInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: DropDictionaryPlan,
+}
+
+impl DropDictionaryInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: DropDictionaryPlan) -> Result<Self> {
+        Ok(DropDictionaryInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for DropDictionaryInterpreter {
+    fn name(&self) -> &str {
+        "DropDictionaryInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let catalog_name = self.plan.catalog.as_str();
+        let tenant = self.ctx.get_tenant();
+        let db_id = self.plan.database_id;
+        let dict_name = self.plan.dictionary.as_str();
+        let catalog = self.ctx.get_catalog(catalog_name).await?;
+        let dict_ident = TenantDictionaryIdent::new(
+            tenant,
+            DictionaryIdentity::new(db_id, dict_name.to_string()),
+        );
+        let reply = catalog.drop_dictionary(dict_ident.clone()).await?;
+        if self.plan.if_exists || reply.is_some() {
+            return Ok(PipelineBuildResult::create());
+        } else {
+            return Err(ErrorCode::UnknownDictionary(format!(
+                "Unknown dictionary {}",
+                dict_name,
+            )));
+        }
+    }
+}
diff --git a/src/query/service/src/interpreters/interpreter_dictionary_show_create.rs b/src/query/service/src/interpreters/interpreter_dictionary_show_create.rs
new file mode 100644
index 000000000000..7ac7957285cb
--- /dev/null
+++ b/src/query/service/src/interpreters/interpreter_dictionary_show_create.rs
@@ -0,0 +1,194 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_ast::ast::quote::display_ident;
+use databend_common_ast::parser::Dialect;
+use databend_common_catalog::catalog::Catalog;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_expression::types::DataType;
+use databend_common_expression::BlockEntry;
+use databend_common_expression::DataBlock;
+use databend_common_expression::Scalar;
+use databend_common_expression::Value;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
+use databend_common_meta_app::schema::DictionaryIdentity;
+use databend_common_meta_app::schema::DictionaryMeta;
+use databend_common_sql::plans::ShowCreateDictionaryPlan;
+
+use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+pub struct ShowCreateDictionaryInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: ShowCreateDictionaryPlan,
+}
+
+pub struct ShowCreateQuerySettings {
+    pub sql_dialect: Dialect,
+    pub quoted_ident_case_sensitive: bool,
+}
+
+impl ShowCreateDictionaryInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: ShowCreateDictionaryPlan) -> Result<Self> {
+        Ok(ShowCreateDictionaryInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for ShowCreateDictionaryInterpreter {
+    fn name(&self) -> &str {
+        "ShowCreateDictionaryInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let tenant = self.ctx.get_tenant();
+        let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
+        let dict_name = self.plan.dictionary.clone();
+
+        let dict_ident = TenantDictionaryIdent::new(
+            tenant,
+            DictionaryIdentity::new(self.plan.database_id, dict_name.clone()),
+        );
+
+        let dictionary = if let Some(reply) = catalog.get_dictionary(dict_ident).await? {
+            reply.dictionary_meta
+        } else {
+            return Err(ErrorCode::UnknownDictionary(format!(
+                "Unknown dictionary {}",
+                dict_name.clone(),
+            )));
+        };
+        let settings = self.ctx.get_settings();
+        let settings = ShowCreateQuerySettings {
+            sql_dialect: settings.get_sql_dialect()?,
+            quoted_ident_case_sensitive: settings.get_quoted_ident_case_sensitive()?,
+        };
+
+        let create_query: String =
+            Self::show_create_query(catalog.as_ref(), &dictionary, &dict_name, &settings).await?;
+        let block = DataBlock::new(
+            vec![
+                BlockEntry::new(DataType::String, Value::Scalar(Scalar::String(dict_name))),
+                BlockEntry::new(
+                    DataType::String,
+                    Value::Scalar(Scalar::String(create_query)),
+                ),
+            ],
+            1,
+        );
+        PipelineBuildResult::from_blocks(vec![block])
+    }
+}
+
+impl ShowCreateDictionaryInterpreter {
+    pub async fn show_create_query(
+        _catalog: &dyn Catalog,
+        dictionary: &DictionaryMeta,
+        dict_name: &str,
+        settings: &ShowCreateQuerySettings,
+    ) -> Result<String> {
+        let sql_dialect = settings.sql_dialect;
+        let quoted_ident_case_sensitive = settings.quoted_ident_case_sensitive;
+        let schema = dictionary.schema.clone();
+        let source = dictionary.source.clone();
+        let source_options = dictionary.options.clone();
+        let comment = dictionary.comment.clone();
+        let pk_id_list = dictionary.primary_column_ids.clone();
+        let field_comments = dictionary.field_comments.clone();
+
+        let mut dict_create_sql = format!(
+            "CREATE DICTIONARY {}\n(\n",
+            display_ident(dict_name, quoted_ident_case_sensitive, sql_dialect)
+        );
+
+        // Append columns and indexes.
+        {
+            let mut create_defs = vec![];
+            for field in schema.fields().iter() {
+                let nullable = if field.is_nullable() {
+                    " NULL".to_string()
+                } else {
+                    " NOT NULL".to_string()
+                };
+                // compatibility: creating table in the old planner will not have `fields_comments`
+                let comment = field_comments
+                    .get(&field.column_id)
+                    .and_then(|c| format!(" COMMENT '{}'", c).into())
+                    .unwrap_or_default();
+                let column_str = format!(
+                    "  {} {}{}{}",
+                    display_ident(field.name(), quoted_ident_case_sensitive, sql_dialect),
+                    field.data_type().remove_recursive_nullable().sql_name(),
+                    nullable,
+                    comment
+                );
+
+                create_defs.push(column_str);
+            }
+
+            let create_defs_str = format!("{}\n", create_defs.join(",\n"));
+            dict_create_sql.push_str(&create_defs_str);
+        }
+        // Append primary keys.
+        {
+            dict_create_sql.push_str(")\nPRIMARY KEY ");
+            let mut primary_names = Vec::new();
+            for pk_id in pk_id_list {
+                let field = schema.field_of_column_id(pk_id)?;
+                primary_names.push(field.name());
+            }
+            let res: String = primary_names
+                .iter()
+                .map(|s| s.as_ref())
+                .collect::<Vec<&str>>()
+                .join(",");
+            dict_create_sql.push_str(&res);
+            dict_create_sql.push('\n');
+        }
+        // Append source options.
+        {
+            dict_create_sql.push_str(&format!("SOURCE({}", source));
+            dict_create_sql.push('(');
+            let mut show_options = Vec::new();
+            for (key, value) in source_options {
+                if key == "password" {
+                    show_options.push(format!("{}='[HIDDEN]'", key));
+                } else {
+                    show_options.push(format!("{}='{}'", key, value));
+                }
+            }
+            let res: String = show_options.join(" ");
+            dict_create_sql.push_str(&res);
+            dict_create_sql.push_str("))\n");
+        }
+        // Append comment.
+        {
+            if !comment.is_empty() {
+                dict_create_sql.push_str("COMMENT ");
+                dict_create_sql.push_str(&format!("'{}'", comment));
+            }
+        }
+        Ok(dict_create_sql)
+    }
+}
diff --git a/src/query/service/src/interpreters/interpreter_factory.rs b/src/query/service/src/interpreters/interpreter_factory.rs
index b95c065a2832..7ed7f0fc1248 100644
--- a/src/query/service/src/interpreters/interpreter_factory.rs
+++ b/src/query/service/src/interpreters/interpreter_factory.rs
@@ -23,6 +23,9 @@ use log::error;
 
 use super::interpreter_catalog_create::CreateCatalogInterpreter;
 use super::interpreter_catalog_show_create::ShowCreateCatalogInterpreter;
+use super::interpreter_dictionary_create::CreateDictionaryInterpreter;
+use super::interpreter_dictionary_drop::DropDictionaryInterpreter;
+use super::interpreter_dictionary_show_create::ShowCreateDictionaryInterpreter;
 use super::interpreter_index_create::CreateIndexInterpreter;
 use super::interpreter_index_drop::DropIndexInterpreter;
 use super::interpreter_mutation::MutationInterpreter;
@@ -613,6 +616,17 @@ impl InterpreterFactory {
                 ctx,
                 *p.clone(),
             )?)),
+            // Dictionary
+            Plan::CreateDictionary(create_dictionary) => Ok(Arc::new(
+                CreateDictionaryInterpreter::try_create(ctx, *create_dictionary.clone())?,
+            )),
+            Plan::ShowCreateDictionary(show_create_table) => Ok(Arc::new(
+                ShowCreateDictionaryInterpreter::try_create(ctx, *show_create_table.clone())?,
+            )),
+            Plan::DropDictionary(drop_dict) => Ok(Arc::new(DropDictionaryInterpreter::try_create(
+                ctx,
+                *drop_dict.clone(),
+            )?)),
         }
     }
 }
diff --git a/src/query/service/src/interpreters/mod.rs b/src/query/service/src/interpreters/mod.rs
index 4b5e1854dda3..b1fff3e7a59d 100644
--- a/src/query/service/src/interpreters/mod.rs
+++ b/src/query/service/src/interpreters/mod.rs
@@ -36,6 +36,9 @@ mod interpreter_database_drop;
 mod interpreter_database_rename;
 mod interpreter_database_show_create;
 mod interpreter_database_undrop;
+mod interpreter_dictionary_create;
+mod interpreter_dictionary_drop;
+mod interpreter_dictionary_show_create;
 mod interpreter_execute_immediate;
 mod interpreter_explain;
 mod interpreter_factory;
diff --git a/src/query/service/tests/it/sql/exec/get_table_bind_test.rs b/src/query/service/tests/it/sql/exec/get_table_bind_test.rs
index 759f6918261d..0697ab63bcbb 100644
--- a/src/query/service/tests/it/sql/exec/get_table_bind_test.rs
+++ b/src/query/service/tests/it/sql/exec/get_table_bind_test.rs
@@ -54,11 +54,14 @@ use databend_common_meta_app::principal::RoleInfo;
 use databend_common_meta_app::principal::UserDefinedConnection;
 use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeType;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -72,6 +75,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -85,6 +89,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -94,6 +99,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -115,6 +121,8 @@ use databend_common_meta_app::schema::UndropDatabaseReply;
 use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateVirtualColumnReply;
@@ -415,6 +423,35 @@ impl Catalog for FakedCatalog {
     async fn get_table_meta_by_id(&self, table_id: MetaId) -> Result<Option<SeqV<TableMeta>>> {
         self.cat.get_table_meta_by_id(table_id).await
     }
+
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        todo!()
+    }
+
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        todo!()
+    }
+
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        todo!()
+    }
+
+    async fn get_dictionary(
+        &self,
+        req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        self.cat.get_dictionary(req).await
+    }
+
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        todo!()
+    }
 }
 
 struct CtxDelegation {
diff --git a/src/query/service/tests/it/storages/fuse/operations/commit.rs b/src/query/service/tests/it/storages/fuse/operations/commit.rs
index e138d5755c9c..83473a6ee70d 100644
--- a/src/query/service/tests/it/storages/fuse/operations/commit.rs
+++ b/src/query/service/tests/it/storages/fuse/operations/commit.rs
@@ -53,11 +53,14 @@ use databend_common_meta_app::principal::RoleInfo;
 use databend_common_meta_app::principal::UserDefinedConnection;
 use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeType;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -71,6 +74,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -84,6 +88,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -93,6 +98,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -114,6 +120,8 @@ use databend_common_meta_app::schema::UndropDatabaseReply;
 use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
@@ -1142,4 +1150,33 @@ impl Catalog for FakedCatalog {
             self.cat.retryable_update_multi_table_meta(req).await
         }
     }
+
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        todo!()
+    }
+
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        todo!()
+    }
+
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        todo!()
+    }
+
+    async fn get_dictionary(
+        &self,
+        _req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        todo!()
+    }
+
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        todo!()
+    }
 }
diff --git a/src/query/sql/src/planner/binder/binder.rs b/src/query/sql/src/planner/binder/binder.rs
index c72fcf8b17b4..e1c2ede2264f 100644
--- a/src/query/sql/src/planner/binder/binder.rs
+++ b/src/query/sql/src/planner/binder/binder.rs
@@ -271,9 +271,9 @@ impl<'a> Binder {
             Statement::AnalyzeTable(stmt) => self.bind_analyze_table(stmt).await?,
             Statement::ExistsTable(stmt) => self.bind_exists_table(stmt).await?,
             // Dictionaries
-            Statement::CreateDictionary(_stmt) => todo!(),
-            Statement::DropDictionary(_stmt) => todo!(),
-            Statement::ShowCreateDictionary(_stmt) => todo!(),
+            Statement::CreateDictionary(stmt) => self.bind_create_dictionary(stmt).await?,
+            Statement::DropDictionary(stmt) => self.bind_drop_dictionary(stmt).await?,
+            Statement::ShowCreateDictionary(stmt) => self.bind_show_create_dictionary(stmt).await?,
             Statement::ShowDictionaries { show_options: _ } => todo!(),
             // Views
             Statement::CreateView(stmt) => self.bind_create_view(stmt).await?,
diff --git a/src/query/sql/src/planner/binder/ddl/dictionary.rs b/src/query/sql/src/planner/binder/ddl/dictionary.rs
new file mode 100644
index 000000000000..c676c2afb25a
--- /dev/null
+++ b/src/query/sql/src/planner/binder/ddl/dictionary.rs
@@ -0,0 +1,202 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::BTreeMap;
+
+use databend_common_ast::ast::CreateDictionaryStmt;
+use databend_common_ast::ast::DropDictionaryStmt;
+use databend_common_ast::ast::ShowCreateDictionaryStmt;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_expression::types::DataType;
+use databend_common_expression::DataField;
+use databend_common_expression::DataSchemaRefExt;
+use databend_common_meta_app::schema::DictionaryMeta;
+
+use crate::plans::CreateDictionaryPlan;
+use crate::plans::DropDictionaryPlan;
+use crate::plans::Plan;
+use crate::plans::ShowCreateDictionaryPlan;
+use crate::Binder;
+
+impl Binder {
+    #[async_backtrace::framed]
+    pub(in crate::planner::binder) async fn bind_create_dictionary(
+        &mut self,
+        stmt: &CreateDictionaryStmt,
+    ) -> Result<Plan> {
+        let CreateDictionaryStmt {
+            create_option,
+            catalog,
+            database,
+            dictionary_name,
+            columns,
+            primary_keys,
+            source_name,
+            source_options,
+            comment,
+        } = stmt;
+
+        let tenant = self.ctx.get_tenant();
+        let (catalog, database, dictionary_name) =
+            self.normalize_object_identifier_triple(catalog, database, dictionary_name);
+
+        let database_id;
+        {
+            let catalog = self.ctx.get_catalog(&catalog).await?;
+            let db = catalog.get_database(&tenant, &database).await?;
+            database_id = db.get_db_info().database_id.db_id;
+        }
+
+        let source = self.normalize_object_identifier(source_name);
+
+        if source.to_lowercase() != *"mysql" {
+            return Err(ErrorCode::UnsupportedDictionarySource(format!(
+                "The specified source '{}' is not currently supported.",
+                source.to_lowercase(),
+            )));
+        }
+        // TODO: Authentication to connect to MySQL database will be implemented later
+
+        let options: BTreeMap<String, String> = source_options
+            .iter()
+            .map(|(k, v)| (k.to_lowercase(), v.to_string().to_lowercase()))
+            .collect();
+        let required_options = ["host", "port", "username", "password", "db"];
+        for option in required_options {
+            if !options.contains_key(option) {
+                return Err(ErrorCode::MissingDictionaryOption(
+                    "The configuration is missing one or more required options. ".to_owned()
+                        + "Please ensure you have provided values for 'host', 'port', 'username', 'password', and 'db'.",
+                ));
+            }
+        }
+        if required_options.len() != options.len() {
+            return Err(ErrorCode::UnsupportedDictionaryOption(format!(
+                "The provided options are not recognized."
+            )));
+        }
+
+        let mut field_comments = BTreeMap::new();
+        let mut primary_column_ids = Vec::new();
+
+        let (schema, _) = self.analyze_create_table_schema_by_columns(columns).await?;
+        for table_field in schema.fields() {
+            if table_field.default_expr.is_some() || table_field.computed_expr.is_some() {
+                return Err(ErrorCode::WrongDictionaryFieldExpr(
+                    "The table field configuration is invalid. ".to_owned()
+                        + "Default expressions and computed expressions for the table fields should not be set.",
+                ));
+            }
+        }
+        for column in columns {
+            if column.comment.is_some() {
+                let column_id = schema.column_id_of(column.name.name.as_str())?;
+                field_comments.insert(column_id, column.comment.clone().unwrap_or_default());
+            }
+        }
+        for primary_key in primary_keys {
+            let pk_id = schema.column_id_of(primary_key.name.as_str())?;
+            primary_column_ids.push(pk_id);
+        }
+
+        let comment = comment.clone().unwrap_or("".to_string());
+        let meta = DictionaryMeta {
+            source,
+            options,
+            schema,
+            field_comments,
+            primary_column_ids,
+            comment,
+            ..Default::default()
+        };
+
+        Ok(Plan::CreateDictionary(Box::new(CreateDictionaryPlan {
+            create_option: create_option.clone(),
+            tenant,
+            catalog,
+            database_id,
+            dictionary: dictionary_name,
+            meta,
+        })))
+    }
+
+    #[async_backtrace::framed]
+    pub(in crate::planner::binder) async fn bind_drop_dictionary(
+        &mut self,
+        stmt: &DropDictionaryStmt,
+    ) -> Result<Plan> {
+        let DropDictionaryStmt {
+            if_exists,
+            catalog,
+            database,
+            dictionary_name,
+        } = stmt;
+
+        let tenant = self.ctx.get_tenant();
+        let (catalog, database, dictionary_name) =
+            self.normalize_object_identifier_triple(catalog, database, dictionary_name);
+
+        let database_id;
+        {
+            let catalog = self.ctx.get_catalog(&catalog).await?;
+            let db = catalog.get_database(&tenant, &database).await?;
+            database_id = db.get_db_info().database_id.db_id;
+        }
+        Ok(Plan::DropDictionary(Box::new(DropDictionaryPlan {
+            if_exists: *if_exists,
+            tenant,
+            catalog,
+            database_id,
+            dictionary: dictionary_name,
+        })))
+    }
+
+    #[async_backtrace::framed]
+    pub(in crate::planner::binder) async fn bind_show_create_dictionary(
+        &mut self,
+        stmt: &ShowCreateDictionaryStmt,
+    ) -> Result<Plan> {
+        let ShowCreateDictionaryStmt {
+            catalog,
+            database,
+            dictionary_name,
+        } = stmt;
+
+        let (catalog, database, dictionary_name) =
+            self.normalize_object_identifier_triple(catalog, database, dictionary_name);
+
+        let schema = DataSchemaRefExt::create(vec![
+            DataField::new("Dictionary", DataType::String),
+            DataField::new("Create Dictionary", DataType::String),
+        ]);
+
+        let database_id;
+        {
+            let tenant = self.ctx.get_tenant();
+            let catalog = self.ctx.get_catalog(&catalog).await?;
+            let db = catalog.get_database(&tenant, &database).await?;
+            database_id = db.get_db_info().database_id.db_id;
+        }
+
+        Ok(Plan::ShowCreateDictionary(Box::new(
+            ShowCreateDictionaryPlan {
+                catalog,
+                database_id,
+                dictionary: dictionary_name,
+                schema,
+            },
+        )))
+    }
+}
diff --git a/src/query/sql/src/planner/binder/ddl/mod.rs b/src/query/sql/src/planner/binder/ddl/mod.rs
index a344e82f6b66..56665f90f7dc 100644
--- a/src/query/sql/src/planner/binder/ddl/mod.rs
+++ b/src/query/sql/src/planner/binder/ddl/mod.rs
@@ -18,6 +18,7 @@ mod column;
 mod connection;
 mod data_mask;
 mod database;
+mod dictionary;
 mod dynamic_table;
 mod index;
 mod network_policy;
diff --git a/src/query/sql/src/planner/binder/ddl/table.rs b/src/query/sql/src/planner/binder/ddl/table.rs
index 900a24834c26..b34e932fe1c6 100644
--- a/src/query/sql/src/planner/binder/ddl/table.rs
+++ b/src/query/sql/src/planner/binder/ddl/table.rs
@@ -1386,7 +1386,7 @@ impl Binder {
     }
 
     #[async_backtrace::framed]
-    async fn analyze_create_table_schema_by_columns(
+    pub async fn analyze_create_table_schema_by_columns(
         &self,
         columns: &[ColumnDefinition],
     ) -> Result<(TableSchemaRef, Vec<String>)> {
diff --git a/src/query/sql/src/planner/format/display_plan.rs b/src/query/sql/src/planner/format/display_plan.rs
index e7097cb84bf4..f034eb5bd8d4 100644
--- a/src/query/sql/src/planner/format/display_plan.rs
+++ b/src/query/sql/src/planner/format/display_plan.rs
@@ -211,6 +211,11 @@ impl Plan {
 
             Plan::SetPriority(_) => Ok("SetPriority".to_string()),
             Plan::System(_) => Ok("System".to_string()),
+
+            // Dictionary
+            Plan::CreateDictionary(_) => Ok("CreateDictionary".to_string()),
+            Plan::DropDictionary(_) => Ok("DropDictionary".to_string()),
+            Plan::ShowCreateDictionary(_) => Ok("ShowCreateDictionary".to_string()),
         }
     }
 }
diff --git a/src/query/sql/src/planner/plans/ddl/dictionary.rs b/src/query/sql/src/planner/plans/ddl/dictionary.rs
new file mode 100644
index 000000000000..845d66a6155a
--- /dev/null
+++ b/src/query/sql/src/planner/plans/ddl/dictionary.rs
@@ -0,0 +1,51 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use databend_common_ast::ast::CreateOption;
+use databend_common_expression::DataSchemaRef;
+use databend_common_meta_app::schema::DictionaryMeta;
+use databend_common_meta_app::tenant::Tenant;
+
+#[derive(Clone, Debug)]
+pub struct CreateDictionaryPlan {
+    pub create_option: CreateOption,
+    pub tenant: Tenant,
+    pub catalog: String,
+    pub database_id: u64,
+    pub dictionary: String,
+    pub meta: DictionaryMeta,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct DropDictionaryPlan {
+    pub if_exists: bool,
+    pub tenant: Tenant,
+    pub catalog: String,
+    pub database_id: u64,
+    pub dictionary: String,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ShowCreateDictionaryPlan {
+    pub catalog: String,
+    pub database_id: u64,
+    pub dictionary: String,
+    pub schema: DataSchemaRef,
+}
+
+impl ShowCreateDictionaryPlan {
+    pub fn schema(&self) -> DataSchemaRef {
+        self.schema.clone()
+    }
+}
diff --git a/src/query/sql/src/planner/plans/ddl/mod.rs b/src/query/sql/src/planner/plans/ddl/mod.rs
index 1d03ba15062b..fe0b14ce21ad 100644
--- a/src/query/sql/src/planner/plans/ddl/mod.rs
+++ b/src/query/sql/src/planner/plans/ddl/mod.rs
@@ -16,6 +16,7 @@ mod account;
 mod catalog;
 mod connection;
 mod database;
+mod dictionary;
 mod dynamic_table;
 mod file_format;
 mod index;
@@ -34,6 +35,7 @@ pub use account::*;
 pub use catalog::*;
 pub use connection::*;
 pub use database::*;
+pub use dictionary::*;
 pub use dynamic_table::*;
 pub use file_format::*;
 pub use index::*;
diff --git a/src/query/sql/src/planner/plans/plan.rs b/src/query/sql/src/planner/plans/plan.rs
index 7b4f277c7f45..c633c1a08d28 100644
--- a/src/query/sql/src/planner/plans/plan.rs
+++ b/src/query/sql/src/planner/plans/plan.rs
@@ -24,6 +24,9 @@ use databend_common_expression::DataSchema;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
 
+use super::CreateDictionaryPlan;
+use super::DropDictionaryPlan;
+use super::ShowCreateDictionaryPlan;
 use crate::binder::ExplainConfig;
 use crate::optimizer::SExpr;
 use crate::plans::copy_into_location::CopyIntoLocationPlan;
@@ -377,6 +380,11 @@ pub enum Plan {
     // sequence
     CreateSequence(Box<CreateSequencePlan>),
     DropSequence(Box<DropSequencePlan>),
+
+    // Dictionary
+    CreateDictionary(Box<CreateDictionaryPlan>),
+    DropDictionary(Box<DropDictionaryPlan>),
+    ShowCreateDictionary(Box<ShowCreateDictionaryPlan>),
 }
 
 #[derive(Clone, Debug)]
@@ -460,6 +468,7 @@ impl Plan {
             Plan::DataMutation { schema, .. } => schema.clone(),
             Plan::ShowCreateCatalog(plan) => plan.schema(),
             Plan::ShowCreateDatabase(plan) => plan.schema(),
+            Plan::ShowCreateDictionary(plan) => plan.schema(),
             Plan::ShowCreateTable(plan) => plan.schema(),
             Plan::DescribeTable(plan) => plan.schema(),
             Plan::VacuumTable(plan) => plan.schema(),
diff --git a/src/query/storages/hive/hive/src/hive_catalog.rs b/src/query/storages/hive/hive/src/hive_catalog.rs
index a2ebfcb1e5f7..78dde7b33a13 100644
--- a/src/query/storages/hive/hive/src/hive_catalog.rs
+++ b/src/query/storages/hive/hive/src/hive_catalog.rs
@@ -28,12 +28,15 @@ use databend_common_catalog::table_function::TableFunction;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CatalogOption;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -47,6 +50,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -60,6 +64,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -69,6 +74,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -90,6 +96,8 @@ use databend_common_meta_app::schema::UndropDatabaseReply;
 use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateVirtualColumnReply;
@@ -701,4 +709,39 @@ impl Catalog for HiveCatalog {
     async fn drop_sequence(&self, _req: DropSequenceReq) -> Result<DropSequenceReply> {
         unimplemented!()
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        _req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        unimplemented!()
+    }
 }
diff --git a/src/query/storages/iceberg/src/catalog.rs b/src/query/storages/iceberg/src/catalog.rs
index 1394c71acc3c..338d316de089 100644
--- a/src/query/storages/iceberg/src/catalog.rs
+++ b/src/query/storages/iceberg/src/catalog.rs
@@ -26,12 +26,15 @@ use databend_common_catalog::table_function::TableFunction;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::tenant_dictionary_ident::TenantDictionaryIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CatalogOption;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
 use databend_common_meta_app::schema::CreateDatabaseReply;
 use databend_common_meta_app::schema::CreateDatabaseReq;
+use databend_common_meta_app::schema::CreateDictionaryReply;
+use databend_common_meta_app::schema::CreateDictionaryReq;
 use databend_common_meta_app::schema::CreateIndexReply;
 use databend_common_meta_app::schema::CreateIndexReq;
 use databend_common_meta_app::schema::CreateLockRevReply;
@@ -45,6 +48,7 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::CreateVirtualColumnReply;
 use databend_common_meta_app::schema::CreateVirtualColumnReq;
 use databend_common_meta_app::schema::DeleteLockRevReq;
+use databend_common_meta_app::schema::DictionaryMeta;
 use databend_common_meta_app::schema::DropDatabaseReply;
 use databend_common_meta_app::schema::DropDatabaseReq;
 use databend_common_meta_app::schema::DropIndexReply;
@@ -58,6 +62,7 @@ use databend_common_meta_app::schema::DropTableReply;
 use databend_common_meta_app::schema::DropVirtualColumnReply;
 use databend_common_meta_app::schema::DropVirtualColumnReq;
 use databend_common_meta_app::schema::ExtendLockRevReq;
+use databend_common_meta_app::schema::GetDictionaryReply;
 use databend_common_meta_app::schema::GetIndexReply;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::GetSequenceNextValueReply;
@@ -68,6 +73,7 @@ use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IcebergCatalogOption;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListIndexesReq;
 use databend_common_meta_app::schema::ListLockRevReq;
@@ -89,6 +95,8 @@ use databend_common_meta_app::schema::UndropDatabaseReply;
 use databend_common_meta_app::schema::UndropDatabaseReq;
 use databend_common_meta_app::schema::UndropTableReply;
 use databend_common_meta_app::schema::UndropTableReq;
+use databend_common_meta_app::schema::UpdateDictionaryReply;
+use databend_common_meta_app::schema::UpdateDictionaryReq;
 use databend_common_meta_app::schema::UpdateIndexReply;
 use databend_common_meta_app::schema::UpdateIndexReq;
 use databend_common_meta_app::schema::UpdateVirtualColumnReply;
@@ -544,4 +552,39 @@ impl Catalog for IcebergCatalog {
     async fn drop_sequence(&self, _req: DropSequenceReq) -> Result<DropSequenceReply> {
         unimplemented!()
     }
+
+    /// Dictionary
+    #[async_backtrace::framed]
+    async fn create_dictionary(&self, _req: CreateDictionaryReq) -> Result<CreateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn update_dictionary(&self, _req: UpdateDictionaryReq) -> Result<UpdateDictionaryReply> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn drop_dictionary(
+        &self,
+        _dict_ident: TenantDictionaryIdent,
+    ) -> Result<Option<SeqV<DictionaryMeta>>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn get_dictionary(
+        &self,
+        _req: TenantDictionaryIdent,
+    ) -> Result<Option<GetDictionaryReply>> {
+        unimplemented!()
+    }
+
+    #[async_backtrace::framed]
+    async fn list_dictionaries(
+        &self,
+        _req: ListDictionaryReq,
+    ) -> Result<Vec<(String, DictionaryMeta)>> {
+        unimplemented!()
+    }
 }
diff --git a/tests/sqllogictests/src/client/http_client.rs b/tests/sqllogictests/src/client/http_client.rs
index 2eebf2d6b0a6..25fadca0efda 100644
--- a/tests/sqllogictests/src/client/http_client.rs
+++ b/tests/sqllogictests/src/client/http_client.rs
@@ -45,7 +45,7 @@ struct QueryResponse {
 // make error message the same with ErrorCode::display
 fn format_error(value: serde_json::Value) -> String {
     let value = value.as_object().unwrap();
-    let detail = value["detail"].as_str();
+    let detail = value.get("detail").map(|v| v.as_str()).flatten();
     let code = value["code"].as_u64().unwrap();
     let message = value["message"].as_str().unwrap();
     if let Some(detail) = detail {
diff --git a/tests/sqllogictests/suites/base/05_ddl/05_0037_ddl_dictionary.test b/tests/sqllogictests/suites/base/05_ddl/05_0037_ddl_dictionary.test
new file mode 100644
index 000000000000..d41c3dd58bbe
--- /dev/null
+++ b/tests/sqllogictests/suites/base/05_ddl/05_0037_ddl_dictionary.test
@@ -0,0 +1,84 @@
+statement ok
+DROP DICTIONARY IF EXISTS d
+
+statement ok
+DROP DICTIONARY IF EXISTS d2
+
+statement ok
+DROP DICTIONARY IF EXISTS d3
+
+statement ok
+DROP DICTIONARY IF EXISTS d4
+
+statement error 3117
+CREATE DICTIONARY d(c1 int, c2 Varchar) PRIMARY KEY c1 SOURCE(postgresql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement ok
+CREATE DICTIONARY d(c1 int, c2 Varchar) PRIMARY KEY c1 SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement ok
+CREATE DICTIONARY IF NOT EXISTS d(c1 int, c2 Varchar) PRIMARY KEY c1 SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement error 3113
+CREATE DICTIONARY d(c1 int, c2 Varchar) PRIMARY KEY c1 SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement ok
+CREATE DICTIONARY d2(a int, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement error 3113
+CREATE DICTIONARY d2(a int, b int) PRIMARY KEY b SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement error 3118
+create dictionary d3(`a` int, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234'))
+
+statement ok
+create dictionary d3(`a` int, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement ok
+create or replace dictionary d3(a int, b Varchar) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1')) comment 'comment'
+
+statement error 3116
+create dictionary d4(a int, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1' name='dummy'))
+
+statement ok
+create or replace dictionary d4(a Varchar, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+query TT
+show create dictionary d
+----
+d CREATE DICTIONARY d ( c1 INT NULL, c2 VARCHAR NULL ) PRIMARY KEY c1 SOURCE(mysql(db='db1' host='localhost' password='[HIDDEN]' port='3306' username='root'))
+
+query TT
+show create dictionary d3
+----
+d3 CREATE DICTIONARY d3 ( a INT NULL, b VARCHAR NULL ) PRIMARY KEY a SOURCE(mysql(db='db1' host='localhost' password='[HIDDEN]' port='3306' username='root')) COMMENT 'comment'
+
+statement error 3114
+show create dictionary test
+
+statement ok
+DROP DICTIONARY IF EXISTS d
+
+statement ok
+DROP DICTIONARY IF EXISTS d2
+
+statement ok
+DROP DICTIONARY IF EXISTS d3
+
+statement ok
+DROP DICTIONARY IF EXISTS d4
+
+statement error 3114
+drop dictionary test
+
+statement ok
+DROP DATABASE IF EXISTS db1
+
+statement ok
+CREATE DATABASE db1
+
+statement ok
+CREATE DICTIONARY db1.test1(a int, b int) PRIMARY KEY a SOURCE(mysql(host='localhost' port='3306' username='root' password='1234' db='db1'))
+
+statement ok
+DROP DATABASE db1