diff --git a/common/templates/config.html b/common/templates/config.html
index 93af8eeb1f..236247dd5e 100755
--- a/common/templates/config.html
+++ b/common/templates/config.html
@@ -304,6 +304,18 @@ <h5 style="color: darkgrey"><b>SQL上线</b></h5>
                                     </div>
                                 </div>
                             </div>
+                            <div class="form-group">
+                                <label for="query_low_peak_query"
+                                       class="col-sm-4 control-label">QUERY_LOW_PEAK_QUERY</label>
+                                <div class="col-sm-5">
+                                    <select id="query_low_peak_query" key="query_low_peak_query"
+                                            class="selectpicker show-tick form-control bs-select-hidden"
+                                            data-live-search="true"
+                                            data-none-selected-text="必须在低峰期执行操作操作？默认DDL"
+                                            multiple required>
+                                    </select>
+                                </div>
+                            </div>
                             <div class="form-group">
                                 <label for="ban_self_audit"
                                        class="col-sm-4 control-label">BAN_SELF_AUDIT</label>
@@ -1183,6 +1195,11 @@ <h5 class="control-label text-bold">当前审批流程：<b id="workflow_auditor
             $("#notify_phase_control").selectpicker('render');
             $("#notify_phase_control").selectpicker('refresh');
 
+            // query_low_peak_query参数处理
+            let queryConfig = "{{ config.query_low_peak_query }}";
+            let queryPhases = ['DML', 'DDL'];
+            setupSelectPicker("#query_low_peak_query", queryConfig, queryPhases);
+
             // api_user_whitelist参数处理
             let api_config = "{{ config.api_user_whitelist }}";
             $.ajax({
diff --git a/sql/models.py b/sql/models.py
index 405692459d..9f8eec46bd 100755
--- a/sql/models.py
+++ b/sql/models.py
@@ -1,4 +1,5 @@
 # -*- coding: UTF-8 -*-
+import re
 from typing import Optional
 
 from django.db import models
@@ -6,6 +7,7 @@
 from mirage import fields
 from django.utils.translation import gettext as _
 from mirage.crypto import Crypto
+from django.core.exceptions import ValidationError
 
 from common.utils.const import WorkflowStatus, WorkflowType, WorkflowAction
 
@@ -177,6 +179,21 @@ class Meta:
         verbose_name_plural = "隧道配置"
 
 
+def validate_peak_time_period(value):
+    """
+    验证 peak_time_period 格式是否正确。
+    格式示例: "14:00-14:50,15:00-16:00"
+    """
+    if value is None or value == "":
+        return
+
+    time_period_regex = r"^(\d{2}:\d{2}-\d{2}:\d{2})(,\d{2}:\d{2}-\d{2}:\d{2})*$"
+    if not re.match(time_period_regex, value):
+        raise ValidationError(
+            '时间段格式不正确，正确格式为 "HH:MM-HH:MM" 或 "HH:MM-HH:MM,HH:MM-HH:MM"'
+        )
+
+
 class Instance(models.Model):
     """
     各个线上实例配置
@@ -223,6 +240,13 @@ class Instance(models.Model):
         on_delete=models.CASCADE,
         default=None,
     )
+    peak_time_period = models.CharField(
+        "业务低峰时间段",
+        max_length=255,
+        default="",
+        blank=True,
+        validators=[validate_peak_time_period],
+    )
     create_time = models.DateTimeField("创建时间", auto_now_add=True)
     update_time = models.DateTimeField("更新时间", auto_now=True)
 
diff --git a/sql/sql_workflow.py b/sql/sql_workflow.py
index b812be7c51..48a73bfe6a 100644
--- a/sql/sql_workflow.py
+++ b/sql/sql_workflow.py
@@ -24,6 +24,7 @@
     can_timingtask,
     can_cancel,
     can_execute,
+    on_query_low_peak_time_ddl,
     on_correct_time_period,
     can_view,
     can_rollback,
@@ -306,6 +307,17 @@ def execute(request):
             "errMsg": "不在可执行时间范围内，如果需要修改执行时间请重新提交工单!"
         }
         return render(request, "error.html", context)
+    if not request.user.is_superuser:
+        sys_config = SysConfig()
+        is_allowed, time_periods = on_query_low_peak_time_ddl(workflow_id)
+        if not is_allowed:
+            peak_action = sys_config.get("query_low_peak_query", "")
+            context = {
+                "errMsg": "管理员设置了实例业务低峰时间范围:%s,你只能在业务低峰时间范围执行%s工单操作!"
+                % (time_periods, peak_action)
+            }
+        return render(request, "error.html", context)
+
     # 获取审核信息
     audit_id = Audit.detail_by_workflow_id(
         workflow_id=workflow_id, workflow_type=WorkflowType.SQL_REVIEW
diff --git a/sql/utils/sql_review.py b/sql/utils/sql_review.py
index 78296247f6..a1438ed9da 100644
--- a/sql/utils/sql_review.py
+++ b/sql/utils/sql_review.py
@@ -4,7 +4,7 @@
 from django.db import transaction
 
 from sql.engines.models import ReviewResult
-from sql.models import SqlWorkflow
+from sql.models import SqlWorkflow, Instance
 from common.config import SysConfig
 from sql.utils.resource_group import user_groups
 from sql.utils.sql_utils import remove_comments
@@ -41,6 +41,38 @@ def can_execute(user, workflow_id):
     return result
 
 
+def on_query_low_peak_time_ddl(workflow_id, run_date=None):
+    """
+    判断是否是DDL，DDL必须在业务低峰期执行，包括人工执行和定时执行
+    :param workflow_id:
+    :param run_date:
+    :return: bool
+    """
+    workflow_detail = SqlWorkflow.objects.get(id=workflow_id)
+    instance_id = workflow_detail.instance_id  # 假设 SqlWorkflow 有 instance_id 字段
+    instance = Instance.objects.get(id=instance_id)
+    peak_time_period = instance.peak_time_period
+
+    # 解析 peak_time_period 字符串为时间段列表
+    if peak_time_period:
+        time_periods = [
+            tuple(map(int, period.split("-"))) for period in peak_time_period.split(",")
+        ]
+    else:
+        time_periods = []
+
+    ctime = run_date or datetime.datetime.now()
+    current_hour = ctime.hour
+    syntax_type = workflow_detail.syntax_type
+
+    # 如果是DDL操作，检查当前时间是否在低峰期
+    if syntax_type == 1:
+        if not any(start <= current_hour < end for start, end in time_periods):
+            return False, time_periods
+
+    return True, None
+
+
 def on_correct_time_period(workflow_id, run_date=None):
     """
     判断是否在可执行时间段内，包括人工执行和定时执行
diff --git a/sql_api/api_workflow.py b/sql_api/api_workflow.py
index d834a5af30..809bdcbc00 100644
--- a/sql_api/api_workflow.py
+++ b/sql_api/api_workflow.py
@@ -26,7 +26,12 @@
 from sql.notify import notify_for_audit, notify_for_execute
 from sql.query_privileges import _query_apply_audit_call_back
 from sql.utils.resource_group import user_groups
-from sql.utils.sql_review import can_cancel, can_execute, on_correct_time_period
+from sql.utils.sql_review import (
+    can_cancel,
+    can_execute,
+    on_correct_time_period,
+    on_query_low_peak_time_ddl,
+)
 from sql.utils.tasks import del_schedule
 from sql.utils.workflow_audit import Audit, get_auditor, AuditException
 from .filters import WorkflowFilter, WorkflowAuditFilter
@@ -340,6 +345,20 @@ def post(self, request):
                         "errors": "不在可执行时间范围内，如果需要修改执行时间请重新提交工单!"
                     }
                 )
+            is_allowed, error_message = on_query_low_peak_time_ddl(workflow_id)
+            if not request.user.is_superuser and not is_allowed:
+                raise serializers.ValidationError({"errMsg": error_message})
+            if not request.user.is_superuser:
+                sys_config = SysConfig()
+                is_allowed, time_periods = on_query_low_peak_time_ddl(workflow_id)
+                if not is_allowed:
+                    peak_action = sys_config.get("query_low_peak_query", "")
+                    raise serializers.ValidationError(
+                        {
+                            "errMsg": "管理员设置了业务低峰期时间范围:%s,你只能在业务低峰时间范围执行%s工单操作!"
+                            % (time_periods, peak_action)
+                        }
+                    )
 
             # 获取审核信息
             audit_id = Audit.detail_by_workflow_id(
diff --git a/src/init_sql/v1.11.4.sql b/src/init_sql/v1.11.4.sql
new file mode 100644
index 0000000000..bfebc35083
--- /dev/null
+++ b/src/init_sql/v1.11.4.sql
@@ -0,0 +1 @@
+ALTER TABLE sql_instance ADD COLUMN peak_time_period VARCHAR(255);
\ No newline at end of file