diff --git a/pkg/src/common/cm_key_word.h b/pkg/src/common/cm_key_word.h
index a0b8a52428c55e4ab03e0a947dd1c45c3004c8fd..909fbe317021c092e7de5a8173005edde994800d 100644
--- a/pkg/src/common/cm_key_word.h
+++ b/pkg/src/common/cm_key_word.h
@@ -126,6 +126,7 @@ typedef enum en_key_wid {
     KEY_WORD_DATABASE,
     KEY_WORD_DATAFILE,
     KEY_WORD_DB,   /* for mysql db add datafile */
+    KEY_WORD_DDL_ENABLED,
     KEY_WORD_DEBUG,
     KEY_WORD_DECLARE,
     KEY_WORD_DEFERRABLE, /* for constraint state */
diff --git a/pkg/src/ctc/ctc_ddl.c b/pkg/src/ctc/ctc_ddl.c
index e8be4d4305992d3f1f26c15a09e26e134fa2eb02..a101eec58af572c6cc1162f8f7ed335f426f96ca 100644
--- a/pkg/src/ctc/ctc_ddl.c
+++ b/pkg/src/ctc/ctc_ddl.c
@@ -61,6 +61,9 @@
 
 #define META_SEARCH_TIMES 3
 #define META_SEARCH_WAITING_TIME_IN_MS 10000
+#define MDL_NAMESPACE_FOR_BACKUP 1
+#define THREAD_ID_FOR_BACKUP CT_INVALID_ID32 - 2
+#define INST_ID_FOR_BACKUP CT_INVALID_ID32 - 2
 
 #define USER_LOCK_MAX_WAIT 10000
 
@@ -509,6 +512,79 @@ EXTER_ATTACK int ctc_lock_table(ctc_handler_t *tch, const char *db_name, ctc_loc
     return CT_SUCCESS;
 }
 
+status_t ctc_alter_ddl_enabled_for_backup(knl_handle_t knl_session, bool32 is_enabled)
+{
+    ctc_handler_t tch;
+    status_t ret;
+    ret = memset_s(&tch, sizeof(ctc_handler_t), 0, sizeof(ctc_handler_t));
+    knl_securec_check(ret);
+    // Do not use 0 or 0xFFFFFFFF, or mysql will close all the connections.
+    tch.thd_id = THREAD_ID_FOR_BACKUP;
+    tch.inst_id = INST_ID_FOR_BACKUP;
+    tch.sess_addr = (uint64_t)knl_session;
+
+    ctc_lock_table_info lock_info;
+    ret = memset_s(&lock_info, sizeof(ctc_lock_table_info), 0, sizeof(ctc_lock_table_info));
+    knl_securec_check(ret);
+    lock_info.mdl_namespace = MDL_NAMESPACE_FOR_BACKUP;
+
+    if (!is_enabled) {
+        int error_code = 0;
+        tch.sql_command = SQLCOM_LOCK_INSTANCE;
+        lock_info.sql_type = SQLCOM_LOCK_INSTANCE;
+        return ctc_lock_table(&tch, "", &lock_info, &error_code);
+    } else {
+        tch.sql_command = SQLCOM_UNLOCK_INSTANCE;
+        lock_info.sql_type = SQLCOM_UNLOCK_INSTANCE;
+        return ctc_unlock_table(&tch, tch.inst_id, &lock_info);
+    }
+}
+
+status_t ctc_alter_ddl_enabled_for_backup_ex(knl_handle_t knl_session, bool32 is_enabled)
+{
+    ctc_handler_t tch;
+    status_t ret;
+    ret = memset_s(&tch, sizeof(ctc_handler_t), 0, sizeof(ctc_handler_t));
+    knl_securec_check(ret);
+    // Do not use 0 or 0xFFFFFFFF, or mysql will close all the connections.
+    tch.thd_id = THREAD_ID_FOR_BACKUP;
+    tch.inst_id = INST_ID_FOR_BACKUP;
+    tch.sess_addr = (uint64_t)knl_session;
+
+    ctc_ddl_broadcast_request *broadcast_req = (ctc_ddl_broadcast_request *)cm_malloc(sizeof(ctc_ddl_broadcast_request));
+    if (broadcast_req == NULL) {
+        CT_LOG_RUN_ERR("[ctc_alter_ddl_enabled_for_backup_ex] malloc broadcast_req failed!");
+        return CT_ERROR;
+    }
+
+    ret = memset_s(broadcast_req, sizeof(ctc_ddl_broadcast_request), 0, sizeof(ctc_ddl_broadcast_request));
+    knl_securec_check(ret);
+    broadcast_req->mysql_inst_id =  tch.inst_id ;
+    char *sql_text = "";
+    if (!is_enabled) {
+        sql_text = "lock instance for backup;";
+        tch.sql_command = SQLCOM_LOCK_INSTANCE;
+        broadcast_req->sql_command = SQLCOM_LOCK_INSTANCE;
+    } else {
+        sql_text = "unlock instance;";
+        tch.sql_command = SQLCOM_UNLOCK_INSTANCE;
+        broadcast_req->sql_command = SQLCOM_UNLOCK_INSTANCE;
+    }
+    uint32 sql_len = strlen(sql_text) + 1;
+    memcpy_sp(broadcast_req->sql_str, MAX_DDL_SQL_LEN, sql_text, sql_len);
+    ret = ctc_ddl_execute_and_broadcast(&tch, broadcast_req, false, (knl_session_t *)knl_session);
+    if (ret != CT_SUCCESS) {
+        CT_LOG_RUN_ERR(
+            "[ctc_alter_ddl_enabled_for_backup_ex]:ctc_ddl_execute_and_broadcast failed in alter_ddl_enabled_for_backup_ex. sql_str:%s",
+            broadcast_req->sql_str);
+        cm_free(broadcast_req);
+        return ret;
+    }
+    cm_free(broadcast_req);
+
+    return CT_SUCCESS;
+}
+
 int ctc_broadcast_mysql_dd_invalidate_impl(ctc_handler_t *tch, knl_handle_t knl_session,
                                            ctc_invalidate_broadcast_request *broadcast_req)
 {
diff --git a/pkg/src/ctc/ctc_ddl.h b/pkg/src/ctc/ctc_ddl.h
index 83ae3ead2a2d5a6ee96edcedfe869a63c8d2c8bb..25ab71f30e4e821331e7fc70540c2d1b80fe7e64 100644
--- a/pkg/src/ctc/ctc_ddl.h
+++ b/pkg/src/ctc/ctc_ddl.h
@@ -143,4 +143,6 @@ int ctc_unlock_table_impl(ctc_handler_t *tch, knl_handle_t knl_session, uint32_t
 int ctc_broadcast_mysql_dd_invalidate_impl(ctc_handler_t *tch, knl_handle_t knl_session,
                                            ctc_invalidate_broadcast_request *broadcast_req);
 int ctc_unlock_mdl_key_impl(ctc_handler_t *tch, knl_handle_t knl_session, uint32_t mysql_inst_id);
+status_t ctc_alter_ddl_enabled_for_backup(knl_handle_t knl_session, bool32 is_enabled);
+status_t ctc_alter_ddl_enabled_for_backup_ex(knl_handle_t knl_session, bool32 is_enabled);
 #endif //__CTC_DDL_H__
diff --git a/pkg/src/kernel/common/knl_context.h b/pkg/src/kernel/common/knl_context.h
index c37ad8520c65c251d2dd19fae58612e41eb1ad34..ef1ff410f67a6c82c90ed3137cb98049e153c042 100644
--- a/pkg/src/kernel/common/knl_context.h
+++ b/pkg/src/kernel/common/knl_context.h
@@ -314,6 +314,7 @@ typedef struct st_knl_attr {
     uint32 prevent_snapshot_backup_recycle_redo_timeout;
     bool32 prevent_create_snapshot;
     bool32 ctsql_read_write;
+    bool32 ddl_enabled;
 } knl_attr_t;
 
 typedef struct st_sys_name_context {  // for system name
diff --git a/pkg/src/server/params/load_server.c b/pkg/src/server/params/load_server.c
index 1bef06ea617c80a75e6fd2366cad39f758189b7c..56f2832579511c9e782f3782123c5127d116fe43 100644
--- a/pkg/src/server/params/load_server.c
+++ b/pkg/src/server/params/load_server.c
@@ -1026,6 +1026,8 @@ status_t srv_load_server_params(void)
         srv_get_param_uint32("PREVENT_SNAPSHOT_BACKUP_RECYCLE_REDO_TIMEOUT", &g_instance->kernel.attr.prevent_snapshot_backup_recycle_redo_timeout));
     CT_RETURN_IFERR(
         srv_get_param_bool32("PREVENT_CREATE_SNAPSHOT", &g_instance->kernel.attr.prevent_create_snapshot));
+    CT_RETURN_IFERR(srv_get_param_bool32("DDL_ENABLED",
+                                         &g_instance->kernel.attr.ddl_enabled));
     return CT_SUCCESS;
 }
 
diff --git a/pkg/src/server/params/set_kernel.c b/pkg/src/server/params/set_kernel.c
index aa91d3ca98ad0119fd1ab371e7df2b8a37f62ff4..e09a3519b293c609ff5313e7bf9c9c6fc2545c35 100644
--- a/pkg/src/server/params/set_kernel.c
+++ b/pkg/src/server/params/set_kernel.c
@@ -28,6 +28,7 @@
 #include "srv_param_common.h"
 #include "cm_io_record.h"
 #include "cm_log.h"
+#include "ctc_ddl.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -2331,6 +2332,19 @@ status_t sql_notify_enable_broadcast_on_commit(void *se, void *item, char *value
     return sql_notify_als_bool(se, item, value);
 }
 
+status_t sql_notify_ddl_enabled(void *se, void *item, char *value)
+{
+    if (DB_ATTR_MYSQL_META_IN_DACC((knl_session_t *)se)) {
+        CT_RETURN_IFERR(ctc_alter_ddl_enabled_for_backup(se, (bool32)value[0]));
+    } else {
+        CT_RETURN_IFERR(ctc_alter_ddl_enabled_for_backup_ex(se, (bool32)value[0]));
+    }
+
+    g_instance->kernel.attr.ddl_enabled = (bool32)value[0];
+    // restore value for alter config.
+    return sql_notify_als_bool(se, item, value);
+}
+
 status_t sql_notify_enable_enable_check_security_log(void *se, void *item, char *value)
 {
     g_filter_enable = (bool32)value[0];
diff --git a/pkg/src/server/params/set_kernel.h b/pkg/src/server/params/set_kernel.h
index 08abcbff1999d43a6f35f0b4edfef89557b720bd..86a824f8ce000a4f55dadcc9b0a25ddd4017c55c 100644
--- a/pkg/src/server/params/set_kernel.h
+++ b/pkg/src/server/params/set_kernel.h
@@ -207,6 +207,7 @@ status_t sql_notify_ctc_stats(void *se, void *item, char *value);
 status_t sql_notify_event_tracking_stats(void *se, void *item, char *value);
 status_t sql_notify_als_page_clean_mode(void *se, void *item, char *value);
 status_t sql_notify_enable_broadcast_on_commit(void *se, void *item, char *value);
+status_t sql_notify_ddl_enabled(void *se, void *item, char *value);
 status_t sql_notify_enable_enable_check_security_log(void *se, void *item, char *value);
 status_t sql_notify_enable_crc_check(void *se, void *item, char *value);
 #ifdef __cplusplus
diff --git a/pkg/src/server/params/srv_param.c b/pkg/src/server/params/srv_param.c
index d6c885bb7b0e60fd5bdbb9afcdbf91bca971430f..340b01efe8c491019108f29f541945f198f057d9 100644
--- a/pkg/src/server/params/srv_param.c
+++ b/pkg/src/server/params/srv_param.c
@@ -1181,6 +1181,8 @@ config_item_t g_parameters[] = {
     { "CTSQL_READ_WRITE", CT_TRUE, ATTR_READONLY, "FALSE", NULL, NULL, "-", "FALSE,TRUE", "CT_TYPE_BOOLEAN",
       NULL, PARAM_CTSQL_READ_WRITE, EFFECT_REBOOT, CFG_DB, sql_verify_als_bool, sql_notify_als_bool,
       sql_notify_als_bool, NULL },
+    { "DDL_ENABLED", CT_TRUE, ATTR_NONE, "TRUE", NULL, NULL, "-", "FALSE,TRUE", "CT_TYPE_BOOLEAN", NULL,
+      PARAM_DDL_ENABLED, EFFECT_REBOOT, CFG_INS, sql_verify_als_bool, sql_notify_ddl_enabled, sql_notify_als_bool, NULL },
 };
 
 void srv_get_config_info(config_item_t **params, uint32 *count)
diff --git a/pkg/src/server/params/srv_param_def.h b/pkg/src/server/params/srv_param_def.h
index 4bd9ace0a163257fadbd3a344a957eb58ace1600..aa28566846765a47b8909968b29240ba29c13e08 100644
--- a/pkg/src/server/params/srv_param_def.h
+++ b/pkg/src/server/params/srv_param_def.h
@@ -452,6 +452,7 @@ typedef enum en_param_global {
     PARAM_PREVENT_SNAPSHOT_BACKUP_RECYCLE_REDO_TIMEOUT,
     PARAM_PREVENT_CREATE_SNAPSHOT,
     PARAM_CTSQL_READ_WRITE,
+    PARAM_DDL_ENABLED,
     PARAM_CEIL,
 } param_global_t;
 
diff --git a/pkg/src/utils/ctbackup/ctbackup_common.h b/pkg/src/utils/ctbackup/ctbackup_common.h
index 44aba7f06703659de4c7d4109e161e8ad047f70d..067fd9ac2c07b0fa76677887e1e21821309ddf30 100644
--- a/pkg/src/utils/ctbackup/ctbackup_common.h
+++ b/pkg/src/utils/ctbackup/ctbackup_common.h
@@ -150,6 +150,8 @@ extern "C" {
 #define CTSQL_CMD_IN_BUFFER_SIZE      (CT_MAX_CMD_LEN + 1)
 #define CT_SNAPSHOT_BACKUP_NAME_MAX_LEN 255
 
+#define CTSQL_DDL_ENABLED_STATEMENT_PREFIX "ALTER SYSTEM SET DDL_ENABLED ="
+
 #ifndef WIFEXITED
 #define WIFEXITED(w)	(((w) & 0XFFFFFF00) == 0)
 #define WIFSIGNALED(w)	(!WIFEXITED(w))
diff --git a/pkg/src/utils/ctbackup/ctbackup_snapshot.c b/pkg/src/utils/ctbackup/ctbackup_snapshot.c
index 695b6e9089f224bd0e73c7a5b4b48fcce596afde..e07c52ae0a7db9b9eb5aa6dc8d668c33d58ce194 100644
--- a/pkg/src/utils/ctbackup/ctbackup_snapshot.c
+++ b/pkg/src/utils/ctbackup/ctbackup_snapshot.c
@@ -685,6 +685,156 @@ status_t ctbak_do_snapshot(ctbak_param_t* ctbak_param)
     return CT_ERROR;
 }
 
+status_t ctbak_check_snapshot_params(ctbak_param_t* ctbak_param)
+{
+    int32 parallelism_count;
+    if (ctbak_param->is_mysql_metadata_in_cantian == CT_FALSE &&
+        (ctbak_param->target_dir.str == NULL || ctbak_param->target_dir.len == 0)) {
+        printf("[ctbackup]The --target-dir parameter cannot be NULL!\n");
+        free_input_params(ctbak_param);
+        return CT_ERROR;
+    }
+    if (ctbak_param->is_mysql_metadata_in_cantian == CT_FALSE &&
+        ctbak_param->target_dir.len > MAX_TARGET_DIR_LENGTH) {
+        printf("[ctbackup]The --target-dir parameter length is too long!\n");
+        free_input_params(ctbak_param);
+        return CT_ERROR;
+    }
+
+    if (ctbak_param->parallelism.str != NULL && ctbak_param->parallelism.len != 0) {
+        if (cm_str2int(ctbak_param->parallelism.str, &parallelism_count) != CT_SUCCESS) {
+            printf("[ctbackup]convert parallelism to int32 failed!\n");
+            free_input_params(ctbak_param);
+            return CT_ERROR;
+        }
+
+        if (parallelism_count > MAX_PARALLELISM_COUNT || parallelism_count <= 0) {
+            printf("[ctbackup]The --parallel parameter value should be in [1, 16].\n");
+            free_input_params(ctbak_param);
+            return CT_ERROR;
+        }
+    }
+    return CT_SUCCESS;
+}
+
+static status_t get_statement_for_ddl_lock(uint8 is_enabled, uint64_t len, char *statement)
+{
+    errno_t ret;
+    if (is_enabled == CT_TRUE) {
+        ret = snprintf_s(statement, len, len - 1, "%s%s%s", CTSQL_DDL_ENABLED_STATEMENT_PREFIX,
+                         CTSQL_TRUE, CTSQL_STATEMENT_END_CHARACTER);
+    } else {
+        ret = snprintf_s(statement, len, len - 1, "%s%s%s", CTSQL_DDL_ENABLED_STATEMENT_PREFIX,
+                         CTSQL_FALSE, CTSQL_STATEMENT_END_CHARACTER);
+    }
+    FREE_AND_RETURN_ERROR_IF_SNPRINTF_FAILED(ret, statement);
+    return CT_SUCCESS;
+}
+
+static status_t fill_params_for_ddl_lock(uint8 is_enabled, char *ct_params[])
+{
+    int param_index = 0;
+    uint64_t len;
+    if (fill_params_for_ctsql_login(ct_params, &param_index, CTBAK_CTSQL_EXECV_MODE) != CT_SUCCESS) {
+        printf("[fill_params_for_ddl_lock]failed to fill params for ctsql login!\n");
+        return CT_ERROR;
+    }
+    len = strlen(CTSQL_DDL_ENABLED_STATEMENT_PREFIX);
+    if (is_enabled == CT_TRUE) {
+        len = len + strlen(CTSQL_TRUE) + strlen(CTSQL_STATEMENT_END_CHARACTER) + 1;
+    } else {
+        len = len + strlen(CTSQL_FALSE) + strlen(CTSQL_STATEMENT_END_CHARACTER) + 1;
+    }
+    char *statement = (char *)malloc(len);
+    if (statement == NULL) {
+        CM_FREE_PTR(ct_params[CTSQL_LOGININFO_INDEX]);
+        printf("[ctbackup]failed to fill_params_for_ddl_lock!\n");
+        CTBAK_RETURN_ERROR_IF_NULL(statement);
+    }
+    if (get_statement_for_ddl_lock(is_enabled, len, statement) != CT_SUCCESS) {
+        CM_FREE_PTR(ct_params[CTSQL_LOGININFO_INDEX]);
+        printf("[ctbackup]get statement for fill_params_for_ddl_lock!\n");
+        return CT_ERROR;
+    }
+    ct_params[param_index++] = statement;
+    ct_params[param_index++] = NULL;
+    return CT_SUCCESS;
+}
+
+static status_t ctbak_do_ctsql_ddl_enabled(char *path, char *params[], uint8 *retry)
+{
+    errno_t status = 0;
+    int32 pipe_stdout[2] = { 0 };
+    if (pipe(pipe_stdout) != EOK) {
+        printf("[ctbackup]create stdout pipe failed!\n");
+        return CT_ERROR;
+    }
+
+    pid_t child_pid = fork();
+    if (child_pid == 0) {
+        prctl(PR_SET_PDEATHSIG, SIGKILL);
+        close(pipe_stdout[PARENT_ID]);
+        dup2(pipe_stdout[CHILD_ID], STD_OUT_ID);
+        status = execv(path, params);
+        perror("execve");
+        if (status != EOK) {
+            printf("[ctbackup]failed to execute shell command %d:%s\n", errno, strerror(errno));
+            exit(CT_ERROR);
+        }
+    } else if (child_pid < 0) {
+        printf("[ctbackup]failed to fork child process with result %d:%s\n", errno, strerror(errno));
+        return CT_ERROR;
+    }
+    close(pipe_stdout[CHILD_ID]);
+    bool32 need_retry = CT_FALSE;
+    int32 wait = waitpid(child_pid, &status, 0);
+    if (wait == child_pid && WIFEXITED((unsigned int)status) && WEXITSTATUS((unsigned int)status) != 0) {
+        printf("[ctbackup]child process exec ddl_enabled failed, ret=%d, try to check cantian stat.\n",  status);
+        if (need_retry == CT_TRUE &&
+            ctbak_check_ctsql_online(CTBAK_SNAPSHOT_CECHK_ONLINE_TIME) == CT_SUCCESS) {
+            *retry = CT_TRUE;
+        }
+        return CT_ERROR;
+    }
+    printf("[ctbackup]ctbak_do_ctsql_ddl_enabled execute success and exit with: %d\n",
+           WEXITSTATUS((unsigned int)status));
+    return CT_SUCCESS;
+}
+
+ /**
+  * @brief DDL开启/禁用接口
+  * 
+  * @param is_enabled CT_TRUE表示开启,CT_FALSE表示禁用
+  * @param retry 
+  * @return status_t 
+  */
+status_t ctbak_alter_ddl_enabled(uint8 is_enabled, uint8 *retry)
+{
+    char *ct_params[CTBACKUP_MAX_PARAMETER_CNT] = { 0 };
+    status_t status = fill_params_for_ddl_lock(is_enabled, ct_params);
+    if (status != CT_SUCCESS) {
+        printf("[ctbackup]fill_params_for_ddl_lock failed!\n");
+        return CT_ERROR;
+    }
+    char *ctsql_binary_path = NULL;
+    if (get_ctsql_binary_path(&ctsql_binary_path) != CT_SUCCESS) {
+        CM_FREE_PTR(ct_params[CTSQL_LOGININFO_INDEX]);
+        CM_FREE_PTR(ct_params[CTSQL_STATEMENT_INDEX]);
+        printf("[ctbackup]get_ctsql_binary_path failed!\n");
+        return CT_ERROR;
+    }
+    status = ctbak_do_ctsql_ddl_enabled(ctsql_binary_path, ct_params, retry);
+    // free space of heap
+    CM_FREE_PTR(ct_params[CTSQL_LOGININFO_INDEX]);
+    CM_FREE_PTR(ct_params[CTSQL_STATEMENT_INDEX]);
+    CM_FREE_PTR(ctsql_binary_path);
+    if (status != CT_SUCCESS) {
+        printf("[ctbackup]ctsql set ddl enabled %d failed!\n", is_enabled);
+        return CT_ERROR;
+    }
+    return CT_SUCCESS;
+}
+
 static inline void ctbak_hide_password(char* password)
 {
     while (*password) {