diff --git a/kystudio/src/components/setting/SettingBasic/SettingBasic.vue b/kystudio/src/components/setting/SettingBasic/SettingBasic.vue
index a0268f50670..7d1dd631c9b 100644
--- a/kystudio/src/components/setting/SettingBasic/SettingBasic.vue
+++ b/kystudio/src/components/setting/SettingBasic/SettingBasic.vue
@@ -93,6 +93,16 @@
       @submit="(scb, ecb) => handleSubmit('segment-settings', scb, ecb)"
       @cancel="(scb, ecb) => handleResetForm('segment-settings', scb, ecb)">
       <el-form ref="segment-setting-form" :model="form" :rules="rules">
+        <div class="setting-item">
+          <span class="setting-label font-medium">{{$t('autoSegmentBuild')}}</span><span class="setting-value fixed">
+            <el-switch
+              v-model="form.auto_segment_build_enabled"
+              :active-text="$t('kylinLang.common.OFF')"
+              :inactive-text="$t('kylinLang.common.ON')">
+            </el-switch>
+          </span>
+          <div class="setting-desc">{{$t('autoSegmentBuildDesc')}}</div>
+        </div>
         <div class="setting-item">
           <span class="setting-label font-medium">{{$t('segmentMerge')}}</span><span class="setting-value fixed">
             <el-switch
diff --git a/kystudio/src/components/setting/SettingBasic/handler.js b/kystudio/src/components/setting/SettingBasic/handler.js
index 9a5e35ecf50..6eb37702f93 100644
--- a/kystudio/src/components/setting/SettingBasic/handler.js
+++ b/kystudio/src/components/setting/SettingBasic/handler.js
@@ -24,6 +24,7 @@ export const initialFormValue = {
   push_down_range_limited: true,
   auto_merge_enabled: true,
   auto_merge_time_ranges: [ 'WEEK', 'MONTH' ],
+  auto_segment_build_enabled: false,
   storage_garbage: true,
   storage_quota_size: 0,
   storage_quota_tb_size: 0,
@@ -81,6 +82,7 @@ export function _getSegmentSettings (data, project) {
     project: data.project,
     auto_merge_time_ranges: data.auto_merge_time_ranges,
     auto_merge_enabled: data.auto_merge_enabled,
+    auto_segment_build_enabled: data.auto_segment_build_enabled,
     volatile_range: {
       ...data.volatile_range,
       volatile_range_number: data.volatile_range.volatile_range_number + '',
diff --git a/kystudio/src/components/setting/SettingBasic/locales.js b/kystudio/src/components/setting/SettingBasic/locales.js
index 8cbd0b18d0b..5a7874cd8b4 100644
--- a/kystudio/src/components/setting/SettingBasic/locales.js
+++ b/kystudio/src/components/setting/SettingBasic/locales.js
@@ -11,6 +11,8 @@ export default {
     pushdownRange: 'Pushdown Range',
     pushdownRangeDesc: 'Turn off the pushdown range setting may cause the result difference between query on source data and indexes, especially when a source table has new data yet indexes don\'t.',
     segmentSettings: 'Segment Settings',
+    autoSegmentBuild: 'Auto Segment Build',
+    autoSegmentBuildDesc: 'When enabled, the master node periodically scans this project and submits jobs for eligible model-level auto segment build schedules.',
     datasourceSetting: 'Data Source Setting',
     gbaseConnectInfo: 'GBase Data Source Information',
     JDBCDataSource: 'JDBC Datasource',
diff --git a/kystudio/src/components/setting/SettingModel/SettingModel.vue b/kystudio/src/components/setting/SettingModel/SettingModel.vue
index 9e3725d3762..f2526bb03f2 100644
--- a/kystudio/src/components/setting/SettingModel/SettingModel.vue
+++ b/kystudio/src/components/setting/SettingModel/SettingModel.vue
@@ -46,6 +46,15 @@
               <i class="el-icon-ksd-symbol_type" @click="removeAutoMerge(scope.row, 'retention_range')"></i>
             </common-tip>
           </div>
+          <div v-if="scope.row.auto_segment_build && scope.row.auto_segment_build.enabled">
+            <span class="model-setting-item" @click="editAutoSegmentBuildItem(scope.row)">
+              {{$t('autoSegmentBuild')}}<span>{{formatAutoSegmentBuild(scope.row.auto_segment_build)}}</span>
+            </span><common-tip :content="$t('kylinLang.common.edit')">
+              <i class="el-icon-ksd-table_edit ksd-mr-5 ksd-ml-10" @click="editAutoSegmentBuildItem(scope.row)"></i>
+            </common-tip><common-tip :content="$t('kylinLang.common.delete')">
+              <i class="el-icon-ksd-symbol_type" @click="removeAutoMerge(scope.row, 'auto_segment_build')"></i>
+            </common-tip>
+          </div>
           <div v-if="Object.keys(scope.row.override_props).length">
             <div v-for="(propValue, key) in scope.row.override_props" :key="key">
               <template v-if="key.includes('kylin.engine.spark-conf') && defaultConfigs.includes(key.replace(/^kylin.engine.spark-conf./, ''))">
@@ -141,6 +150,44 @@
             </el-option>
           </el-select>
         </el-form-item>
+        <template v-if="step=='stepTwo'&&modelSettingForm.settingItem==='Auto Segment Build'">
+          <el-form-item :label="$t('autoSegmentBuildTriggerTime')">
+            <el-time-picker
+              style="width: 180px;"
+              size="small"
+              format="HH:mm:ss"
+              value-format="HH:mm:ss"
+              v-model="modelSettingForm.autoSegmentBuild.trigger_time">
+            </el-time-picker>
+          </el-form-item>
+          <el-form-item :label="$t('autoSegmentBuildLogicalOffset')">
+            <el-input v-model="modelSettingForm.autoSegmentBuild.logical_date_offset_days" v-number="modelSettingForm.autoSegmentBuild.logical_date_offset_days" class="retention-input"></el-input>
+          </el-form-item>
+          <el-form-item :label="$t('autoSegmentBuildRangeStart')">
+            <el-time-picker
+              style="width: 180px;"
+              size="small"
+              format="HH:mm:ss"
+              value-format="HH:mm:ss"
+              v-model="modelSettingForm.autoSegmentBuild.data_range_start_time">
+            </el-time-picker>
+          </el-form-item>
+          <el-form-item :label="$t('autoSegmentBuildRangeEnd')">
+            <el-time-picker
+              v-if="!isAutoSegmentBuildEndOfDay"
+              style="width: 180px;"
+              size="small"
+              format="HH:mm:ss"
+              value-format="HH:mm:ss"
+              v-model="modelSettingForm.autoSegmentBuild.data_range_end_time">
+            </el-time-picker>
+            <el-checkbox
+              :class="{'ksd-ml-10': !isAutoSegmentBuildEndOfDay}"
+              v-model="isAutoSegmentBuildEndOfDay">
+              {{$t('autoSegmentBuildEndOfDay')}}
+            </el-checkbox>
+          </el-form-item>
+        </template>
         <el-form-item :label="settingMap[modelSettingForm.settingItem]" v-if="step=='stepTwo'&&modelSettingForm.settingItem.indexOf('spark.')!==-1">
           <el-input v-model="modelSettingForm[modelSettingForm.settingItem]" v-number="modelSettingForm[modelSettingForm.settingItem]" :placeholder="$t('kylinLang.common.pleaseInput')" class="retention-input"></el-input><span
           class="ksd-ml-5" v-if="modelSettingForm.settingItem==='spark.executor.memory'">G</span>
@@ -196,12 +243,15 @@ import { handleSuccess, transToGmtTime, kylinConfirm } from '../../../util/busin
 import { handleSuccessAsync, handleError, objectClone, ArrayFlat } from '../../../util/index'
 import { retentionTypes } from '../handler'
 
+const END_OF_DAY = '24:00:00'
+
 const initialSettingForm = JSON.stringify({
   name: '',
   settingItem: '',
   autoMerge: [],
   volatileRange: {volatile_range_number: 0, volatile_range_type: '', volatile_range_enabled: true},
   retentionThreshold: {retention_range_number: 0, retention_range_type: '', retention_range_enabled: true},
+  autoSegmentBuild: {enabled: true, trigger_time: '01:00:00', logical_date_offset_days: 1, data_range_start_time: '00:00:00', data_range_end_time: END_OF_DAY},
   'spark.executor.cores': null,
   'spark.executor.instances': null,
   'spark.executor.memory': null,
@@ -255,6 +305,7 @@ export default class SettingStorage extends Vue {
     'Auto-merge': 'Auto-merge',
     'Volatile Range': 'Volatile Range',
     'Retention Threshold': 'Retention Threshold',
+    'Auto Segment Build': 'Auto Segment Build',
     'spark.executor.cores': 'kylin.engine.spark-conf.spark.executor.cores',
     'spark.executor.instances': 'kylin.engine.spark-conf.spark.executor.instances',
     'spark.executor.memory': 'kylin.engine.spark-conf.spark.executor.memory',
@@ -275,6 +326,7 @@ export default class SettingStorage extends Vue {
       'Auto-merge',
       'Volatile Range',
       'Retention Threshold',
+      'Auto Segment Build',
       'spark.executor.cores',
       'spark.executor.instances',
       'spark.executor.memory',
@@ -295,6 +347,12 @@ export default class SettingStorage extends Vue {
     // return largestRange || ''
     return 'DAY'
   }
+  get isAutoSegmentBuildEndOfDay () {
+    return this.modelSettingForm.autoSegmentBuild.data_range_end_time === END_OF_DAY
+  }
+  set isAutoSegmentBuildEndOfDay (isEndOfDay) {
+    this.modelSettingForm.autoSegmentBuild.data_range_end_time = isEndOfDay ? END_OF_DAY : null
+  }
   validateSettingItem (rule, value, callback) {
     const autoMergeRanges = this.activeRow && this.activeRow.auto_merge_time_ranges || []
     if (this.step === 'stepOne' && value === 'Retention Threshold' && !autoMergeRanges.length) {
@@ -308,6 +366,7 @@ export default class SettingStorage extends Vue {
       'Auto-merge': this.$t('autoMergeTip'),
       'Volatile Range': this.$t('volatileTip'),
       'Retention Threshold': this.$t('retentionThresholdDesc'),
+      'Auto Segment Build': this.$t('autoSegmentBuildTip'),
       'kylin.engine.spark-conf.spark.executor.cores': this.$t('sparkCores'),
       'kylin.engine.spark-conf.spark.executor.instances': this.$t('sparkInstances'),
       'kylin.engine.spark-conf.spark.executor.memory': this.$t('sparkMemory'),
@@ -346,6 +405,8 @@ export default class SettingStorage extends Vue {
       return true
     } else if (this.modelSettingForm.settingItem === 'Retention Threshold' && !(this.modelSettingForm.retentionThreshold.retention_range_number >= 0 && this.modelSettingForm.retentionThreshold.retention_range_number !== '' && this.modelSettingForm.retentionThreshold.retention_range_type)) {
       return true
+    } else if (this.modelSettingForm.settingItem === 'Auto Segment Build' && !this.isValidAutoSegmentBuild()) {
+      return true
     } else if (this.modelSettingForm.settingItem.indexOf('spark.') !== -1 && !this.modelSettingForm[this.modelSettingForm.settingItem]) {
       return true
     } else if (this.modelSettingForm.settingItem === 'is-base-cuboid-always-valid' && this.modelSettingForm[this.modelSettingForm.settingItem] === '') {
@@ -439,6 +500,15 @@ export default class SettingStorage extends Vue {
     this.isEdit = true
     this.editModelSetting = true
   }
+  editAutoSegmentBuildItem (row) {
+    this.modelSettingForm.name = row.alias
+    this.modelSettingForm.settingItem = 'Auto Segment Build'
+    this.modelSettingForm.autoSegmentBuild = JSON.parse(JSON.stringify(row.auto_segment_build))
+    this.activeRow = row
+    this.step = 'stepTwo'
+    this.isEdit = true
+    this.editModelSetting = true
+  }
   editSparkItem (row, sparkItemKey) {
     this.modelSettingForm.name = row.alias
     this.modelSettingForm.settingItem = sparkItemKey.substring(24)
@@ -489,6 +559,13 @@ export default class SettingStorage extends Vue {
     if (this.modelSettingForm.settingItem === 'Retention Threshold') {
       this.activeRow.retention_range = this.modelSettingForm.retentionThreshold
     }
+    if (this.modelSettingForm.settingItem === 'Auto Segment Build') {
+      this.activeRow.auto_segment_build = {
+        ...this.modelSettingForm.autoSegmentBuild,
+        enabled: true,
+        logical_date_offset_days: Number(this.modelSettingForm.autoSegmentBuild.logical_date_offset_days)
+      }
+    }
     if (this.modelSettingForm.settingItem.indexOf('spark.') !== -1) {
       this.activeRow.override_props['kylin.engine.spark-conf.' + this.modelSettingForm.settingItem] = this.modelSettingForm[this.modelSettingForm.settingItem]
     }
@@ -544,6 +621,22 @@ export default class SettingStorage extends Vue {
   removeCustomSetting (index) {
     this.modelSettingForm[this.modelSettingForm.settingItem].splice(index, 1)
   }
+  isValidAutoSegmentBuild () {
+    const config = this.modelSettingForm.autoSegmentBuild
+    if (!config || !config.trigger_time || !config.data_range_start_time || !config.data_range_end_time) return false
+    if (!(+config.logical_date_offset_days >= 1)) return false
+    if (config.data_range_end_time === END_OF_DAY) return true
+    return config.data_range_start_time < config.data_range_end_time
+  }
+  formatAutoSegmentBuild (config) {
+    if (!config) return ''
+    return this.$t('autoSegmentBuildSummary', {
+      trigger: config.trigger_time,
+      offset: config.logical_date_offset_days,
+      start: config.data_range_start_time,
+      end: config.data_range_end_time
+    })
+  }
   created () {
     this.getConfigList()
   }
diff --git a/kystudio/src/components/setting/SettingModel/locales.js b/kystudio/src/components/setting/SettingModel/locales.js
index a0c7c22c1e8..e9328b21e0f 100644
--- a/kystudio/src/components/setting/SettingModel/locales.js
+++ b/kystudio/src/components/setting/SettingModel/locales.js
@@ -6,6 +6,7 @@ export default {
     segmentMerge: 'Segment Merge:',
     volatileRange: 'Volatile Range:',
     retention: 'Retention Threshold:',
+    autoSegmentBuild: 'Auto Segment Build:',
     newSetting: 'Add Parameter Configuration',
     editSetting: 'Edit Parameter Configuration',
     modelName: 'Model Name',
@@ -34,6 +35,7 @@ export default {
     'isDel_kylin.engine.spark-conf.spark.executor.memory': 'Are you sure delete spark.executor.memory item?',
     'isDel_kylin.engine.spark-conf.spark.sql.shuffle.partitions': 'Are you sure delete spark.sql.shuffle.partitions item?',
     'isDel_kylin.cube.aggrgroup.is-base-cuboid-always-valid': 'Are you sure delete is-base-cuboid-always-valid item?',
+    isDel_auto_segment_build: 'Are you sure delete auto segment build configuration?',
     autoMergeTip: 'The system could auto-merge segment fragments over different merging threshold. Auto-merge will optimize storage to enhance query performance.',
     volatileTip: '"Auto-Merge" will not merge the latest segments defined in "Volatile Range". The default value is 0.',
     retentionThresholdDesc: 'The segments within the retention threshold would be kept. The rest would be removed automatically.',
@@ -41,6 +43,7 @@ export default {
     'Auto-merge': 'Auto Merge',
     'Volatile Range': 'Volatile Range',
     'Retention Threshold': 'Retention Threshold',
+    'Auto Segment Build': 'Auto Segment Build',
     'kylin.engine.spark-conf.spark.executor.cores': 'kylin.engine.spark-conf.spark.executor.cores',
     'kylin.engine.spark-conf.spark.executor.instances': 'kylin.engine.spark-conf.spark.executor.instances',
     'kylin.engine.spark-conf.spark.executor.memory': 'kylin.engine.spark-conf.spark.executor.memory',
@@ -55,6 +58,13 @@ export default {
     customOptions: 'Besides the defined configurations, you can also add some advanced settings.<br/><i class="el-icon-ksd-alert"></i>Note: It\'s highly recommended to use this feature with the support of Kylin 5 Team.',
     customSettingKeyPlaceholder: 'Configuration Name',
     customSettingValuePlaceholder: 'Value',
-    delCustomConfigTip: 'Are you sure you want to delete custom setting item {name}？'
+    delCustomConfigTip: 'Are you sure you want to delete custom setting item {name}？',
+    autoSegmentBuildTriggerTime: 'Trigger Time',
+    autoSegmentBuildLogicalOffset: 'Logical Date Offset (Day)',
+    autoSegmentBuildRangeStart: 'Range Start Time',
+    autoSegmentBuildRangeEnd: 'Range End Time',
+    autoSegmentBuildEndOfDay: 'End of logical day (next day 00:00)',
+    autoSegmentBuildTip: 'Build segment automatically every day with logical day offset and configured time range.',
+    autoSegmentBuildSummary: 'Daily {trigger}, D-{offset} {start}~{end}'
   }
 }
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/request/SegmentConfigRequest.java b/src/common-service/src/main/java/org/apache/kylin/rest/request/SegmentConfigRequest.java
index 0e5aac08910..0c738716145 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/request/SegmentConfigRequest.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/request/SegmentConfigRequest.java
@@ -40,4 +40,6 @@ public class SegmentConfigRequest {
     private RetentionRange retentionRange = new RetentionRange();
     @JsonProperty("create_empty_segment_enabled")
     private Boolean createEmptySegmentEnabled = false;
+    @JsonProperty("auto_segment_build_enabled")
+    private Boolean autoSegmentBuildEnabled;
 }
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java b/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
index 66124c9dfca..e12267c6f41 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
@@ -82,6 +82,8 @@ public class ProjectConfigResponse {
     private VolatileRange volatileRange;
     @JsonProperty("create_empty_segment_enabled")
     private boolean createEmptySegmentEnabled = false;
+    @JsonProperty("auto_segment_build_enabled")
+    private boolean autoSegmentBuildEnabled = false;
 
     @JsonProperty("retention_range")
     private RetentionRange retentionRange;
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java b/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
index 3c5a1a89680..828f97b196c 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
@@ -602,6 +602,8 @@ public ProjectConfigResponse getProjectConfig0(String project) {
         response.setVolatileRange(projectInstance.getSegmentConfig().getVolatileRange());
         response.setRetentionRange(projectInstance.getSegmentConfig().getRetentionRange());
         response.setCreateEmptySegmentEnabled(projectInstance.getSegmentConfig().getCreateEmptySegmentEnabled());
+        response.setAutoSegmentBuildEnabled(
+                Boolean.TRUE.equals(projectInstance.getSegmentConfig().getAutoSegmentBuildEnabled()));
 
         response.setFavoriteQueryThreshold(config.getFavoriteQueryAccelerateThreshold());
         response.setFavoriteQueryTipsEnabled(config.getFavoriteQueryAccelerateTipsEnabled());
@@ -937,6 +939,10 @@ public void updateSegmentConfig(String project, SegmentConfigRequest segmentConf
             copyForWrite.getSegmentConfig().setRetentionRange(segmentConfigRequest.getRetentionRange());
             copyForWrite.getSegmentConfig()
                     .setCreateEmptySegmentEnabled(segmentConfigRequest.getCreateEmptySegmentEnabled());
+            if (segmentConfigRequest.getAutoSegmentBuildEnabled() != null) {
+                copyForWrite.getSegmentConfig()
+                        .setAutoSegmentBuildEnabled(segmentConfigRequest.getAutoSegmentBuildEnabled());
+            }
         });
     }
 
@@ -1199,6 +1205,8 @@ private void resetSegmentConfig(String project) {
                     .setAutoMergeTimeRanges(projectInstance.getSegmentConfig().getAutoMergeTimeRanges());
             copyForWrite.getSegmentConfig().setVolatileRange(projectInstance.getSegmentConfig().getVolatileRange());
             copyForWrite.getSegmentConfig().setRetentionRange(projectInstance.getSegmentConfig().getRetentionRange());
+            copyForWrite.getSegmentConfig()
+                    .setAutoSegmentBuildEnabled(projectInstance.getSegmentConfig().getAutoSegmentBuildEnabled());
         });
     }
 
diff --git a/src/core-common/src/main/resources/kylin-defaults0.properties b/src/core-common/src/main/resources/kylin-defaults0.properties
index dc1b57b9ed2..8a7af4d8c6c 100644
--- a/src/core-common/src/main/resources/kylin-defaults0.properties
+++ b/src/core-common/src/main/resources/kylin-defaults0.properties
@@ -449,6 +449,8 @@ kylin.web.session.jdbc-encode-enabled=false
 kylin.security.user-password-encoder=org.apache.kylin.rest.security.CachedBCryptPasswordEncoder
 
 # model
+# System-level internal scan cadence in milliseconds; model trigger times are read from metadata on every scan.
+kylin.model.auto-segment-build.dispatcher-interval-ms=30000
 kylin.model.recommendation-page-size=500
 kylin.model.dimension-measure-name.max-length=300
 
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/AutoSegmentBuildConfig.java b/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/AutoSegmentBuildConfig.java
new file mode 100644
index 00000000000..0b6fb728cd6
--- /dev/null
+++ b/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/AutoSegmentBuildConfig.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.metadata.model;
+
+import java.io.Serializable;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class AutoSegmentBuildConfig implements Serializable {
+
+    // LocalTime cannot represent 24:00:00; this value denotes midnight at the end of the logical date.
+    public static final String END_OF_DAY = "24:00:00";
+
+    @JsonProperty("enabled")
+    private boolean enabled = false;
+
+    @JsonProperty("trigger_time")
+    private String triggerTime;
+
+    @JsonProperty("logical_date_offset_days")
+    private Integer logicalDateOffsetDays;
+
+    @JsonProperty("data_range_start_time")
+    private String dataRangeStartTime;
+
+    @JsonProperty("data_range_end_time")
+    private String dataRangeEndTime;
+}
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/SegmentConfig.java b/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/SegmentConfig.java
index 9af96a8caef..65529db2750 100644
--- a/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/SegmentConfig.java
+++ b/src/core-metadata/src/main/java/org/apache/kylin/metadata/model/SegmentConfig.java
@@ -21,6 +21,7 @@
 import java.io.Serializable;
 import java.util.List;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import lombok.AllArgsConstructor;
@@ -49,6 +50,13 @@ public class SegmentConfig implements Serializable {
     @JsonProperty("create_empty_segment_enabled")
     private Boolean createEmptySegmentEnabled = false;
 
+    @JsonProperty("auto_segment_build")
+    private AutoSegmentBuildConfig autoSegmentBuild = new AutoSegmentBuildConfig();
+
+    @JsonProperty("auto_segment_build_enabled")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    private Boolean autoSegmentBuildEnabled;
+
     public boolean canSkipAutoMerge() {
         return !autoMergeEnabled;
     }
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/metadata/project/ProjectInstance.java b/src/core-metadata/src/main/java/org/apache/kylin/metadata/project/ProjectInstance.java
index 50c121dc453..db34118c406 100644
--- a/src/core-metadata/src/main/java/org/apache/kylin/metadata/project/ProjectInstance.java
+++ b/src/core-metadata/src/main/java/org/apache/kylin/metadata/project/ProjectInstance.java
@@ -41,6 +41,7 @@
 import org.apache.kylin.guava30.shaded.common.collect.Lists;
 import org.apache.kylin.metadata.MetadataConstants;
 import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
 import org.apache.kylin.metadata.model.ISourceAware;
 import org.apache.kylin.metadata.model.MaintainModelType;
 import org.apache.kylin.metadata.model.RetentionRange;
@@ -114,7 +115,7 @@ public class ProjectInstance extends RootPersistentEntity implements ISourceAwar
     @Setter
     private SegmentConfig segmentConfig = new SegmentConfig(false, Lists.newArrayList(AutoMergeTimeEnum.WEEK,
             AutoMergeTimeEnum.MONTH, AutoMergeTimeEnum.QUARTER, AutoMergeTimeEnum.YEAR), new VolatileRange(),
-            new RetentionRange(), false);
+            new RetentionRange(), false, new AutoSegmentBuildConfig(), false);
 
     public static ProjectInstance create(String name, String owner, String description,
             LinkedHashMap<String, String> overrideProps) {
diff --git a/src/core-metadata/src/test/java/org/apache/kylin/metadata/cube/model/NSegmentConfigHelperTest.java b/src/core-metadata/src/test/java/org/apache/kylin/metadata/cube/model/NSegmentConfigHelperTest.java
index f8d181a84fc..5b29bb0355b 100644
--- a/src/core-metadata/src/test/java/org/apache/kylin/metadata/cube/model/NSegmentConfigHelperTest.java
+++ b/src/core-metadata/src/test/java/org/apache/kylin/metadata/cube/model/NSegmentConfigHelperTest.java
@@ -63,7 +63,7 @@ public void testGetSegmentConfig() {
         // 2. MODEL_BASED && model segmentConfig is not empty, get mergedSegmentConfig of project segmentConfig and model SegmentConfig
         dataModelManager.updateDataModel(model, copyForWrite -> {
             copyForWrite.setSegmentConfig(
-                    new SegmentConfig(false, Lists.newArrayList(AutoMergeTimeEnum.WEEK), null, null, false));
+                    new SegmentConfig(false, Lists.newArrayList(AutoMergeTimeEnum.WEEK), null, null, false, null, null));
         });
         segmentConfig = NSegmentConfigHelper.getModelSegmentConfig(DEFAULT_PROJECT, model);
         Assert.assertEquals(false, segmentConfig.getAutoMergeEnabled());
diff --git a/src/data-loading-service/src/main/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentScheduler.java b/src/data-loading-service/src/main/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentScheduler.java
new file mode 100644
index 00000000000..ccb872b6841
--- /dev/null
+++ b/src/data-loading-service/src/main/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentScheduler.java
@@ -0,0 +1,447 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.rest.scheduler;
+
+import static org.apache.kylin.metadata.model.AutoSegmentBuildConfig.END_OF_DAY;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.time.format.ResolverStyle;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.guava30.shaded.common.cache.Cache;
+import org.apache.kylin.guava30.shaded.common.cache.CacheBuilder;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableManager;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.execution.JobTypeEnum;
+import org.apache.kylin.job.util.JobContextUtil;
+import org.apache.kylin.metadata.cube.model.NDataSegment;
+import org.apache.kylin.metadata.cube.model.NDataflowManager;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
+import org.apache.kylin.metadata.model.NDataModel;
+import org.apache.kylin.metadata.model.PartitionDesc;
+import org.apache.kylin.metadata.model.SegmentRange;
+import org.apache.kylin.metadata.model.SegmentStatusEnum;
+import org.apache.kylin.metadata.project.NProjectManager;
+import org.apache.kylin.metadata.project.ProjectInstance;
+import org.apache.kylin.rest.service.ModelBuildService;
+import org.apache.kylin.rest.service.params.IncrementBuildSegmentParams;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import lombok.val;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+public class AutoBuildSegmentScheduler {
+    private static final int MAX_TRACKED_CYCLES = 10_000;
+    private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss", Locale.ROOT)
+            .withResolverStyle(ResolverStyle.STRICT);
+
+    @Autowired
+    @Qualifier("modelBuildService")
+    private ModelBuildService modelBuildService;
+
+    private final AtomicBoolean dispatching = new AtomicBoolean(false);
+    // A completed cycle is re-evaluated after failover/restart, with segment metadata providing idempotency.
+    private final Cache<String, TargetState> cycleStates = CacheBuilder.newBuilder().maximumSize(MAX_TRACKED_CYCLES)
+            .expireAfterAccess(2, TimeUnit.DAYS).build();
+    // Avoid repeating the same malformed-metadata warning on every dispatcher scan.
+    private final Cache<String, String> invalidConfigWarnings = CacheBuilder.newBuilder()
+            .maximumSize(MAX_TRACKED_CYCLES).expireAfterAccess(1, TimeUnit.DAYS).build();
+
+    @Scheduled(fixedDelayString = "${kylin.model.auto-segment-build.dispatcher-interval-ms:30000}")
+    public void schedulerAutoBuildSegment() {
+        val currentTime = Instant.now();
+        if (!JobContextUtil.getJobContext(KylinConfig.getInstanceFromEnv()).getJobScheduler().isMaster()) {
+            return;
+        }
+        if (!dispatching.compareAndSet(false, true)) {
+            log.warn("Skip auto build segment dispatch because the previous dispatch is still running");
+            return;
+        }
+
+        try {
+            dispatch(currentTime);
+        } finally {
+            dispatching.set(false);
+        }
+    }
+
+    void dispatch(Instant currentTime) {
+        val systemConfig = KylinConfig.readSystemKylinConfig();
+        val projectManager = NProjectManager.getInstance(systemConfig);
+        for (ProjectInstance project : projectManager.listAllProjects()) {
+            try {
+                dispatchProject(systemConfig, project, currentTime);
+            } catch (Exception e) {
+                log.error("Auto build segment dispatch failed for project: {}", project.getName(), e);
+            }
+        }
+    }
+
+    private void dispatchProject(KylinConfig systemConfig, ProjectInstance project, Instant currentTime) {
+        if (project.getSegmentConfig() == null
+                || !Boolean.TRUE.equals(project.getSegmentConfig().getAutoSegmentBuildEnabled())) {
+            return;
+        }
+        val projectName = project.getName();
+        val zoneId = ZoneId.of(project.getConfig().getTimeZone());
+        val dataflowManager = NDataflowManager.getInstance(systemConfig, projectName);
+        for (NDataModel model : dataflowManager.listUnderliningDataModels()) {
+            try {
+                dispatchModel(projectName, model, zoneId, currentTime);
+            } catch (Exception e) {
+                log.error("Auto build segment dispatch failed, project: {}, model: {}", projectName, model.getUuid(),
+                        e);
+            }
+        }
+    }
+
+    private void dispatchModel(String project, NDataModel model, ZoneId zoneId, Instant currentTime) {
+        val autoSegmentBuild = getEligibleConfig(project, model);
+        if (autoSegmentBuild == null) {
+            return;
+        }
+
+        val scheduledTime = getLatestScheduledTime(autoSegmentBuild.getTriggerTime(), zoneId, currentTime);
+        if (scheduledTime.toInstant().isAfter(currentTime)) {
+            return;
+        }
+        dispatchTarget(project, model, autoSegmentBuild, scheduledTime);
+    }
+
+    private AutoSegmentBuildConfig getEligibleConfig(String project, NDataModel model) {
+        val configKey = project + "/" + model.getUuid();
+        if (model.isBroken() || model.isStreaming() || model.isMultiPartitionModel()
+                || PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc()) || model.getSegmentConfig() == null) {
+            invalidConfigWarnings.invalidate(configKey);
+            return null;
+        }
+        val autoSegmentBuild = model.getSegmentConfig().getAutoSegmentBuild();
+        if (autoSegmentBuild == null || !autoSegmentBuild.isEnabled()) {
+            invalidConfigWarnings.invalidate(configKey);
+            return null;
+        }
+
+        val invalidReason = getInvalidConfigReason(autoSegmentBuild);
+        if (invalidReason == null) {
+            invalidConfigWarnings.invalidate(configKey);
+            return autoSegmentBuild;
+        }
+        if (!StringUtils.equals(invalidReason, invalidConfigWarnings.getIfPresent(configKey))) {
+            log.warn("Skip auto build segment because of invalid config: {}, project: {}, model: {}", invalidReason,
+                    project, model.getUuid());
+            invalidConfigWarnings.put(configKey, invalidReason);
+        }
+        return null;
+    }
+
+    private String getInvalidConfigReason(AutoSegmentBuildConfig config) {
+        List<String> missingFields = Lists.newArrayList();
+        if (StringUtils.isBlank(config.getTriggerTime())) {
+            missingFields.add("trigger_time");
+        }
+        if (config.getLogicalDateOffsetDays() == null) {
+            missingFields.add("logical_date_offset_days");
+        }
+        if (StringUtils.isBlank(config.getDataRangeStartTime())) {
+            missingFields.add("data_range_start_time");
+        }
+        if (StringUtils.isBlank(config.getDataRangeEndTime())) {
+            missingFields.add("data_range_end_time");
+        }
+        if (!missingFields.isEmpty()) {
+            return "missing required field(s): " + StringUtils.join(missingFields, ", ");
+        }
+        if (config.getLogicalDateOffsetDays() < 1) {
+            return "logical_date_offset_days must be >= 1";
+        }
+
+        try {
+            LocalTime.parse(config.getTriggerTime(), TIME_FORMATTER);
+        } catch (DateTimeParseException e) {
+            return "invalid trigger_time, expected HH:mm:ss";
+        }
+
+        Pair<LocalTime, Boolean> start;
+        Pair<LocalTime, Boolean> end;
+        try {
+            start = parseTime(config.getDataRangeStartTime(), false);
+        } catch (DateTimeParseException e) {
+            return "invalid data_range_start_time, expected HH:mm:ss";
+        }
+        try {
+            end = parseTime(config.getDataRangeEndTime(), true);
+        } catch (DateTimeParseException e) {
+            return "invalid data_range_end_time, expected HH:mm:ss or " + END_OF_DAY;
+        }
+        if (!end.getSecond() && !start.getFirst().isBefore(end.getFirst())) {
+            return "data_range_start_time must be earlier than data_range_end_time";
+        }
+        return null;
+    }
+
+    ZonedDateTime getLatestScheduledTime(String triggerTime, ZoneId zoneId, Instant currentTime) {
+        val localCurrentTime = currentTime.atZone(zoneId);
+        val parsedTriggerTime = LocalTime.parse(triggerTime, TIME_FORMATTER);
+        ZonedDateTime scheduledTime = ZonedDateTime.of(localCurrentTime.toLocalDate(), parsedTriggerTime, zoneId);
+        if (scheduledTime.toInstant().isAfter(currentTime)) {
+            scheduledTime = scheduledTime.minusDays(1);
+        }
+        return scheduledTime;
+    }
+
+    private void dispatchTarget(String project, NDataModel model, AutoSegmentBuildConfig autoSegmentBuild,
+            ZonedDateTime scheduledTime) {
+        val modelId = model.getUuid();
+        val target = createBuildTarget(model, autoSegmentBuild, scheduledTime);
+        val cycleKey = createCycleKey(project, modelId, scheduledTime, target.range);
+        val previousState = cycleStates.getIfPresent(cycleKey);
+        if (previousState != null && previousState.isComplete()) {
+            return;
+        }
+
+        val dataflow = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project).getDataflow(modelId);
+        List<NDataSegment> segments = dataflow == null ? Lists.newArrayList() : dataflow.getSegments();
+        val jobs = getModelBuildJobs(project, modelId);
+        TargetState state = evaluateTargetState(segments, jobs, target.range);
+        if (state == TargetState.NO_OVERLAP) {
+            try {
+                submitJob(project, model, target);
+                state = TargetState.SUBMITTED;
+            } catch (Exception e) {
+                state = TargetState.RETRYABLE_FAILURE;
+                if (previousState != TargetState.RETRYABLE_FAILURE) {
+                    log.error("Auto build segment submit failed, project: {}, model: {}, range: {}", project, modelId,
+                            target.range, e);
+                } else {
+                    log.debug("Auto build segment submit still failing, project: {}, model: {}, range: {}", project,
+                            modelId, target.range, e);
+                }
+            }
+        }
+        recordCycleState(cycleKey, previousState, state, project, modelId, scheduledTime, target.range, segments);
+    }
+
+    private BuildTarget createBuildTarget(NDataModel model, AutoSegmentBuildConfig autoSegmentBuild,
+            ZonedDateTime scheduledTime) {
+        val zoneId = scheduledTime.getZone();
+        val logicalDate = scheduledTime.toLocalDate().minusDays(autoSegmentBuild.getLogicalDateOffsetDays());
+        val start = parseTime(autoSegmentBuild.getDataRangeStartTime(), false);
+        val end = parseTime(autoSegmentBuild.getDataRangeEndTime(), true);
+        val startDateTime = LocalDateTime.of(logicalDate, start.getFirst());
+        val endDate = end.getSecond() ? logicalDate.plusDays(1) : logicalDate;
+        val endDateTime = LocalDateTime.of(endDate, end.getFirst());
+        if (!startDateTime.isBefore(endDateTime)) {
+            throw new IllegalArgumentException("Auto build segment range start must be earlier than end");
+        }
+
+        val startMillis = String.valueOf(startDateTime.atZone(zoneId).toInstant().toEpochMilli());
+        val endMillis = String.valueOf(endDateTime.atZone(zoneId).toInstant().toEpochMilli());
+        val partitionDateFormat = model.getPartitionDesc().getPartitionDateFormat();
+        val normalizedStart = DateFormat.getFormatTimeStamp(startMillis, partitionDateFormat);
+        val normalizedEnd = DateFormat.getFormatTimeStamp(endMillis, partitionDateFormat);
+        if (normalizedStart >= normalizedEnd) {
+            throw new IllegalArgumentException(
+                    "Auto build segment range is empty after partition format normalization");
+        }
+        return new BuildTarget(startMillis, endMillis,
+                new SegmentRange.TimePartitionedSegmentRange(normalizedStart, normalizedEnd));
+    }
+
+    private String createCycleKey(String project, String modelId, ZonedDateTime scheduledTime,
+            SegmentRange<Long> targetRange) {
+        return project + "/" + modelId + "/" + scheduledTime.toInstant().toEpochMilli() + "/"
+                + targetRange.getStart() + "-" + targetRange.getEnd();
+    }
+
+    private void submitJob(String project, NDataModel model, BuildTarget target) throws Exception {
+        val params = new IncrementBuildSegmentParams(project, model.getUuid(), target.startMillis, target.endMillis,
+                model.getPartitionDesc(), model.getMultiPartitionDesc(), Lists.newArrayList(), true, null);
+        modelBuildService.incrementBuildSegmentsByScheduler(params);
+    }
+
+    List<AbstractExecutable> getModelBuildJobs(String project, String modelId) {
+        val executableManager = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
+        JobTypeEnum[] buildJobTypes = JobTypeEnum.getJobTypeByCategory(JobTypeEnum.Category.BUILD)
+                .toArray(new JobTypeEnum[0]);
+        return executableManager.listExecByModelAndStatus(modelId, ExecutableState::isRunning, buildJobTypes);
+    }
+
+    TargetState evaluateTargetState(List<NDataSegment> segments, List<AbstractExecutable> jobs,
+            SegmentRange<Long> targetRange) {
+        val coverageState = getCoverageState(segments, targetRange);
+        if (coverageState != null) {
+            return coverageState;
+        }
+
+        List<NDataSegment> buildingSegments = segments.stream()
+                .filter(segment -> segment.getStatus() == SegmentStatusEnum.NEW)
+                .filter(segment -> segment.getSegRange().overlaps(targetRange)).collect(Collectors.toList());
+        if (!buildingSegments.isEmpty()) {
+            boolean hasProgressingJob = false;
+            for (NDataSegment segment : buildingSegments) {
+                boolean hasRelatedJob = false;
+                for (AbstractExecutable job : jobs) {
+                    if (job.getTargetSegments() == null || !job.getTargetSegments().contains(segment.getId())) {
+                        continue;
+                    }
+                    val jobState = job.getStatusInMem();
+                    if (!jobState.isRunning()) {
+                        continue;
+                    }
+                    hasRelatedJob = true;
+                    if (!jobState.isProgressing()) {
+                        return TargetState.BLOCKED;
+                    }
+                    hasProgressingJob = true;
+                }
+                if (!hasRelatedJob) {
+                    return TargetState.BLOCKED;
+                }
+            }
+            return hasProgressingJob ? TargetState.IN_PROGRESS : TargetState.BLOCKED;
+        }
+
+        boolean hasAvailableOverlap = segments.stream()
+                .filter(segment -> segment.getStatus() == SegmentStatusEnum.READY
+                        || segment.getStatus() == SegmentStatusEnum.WARNING)
+                .anyMatch(segment -> segment.getSegRange().overlaps(targetRange));
+        return hasAvailableOverlap ? TargetState.PARTIAL_OVERLAP : TargetState.NO_OVERLAP;
+    }
+
+    private TargetState getCoverageState(List<NDataSegment> segments, SegmentRange<Long> targetRange) {
+        List<NDataSegment> availableSegments = new ArrayList<>();
+        for (NDataSegment segment : segments) {
+            if ((segment.getStatus() == SegmentStatusEnum.READY || segment.getStatus() == SegmentStatusEnum.WARNING)
+                    && segment.getSegRange().overlaps(targetRange)) {
+                availableSegments.add(segment);
+            }
+        }
+        availableSegments.sort(Comparator.comparingLong(segment -> (Long) segment.getSegRange().getStart()));
+
+        long coveredUntil = targetRange.getStart();
+        boolean includesWarning = false;
+        for (NDataSegment segment : availableSegments) {
+            long segmentStart = (Long) segment.getSegRange().getStart();
+            long segmentEnd = (Long) segment.getSegRange().getEnd();
+            if (segmentEnd <= coveredUntil) {
+                continue;
+            }
+            if (segmentStart > coveredUntil) {
+                break;
+            }
+            coveredUntil = Math.max(coveredUntil, segmentEnd);
+            includesWarning |= segment.getStatus() == SegmentStatusEnum.WARNING;
+            if (coveredUntil >= targetRange.getEnd()) {
+                return includesWarning ? TargetState.COVERED_WITH_WARNING : TargetState.COVERED;
+            }
+        }
+        return null;
+    }
+
+    private void recordCycleState(String cycleKey, TargetState previousState, TargetState state, String project,
+            String modelId, ZonedDateTime scheduledTime, SegmentRange<Long> targetRange,
+            List<NDataSegment> segments) {
+        cycleStates.put(cycleKey, state);
+        if (state == previousState) {
+            return;
+        }
+
+        val relatedSegments = segments.stream().filter(segment -> segment.getSegRange().overlaps(targetRange))
+                .map(segment -> segment.getId() + ":" + segment.getStatus() + segment.getSegRange())
+                .collect(Collectors.joining(",", "[", "]"));
+        switch (state) {
+        case COVERED:
+            log.info("Auto build segment already covered, project: {}, model: {}, range: {}", project, modelId,
+                    targetRange);
+            break;
+        case COVERED_WITH_WARNING:
+            log.warn("Auto build segment covered by warning segment, project: {}, model: {}, range: {}, segments: {}",
+                    project, modelId, targetRange, relatedSegments);
+            break;
+        case IN_PROGRESS:
+            log.info("Auto build segment waiting for overlapping build job, project: {}, model: {}, range: {}, "
+                    + "segments: {}", project, modelId, targetRange, relatedSegments);
+            break;
+        case BLOCKED:
+            log.warn("Auto build segment blocked by non-progressing job or orphan segment, project: {}, model: {}, "
+                    + "range: {}, segments: {}", project, modelId, targetRange, relatedSegments);
+            break;
+        case PARTIAL_OVERLAP:
+            log.warn("Auto build segment has a partial overlap, project: {}, model: {}, range: {}, segments: {}",
+                    project, modelId, targetRange, relatedSegments);
+            break;
+        case SUBMITTED:
+            log.info("Auto build segment submitted, project: {}, model: {}, scheduled time: {}, range: {}", project,
+                    modelId, scheduledTime, targetRange);
+            break;
+        default:
+            break;
+        }
+    }
+
+    private static class BuildTarget {
+        private final String startMillis;
+        private final String endMillis;
+        private final SegmentRange<Long> range;
+
+        BuildTarget(String startMillis, String endMillis, SegmentRange<Long> range) {
+            this.startMillis = startMillis;
+            this.endMillis = endMillis;
+            this.range = range;
+        }
+    }
+
+    enum TargetState {
+        NO_OVERLAP, COVERED, COVERED_WITH_WARNING, IN_PROGRESS, BLOCKED, PARTIAL_OVERLAP, SUBMITTED, RETRYABLE_FAILURE;
+
+        boolean isComplete() {
+            return this == COVERED || this == COVERED_WITH_WARNING;
+        }
+    }
+
+    private Pair<LocalTime, Boolean> parseTime(String time, boolean allow24Hour) {
+        if (allow24Hour && StringUtils.equals(time, END_OF_DAY)) {
+            return Pair.newPair(LocalTime.MIDNIGHT, true);
+        }
+        return Pair.newPair(LocalTime.parse(time, TIME_FORMATTER), false);
+    }
+}
diff --git a/src/data-loading-service/src/test/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentSchedulerTest.java b/src/data-loading-service/src/test/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentSchedulerTest.java
new file mode 100644
index 00000000000..c8deff1348a
--- /dev/null
+++ b/src/data-loading-service/src/test/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentSchedulerTest.java
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.rest.scheduler;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeParseException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.junit.annotation.MetadataInfo;
+import org.apache.kylin.metadata.cube.model.NDataSegment;
+import org.apache.kylin.metadata.cube.model.NDataflowManager;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
+import org.apache.kylin.metadata.model.NDataModel;
+import org.apache.kylin.metadata.model.NDataModelManager;
+import org.apache.kylin.metadata.model.SegmentRange;
+import org.apache.kylin.metadata.model.SegmentStatusEnum;
+import org.apache.kylin.metadata.project.NProjectManager;
+import org.apache.kylin.metadata.realization.RealizationStatusEnum;
+import org.apache.kylin.rest.service.ModelBuildService;
+import org.apache.kylin.rest.service.params.IncrementBuildSegmentParams;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import lombok.val;
+
+@MetadataInfo
+class AutoBuildSegmentSchedulerTest {
+    private static final String PROJECT = "default";
+    private static final String MODEL_ID = "89af4ee2-2cdb-4b07-b39e-4c29856309aa";
+
+    @Test
+    void testLatestScheduledTimeUsesProjectTimeZone() {
+        val scheduler = new AutoBuildSegmentScheduler();
+        val zoneId = ZoneId.of("Asia/Shanghai");
+        val currentTime = Instant.parse("2026-08-12T16:00:20Z");
+
+        assertEquals(ZonedDateTime.of(2026, 8, 13, 0, 0, 0, 0, zoneId),
+                scheduler.getLatestScheduledTime("00:00:00", zoneId, currentTime));
+        assertEquals(ZonedDateTime.of(2026, 8, 12, 0, 1, 0, 0, zoneId),
+                scheduler.getLatestScheduledTime("00:01:00", zoneId, currentTime));
+        assertThrows(DateTimeParseException.class,
+                () -> scheduler.getLatestScheduledTime("24:00:00", zoneId, currentTime));
+    }
+
+    @Test
+    void testDispatchCatchesUpLatestCycleAndBuildsConfiguredRange() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        Mockito.doReturn(Collections.emptyList()).when(scheduler).getModelBuildJobs(PROJECT, MODEL_ID);
+        Mockito.doReturn(AutoBuildSegmentScheduler.TargetState.NO_OVERLAP).when(scheduler)
+                .evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+        enableAutoSegmentBuild();
+
+        val projectConfig = NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig();
+        val zoneId = ZoneId.of(projectConfig.getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusHours(6).toInstant());
+
+        val paramsCaptor = ArgumentCaptor.forClass(IncrementBuildSegmentParams.class);
+        Mockito.verify(modelBuildService).incrementBuildSegmentsByScheduler(paramsCaptor.capture());
+        val logicalDate = scheduledTime.toLocalDate().minusDays(1);
+        assertEquals(String.valueOf(LocalDateTime.of(logicalDate, LocalTime.MIDNIGHT).atZone(zoneId).toInstant()
+                .toEpochMilli()), paramsCaptor.getValue().getStart());
+        assertEquals(String.valueOf(LocalDateTime.of(logicalDate.plusDays(1), LocalTime.MIDNIGHT).atZone(zoneId)
+                .toInstant().toEpochMilli()), paramsCaptor.getValue().getEnd());
+    }
+
+    @Test
+    void testCompletedCycleIsNotEvaluatedAgain() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        Mockito.doReturn(Collections.emptyList()).when(scheduler).getModelBuildJobs(PROJECT, MODEL_ID);
+        Mockito.doReturn(AutoBuildSegmentScheduler.TargetState.COVERED).when(scheduler)
+                .evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+        enableAutoSegmentBuild();
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(2026, 8, 12, 1, 0, 0, 0, zoneId);
+        scheduler.dispatch(scheduledTime.plusMinutes(1).toInstant());
+        scheduler.dispatch(scheduledTime.plusMinutes(2).toInstant());
+
+        Mockito.verifyNoInteractions(modelBuildService);
+        Mockito.verify(scheduler).evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+    }
+
+    @Test
+    void testBuildOfflineModel() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        Mockito.doReturn(Collections.emptyList()).when(scheduler).getModelBuildJobs(PROJECT, MODEL_ID);
+        Mockito.doReturn(AutoBuildSegmentScheduler.TargetState.NO_OVERLAP).when(scheduler)
+                .evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+        enableAutoSegmentBuild();
+        NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), PROJECT).updateDataflowStatus(MODEL_ID,
+                RealizationStatusEnum.OFFLINE);
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusSeconds(1).toInstant());
+
+        Mockito.verify(modelBuildService).incrementBuildSegmentsByScheduler(Mockito.any());
+    }
+
+    @Test
+    void testSkipProjectWhenAutoSegmentBuildDisabled() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        enableAutoSegmentBuild();
+        setProjectAutoSegmentBuildEnabled(false);
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusSeconds(1).toInstant());
+
+        Mockito.verifyNoInteractions(modelBuildService);
+    }
+
+    @Test
+    void testSkipIncompleteConfig() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        enableAutoSegmentBuild();
+        NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), PROJECT).updateDataModel(MODEL_ID,
+                copyForWrite -> copyForWrite.getSegmentConfig().getAutoSegmentBuild()
+                        .setLogicalDateOffsetDays(null));
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusSeconds(1).toInstant());
+
+        Mockito.verifyNoInteractions(modelBuildService);
+    }
+
+    @Test
+    void testTargetCoveredByExistingSegments() {
+        val scheduler = new AutoBuildSegmentScheduler();
+        val target = range(100L, 200L);
+
+        assertEquals(AutoBuildSegmentScheduler.TargetState.COVERED,
+                scheduler.evaluateTargetState(Collections.singletonList(segment("superset", 50L, 250L,
+                        SegmentStatusEnum.READY)), Collections.emptyList(), target));
+        assertEquals(AutoBuildSegmentScheduler.TargetState.COVERED,
+                scheduler.evaluateTargetState(Arrays.asList(segment("first", 100L, 150L, SegmentStatusEnum.READY),
+                        segment("second", 150L, 200L, SegmentStatusEnum.READY)), Collections.emptyList(), target));
+        assertEquals(AutoBuildSegmentScheduler.TargetState.COVERED_WITH_WARNING,
+                scheduler.evaluateTargetState(Collections.singletonList(segment("warning", 100L, 200L,
+                        SegmentStatusEnum.WARNING)), Collections.emptyList(), target));
+    }
+
+    @Test
+    void testProgressingJobIsCheckedByRange() {
+        val scheduler = new AutoBuildSegmentScheduler();
+        val target = range(100L, 200L);
+
+        NDataSegment larger = segment("larger", 50L, 250L, SegmentStatusEnum.NEW);
+        assertEquals(AutoBuildSegmentScheduler.TargetState.IN_PROGRESS, scheduler.evaluateTargetState(
+                Collections.singletonList(larger), Collections.singletonList(job(ExecutableState.RUNNING, "larger")),
+                target));
+
+        NDataSegment smaller = segment("smaller", 100L, 150L, SegmentStatusEnum.NEW);
+        assertEquals(AutoBuildSegmentScheduler.TargetState.IN_PROGRESS, scheduler.evaluateTargetState(
+                Collections.singletonList(smaller), Collections.singletonList(job(ExecutableState.PENDING, "smaller")),
+                target));
+
+        NDataSegment disjoint = segment("disjoint", 200L, 300L, SegmentStatusEnum.NEW);
+        assertEquals(AutoBuildSegmentScheduler.TargetState.NO_OVERLAP, scheduler.evaluateTargetState(
+                Collections.singletonList(disjoint),
+                Collections.singletonList(job(ExecutableState.RUNNING, "disjoint")), target));
+    }
+
+    @Test
+    void testErrorPausedAndOrphanSegmentsBlockTarget() {
+        val scheduler = new AutoBuildSegmentScheduler();
+        val target = range(100L, 200L);
+        NDataSegment building = segment("building", 100L, 200L, SegmentStatusEnum.NEW);
+
+        assertEquals(AutoBuildSegmentScheduler.TargetState.BLOCKED, scheduler.evaluateTargetState(
+                Collections.singletonList(building), Collections.singletonList(job(ExecutableState.ERROR, "building")),
+                target));
+        assertEquals(AutoBuildSegmentScheduler.TargetState.BLOCKED, scheduler.evaluateTargetState(
+                Collections.singletonList(building), Collections.singletonList(job(ExecutableState.PAUSED, "building")),
+                target));
+        assertEquals(AutoBuildSegmentScheduler.TargetState.BLOCKED, scheduler.evaluateTargetState(
+                Collections.singletonList(building), Collections.emptyList(), target));
+    }
+
+    @Test
+    void testPartialAvailableSegmentIsConflict() {
+        val scheduler = new AutoBuildSegmentScheduler();
+        val target = range(100L, 200L);
+        List<NDataSegment> segments = Collections
+                .singletonList(segment("partial", 100L, 150L, SegmentStatusEnum.READY));
+
+        assertEquals(AutoBuildSegmentScheduler.TargetState.PARTIAL_OVERLAP,
+                scheduler.evaluateTargetState(segments, Collections.emptyList(), target));
+    }
+
+    @Test
+    void testInProgressCycleIsReevaluated() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        Mockito.doReturn(Collections.emptyList()).when(scheduler).getModelBuildJobs(PROJECT, MODEL_ID);
+        Mockito.doReturn(AutoBuildSegmentScheduler.TargetState.IN_PROGRESS,
+                AutoBuildSegmentScheduler.TargetState.NO_OVERLAP).when(scheduler)
+                .evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+        enableAutoSegmentBuild();
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusMinutes(1).toInstant());
+        scheduler.dispatch(scheduledTime.plusMinutes(2).toInstant());
+
+        Mockito.verify(modelBuildService).incrementBuildSegmentsByScheduler(Mockito.any());
+    }
+
+    @Test
+    void testRetryableSubmissionFailureIsReevaluated() throws Exception {
+        val modelBuildService = Mockito.mock(ModelBuildService.class);
+        val scheduler = Mockito.spy(new AutoBuildSegmentScheduler());
+        ReflectionTestUtils.setField(scheduler, "modelBuildService", modelBuildService);
+        Mockito.doReturn(Collections.emptyList()).when(scheduler).getModelBuildJobs(PROJECT, MODEL_ID);
+        Mockito.doReturn(AutoBuildSegmentScheduler.TargetState.NO_OVERLAP).when(scheduler)
+                .evaluateTargetState(Mockito.anyList(), Mockito.anyList(), Mockito.any());
+        Mockito.doThrow(new IllegalStateException("transient failure")).doReturn(null).when(modelBuildService)
+                .incrementBuildSegmentsByScheduler(Mockito.any());
+        enableAutoSegmentBuild();
+
+        val zoneId = ZoneId.of(NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).getProject(PROJECT)
+                .getConfig().getTimeZone());
+        val scheduledTime = ZonedDateTime.of(LocalDate.of(2026, 8, 12), LocalTime.of(1, 0), zoneId);
+        scheduler.dispatch(scheduledTime.plusMinutes(1).toInstant());
+        scheduler.dispatch(scheduledTime.plusMinutes(2).toInstant());
+
+        Mockito.verify(modelBuildService, Mockito.times(2)).incrementBuildSegmentsByScheduler(Mockito.any());
+    }
+
+    private SegmentRange<Long> range(long start, long end) {
+        return new SegmentRange.TimePartitionedSegmentRange(start, end);
+    }
+
+    private NDataSegment segment(String id, long start, long end, SegmentStatusEnum status) {
+        NDataSegment segment = Mockito.mock(NDataSegment.class);
+        Mockito.when(segment.getId()).thenReturn(id);
+        Mockito.when(segment.getStatus()).thenReturn(status);
+        Mockito.when(segment.getSegRange()).thenReturn(range(start, end));
+        return segment;
+    }
+
+    private AbstractExecutable job(ExecutableState state, String... segmentIds) {
+        AbstractExecutable job = Mockito.mock(AbstractExecutable.class);
+        Mockito.when(job.getStatusInMem()).thenReturn(state);
+        Mockito.when(job.getTargetSegments()).thenReturn(Arrays.asList(segmentIds));
+        return job;
+    }
+
+    private NDataModel enableAutoSegmentBuild() {
+        setProjectAutoSegmentBuildEnabled(true);
+        val modelManager = NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), PROJECT);
+        return modelManager.updateDataModel(MODEL_ID, copyForWrite -> {
+            AutoSegmentBuildConfig config = new AutoSegmentBuildConfig();
+            config.setEnabled(true);
+            config.setTriggerTime("01:00:00");
+            config.setLogicalDateOffsetDays(1);
+            config.setDataRangeStartTime("00:00:00");
+            config.setDataRangeEndTime(AutoSegmentBuildConfig.END_OF_DAY);
+            copyForWrite.getSegmentConfig().setAutoSegmentBuild(config);
+        });
+    }
+
+    private void setProjectAutoSegmentBuildEnabled(boolean enabled) {
+        NProjectManager.getInstance(KylinConfig.getInstanceFromEnv()).updateProject(PROJECT,
+                copyForWrite -> copyForWrite.getSegmentConfig().setAutoSegmentBuildEnabled(enabled));
+    }
+}
diff --git a/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/open/OpenModelController.java b/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/open/OpenModelController.java
index a397c6a6059..4aadbeca874 100644
--- a/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/open/OpenModelController.java
+++ b/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/open/OpenModelController.java
@@ -619,6 +619,7 @@ private ModelConfigRequest newModelConfigRequestByConfig(ModelConfigResponse mod
         modelConfigRequest.setRetentionRange(modelConfig.getRetentionRange());
         modelConfigRequest.setVolatileRange(modelConfig.getVolatileRange());
         modelConfigRequest.setAutoMergeTimeRanges(modelConfig.getAutoMergeTimeRanges());
+        modelConfigRequest.setAutoSegmentBuild(modelConfig.getAutoSegmentBuild());
         modelConfigRequest.setOverrideProps(modelConfig.getOverrideProps());
         return modelConfigRequest;
     }
diff --git a/src/metadata-server/src/test/java/org/apache/kylin/rest/controller/NProjectControllerTest.java b/src/metadata-server/src/test/java/org/apache/kylin/rest/controller/NProjectControllerTest.java
index 996fd65ffe5..cd04cf4f3fa 100644
--- a/src/metadata-server/src/test/java/org/apache/kylin/rest/controller/NProjectControllerTest.java
+++ b/src/metadata-server/src/test/java/org/apache/kylin/rest/controller/NProjectControllerTest.java
@@ -328,6 +328,7 @@ public void testUpdateSegmentConfig() throws Exception {
         request.setAutoMergeEnabled(true);
         request.setAutoMergeTimeRanges(Arrays.asList(AutoMergeTimeEnum.DAY));
         request.setCreateEmptySegmentEnabled(true);
+        request.setAutoSegmentBuildEnabled(true);
 
         Mockito.doNothing().when(projectService).updateSegmentConfig("default", request);
         mockMvc.perform(MockMvcRequestBuilders.put("/api/projects/{project}/segment_config", "default")
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/request/ModelConfigRequest.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/request/ModelConfigRequest.java
index 0b05af4e1f8..e8d9ec34926 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/request/ModelConfigRequest.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/request/ModelConfigRequest.java
@@ -24,6 +24,7 @@
 import org.apache.kylin.guava30.shaded.common.collect.Maps;
 import org.apache.kylin.metadata.insensitive.ProjectInsensitiveRequest;
 import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
 import org.apache.kylin.metadata.model.RetentionRange;
 import org.apache.kylin.metadata.model.VolatileRange;
 
@@ -44,6 +45,8 @@ public class ModelConfigRequest implements ProjectInsensitiveRequest {
     private VolatileRange volatileRange;
     @JsonProperty("retention_range")
     private RetentionRange retentionRange;
+    @JsonProperty("auto_segment_build")
+    private AutoSegmentBuildConfig autoSegmentBuild;
     @JsonProperty("override_props")
     LinkedHashMap<String, String> overrideProps = Maps.newLinkedHashMap();
 }
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/ModelConfigResponse.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/ModelConfigResponse.java
index b2e67171e79..4b6137fe781 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/ModelConfigResponse.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/ModelConfigResponse.java
@@ -23,6 +23,7 @@
 
 import org.apache.kylin.guava30.shaded.common.collect.Maps;
 import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
 import org.apache.kylin.metadata.model.RetentionRange;
 import org.apache.kylin.metadata.model.VolatileRange;
 
@@ -48,6 +49,8 @@ public class ModelConfigResponse {
     private VolatileRange volatileRange;
     @JsonProperty("retention_range")
     private RetentionRange retentionRange;
+    @JsonProperty("auto_segment_build")
+    private AutoSegmentBuildConfig autoSegmentBuild;
     @JsonProperty("config_last_modifier")
     private String configLastModifier;
     @JsonProperty("config_last_modified")
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelBuildService.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelBuildService.java
index 89e897a8058..7b4451ecb67 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelBuildService.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelBuildService.java
@@ -103,6 +103,7 @@
 @Component("modelBuildService")
 public class ModelBuildService extends AbstractModelService implements ModelBuildSupporter {
 
+    private static final String SCHEDULER_SUBMITTER = "System";
     private static final Logger logger = LoggerFactory.getLogger(ModelBuildService.class);
     @Autowired
     private ModelService modelService;
@@ -281,9 +282,20 @@ public JobInfoResponse incrementBuildSegmentsManually(String project, String mod
 
     @Override
     public JobInfoResponse incrementBuildSegmentsManually(IncrementBuildSegmentParams params) throws Exception {
+        return incrementBuildSegmentsInternal(params, getUsername(), true);
+    }
+
+    public JobInfoResponse incrementBuildSegmentsByScheduler(IncrementBuildSegmentParams params) throws Exception {
+        return incrementBuildSegmentsInternal(params, SCHEDULER_SUBMITTER, false);
+    }
+
+    private JobInfoResponse incrementBuildSegmentsInternal(IncrementBuildSegmentParams params, String submitter,
+            boolean checkPermission) throws Exception {
         String project = params.getProject();
-        aclEvaluate.checkProjectOperationPermission(project);
-        checkModelPermission(project, params.getModelId());
+        if (checkPermission) {
+            aclEvaluate.checkProjectOperationPermission(project);
+            checkModelPermission(project, params.getModelId());
+        }
         val modelManager = getManager(NDataModelManager.class, project);
         if (PartitionDesc.isEmptyPartitionDesc(params.getPartitionDesc())) {
             throw new KylinException(EMPTY_PARTITION_COLUMN, "Partition column is null.'");
@@ -320,7 +332,7 @@ public JobInfoResponse incrementBuildSegmentsManually(IncrementBuildSegmentParam
                 .withTag(params.getTag());
 
         List<JobInfoResponse.JobInfo> jobIds = EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
-            List<JobParam> paramList = createSegmentsAndJobParams(buildSegmentParams);
+            List<JobParam> paramList = createSegmentsAndJobParams(buildSegmentParams, submitter);
             return createJob(paramList);
         }, project);
 
@@ -342,6 +354,11 @@ private List<JobInfoResponse.JobInfo> createJob(List<JobParam> jobParamList) {
     }
 
     public List<JobParam> createSegmentsAndJobParams(IncrementBuildSegmentParams params) throws IOException {
+        return createSegmentsAndJobParams(params, getUsername());
+    }
+
+    public List<JobParam> createSegmentsAndJobParams(IncrementBuildSegmentParams params, String submitter)
+            throws IOException {
         modelService.checkModelAndIndexManually(params);
         if (CollectionUtils.isEmpty(params.getSegmentHoles())) {
             params.setSegmentHoles(Lists.newArrayList());
@@ -374,7 +391,7 @@ public List<JobParam> createSegmentsAndJobParams(IncrementBuildSegmentParams par
                     .withBatchIndexIds(params.getBatchIndexIds()).withYarnQueue(params.getYarnQueue())
                     .withTag(params.getTag());
             NDataSegment segment = createSegment(relParams);
-            res.add(createJobParam(relParams, segment));
+            res.add(createJobParam(relParams, segment, submitter));
         }
         IncrementBuildSegmentParams relParams = new IncrementBuildSegmentParams(params.getProject(),
                 params.getModelId(), params.getStart(), params.getEnd(), params.getPartitionColFormat(),
@@ -386,18 +403,22 @@ public List<JobParam> createSegmentsAndJobParams(IncrementBuildSegmentParams par
                 .withBatchIndexIds(params.getBatchIndexIds()).withYarnQueue(params.getYarnQueue())
                 .withTag(params.getTag());
         NDataSegment segment = createSegment(relParams);
-        res.add(createJobParam(relParams, segment));
+        res.add(createJobParam(relParams, segment, submitter));
         return res;
     }
 
     public JobParam createJobParam(IncrementBuildSegmentParams params, NDataSegment segment) {
+        return createJobParam(params, segment, getUsername());
+    }
+
+    public JobParam createJobParam(IncrementBuildSegmentParams params, NDataSegment segment, String submitter) {
         if (!params.isNeedBuild()) {
             return null;
         }
         String project = params.getProject();
         String modelId = params.getModelId();
         NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
-        JobParam jobParam = new JobParam(segment, modelId, getUsername())
+        JobParam jobParam = new JobParam(segment, modelId, submitter)
                 .withIgnoredSnapshotTables(params.getIgnoredSnapshotTables()).withPriority(params.getPriority())
                 .withYarnQueue(params.getYarnQueue()).withTag(params.getTag()).withProject(project);
         addJobParamExtParams(jobParam, params);
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
index 33f9c643164..6a7a15ffc67 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
@@ -63,6 +63,7 @@
 import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_NOT_EXIST_ID;
 import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_NOT_EXIST_NAME;
 import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_STATUS;
+import static org.apache.kylin.metadata.model.AutoSegmentBuildConfig.END_OF_DAY;
 import static org.apache.kylin.metadata.model.FunctionDesc.PARAMETER_TYPE_COLUMN;
 
 import java.io.IOException;
@@ -70,6 +71,10 @@
 import java.sql.SQLException;
 import java.text.MessageFormat;
 import java.text.SimpleDateFormat;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.time.format.ResolverStyle;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -158,6 +163,7 @@
 import org.apache.kylin.metadata.cube.model.RuleBasedIndex;
 import org.apache.kylin.metadata.model.AntiFlatChecker;
 import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
 import org.apache.kylin.metadata.model.ColumnDesc;
 import org.apache.kylin.metadata.model.ComputedColumnDesc;
 import org.apache.kylin.metadata.model.DataCheckDesc;
@@ -285,7 +291,8 @@ public class ModelService extends AbstractModelService implements TableModelSupp
 
     private static final List<String> MODEL_CONFIG_BLOCK_LIST = Lists.newArrayList("kylin.index.rule-scheduler-data");
     private static final Set<String> STRING_TYPE_SET = Sets.newHashSet("STRING", "CHAR", "VARCHAR");
-
+    private static final DateTimeFormatter SEGMENT_AUTO_BUILD_TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss",
+            Locale.ROOT).withResolverStyle(ResolverStyle.STRICT);
     //The front-end supports only the following formats
     private static final List<String> SUPPORTED_FORMATS = ImmutableList.of("ZZ", "DD", "D", "Do", "dddd", "ddd", "dd", //
             "d", "MMM", "MM", "M", "yyyy", "yy", "hh", "hh", "h", "HH", "H", "m", "mm", "ss", "s", "SSS", "SS", "S", //
@@ -851,13 +858,14 @@ private ModelStatusToDisplayEnum convertFusionModelStatusToDisplay(NDataModel mo
                 inconsistentBatchSegmentCount);
         if (!batchModel.isBroken() && !modelDesc.isBroken()) {
             switch (modelResponseStatus) {
-            case ONLINE:
-                return (batchModelResponseStatus == ModelStatusToDisplayEnum.WARNING ? ModelStatusToDisplayEnum.WARNING
-                        : modelResponseStatus);
-            case OFFLINE:
-                return batchModelResponseStatus;
-            default:
-                return modelResponseStatus;
+                case ONLINE:
+                    return (batchModelResponseStatus == ModelStatusToDisplayEnum.WARNING
+                            ? ModelStatusToDisplayEnum.WARNING
+                            : modelResponseStatus);
+                case OFFLINE:
+                    return batchModelResponseStatus;
+                default:
+                    return modelResponseStatus;
             }
         } else {
             return modelResponseStatus;
@@ -3261,6 +3269,7 @@ public List<ModelConfigResponse> getModelConfig(String project, String modelName
                     response.setAutoMergeTimeRanges(segmentConfig.getAutoMergeTimeRanges());
                     response.setVolatileRange(segmentConfig.getVolatileRange());
                     response.setRetentionRange(segmentConfig.getRetentionRange());
+                    response.setAutoSegmentBuild(segmentConfig.getAutoSegmentBuild());
                     response.setConfigLastModified(dataModel.getConfigLastModified());
                     response.setConfigLastModifier(dataModel.getConfigLastModifier());
                     val indexPlan = getIndexPlan(dataModel.getUuid(), project);
@@ -3277,7 +3286,7 @@ public List<ModelConfigResponse> getModelConfig(String project, String modelName
     @Transaction(project = 0)
     public void updateModelConfig(String project, String modelId, ModelConfigRequest request) {
         aclEvaluate.checkProjectWritePermission(project);
-        checkModelConfigParameters(request);
+        checkModelConfigParameters(project, modelId, request);
         val dataModelManager = getManager(NDataModelManager.class, project);
         dataModelManager.updateDataModel(modelId, copyForWrite -> {
             val segmentConfig = copyForWrite.getSegmentConfig();
@@ -3285,6 +3294,7 @@ public void updateModelConfig(String project, String modelId, ModelConfigRequest
             segmentConfig.setAutoMergeTimeRanges(request.getAutoMergeTimeRanges());
             segmentConfig.setVolatileRange(request.getVolatileRange());
             segmentConfig.setRetentionRange(request.getRetentionRange());
+            segmentConfig.setAutoSegmentBuild(request.getAutoSegmentBuild());
 
             copyForWrite.setConfigLastModified(System.currentTimeMillis());
             copyForWrite.setConfigLastModifier(BasicService.getUsername());
@@ -3352,10 +3362,16 @@ private void checkPropParameter(ModelConfigRequest request) {
 
     @VisibleForTesting
     public void checkModelConfigParameters(ModelConfigRequest request) {
+        checkModelConfigParameters(request.getProject(), null, request);
+    }
+
+    @VisibleForTesting
+    public void checkModelConfigParameters(String project, String modelId, ModelConfigRequest request) {
         Boolean autoMergeEnabled = request.getAutoMergeEnabled();
         List<AutoMergeTimeEnum> timeRanges = request.getAutoMergeTimeRanges();
         VolatileRange volatileRange = request.getVolatileRange();
         RetentionRange retentionRange = request.getRetentionRange();
+        AutoSegmentBuildConfig autoSegmentBuild = request.getAutoSegmentBuild();
 
         if (Boolean.TRUE.equals(autoMergeEnabled) && (null == timeRanges || timeRanges.isEmpty())) {
             throw new KylinException(INVALID_PARAMETER, MsgPicker.getMsg().getInvalidAutoMergeConfig());
@@ -3368,9 +3384,64 @@ public void checkModelConfigParameters(ModelConfigRequest request) {
                 && retentionRange.getRetentionRangeNumber() < 0) {
             throw new KylinException(INVALID_PARAMETER, MsgPicker.getMsg().getInvalidRetentionRangeConfig());
         }
+        if (null != autoSegmentBuild && autoSegmentBuild.isEnabled()) {
+            checkAutoSegmentBuildConfig(project, modelId, autoSegmentBuild);
+        }
         checkPropParameter(request);
     }
 
+    private void checkAutoSegmentBuildConfig(String project, String modelId, AutoSegmentBuildConfig autoSegmentBuild) {
+        if (StringUtils.isBlank(autoSegmentBuild.getTriggerTime())
+                || StringUtils.isBlank(autoSegmentBuild.getDataRangeStartTime())
+                || StringUtils.isBlank(autoSegmentBuild.getDataRangeEndTime())
+                || autoSegmentBuild.getLogicalDateOffsetDays() == null) {
+            throw new KylinException(INVALID_PARAMETER,
+                    "auto_segment_build requires trigger_time, logical_date_offset_days, data_range_start_time "
+                            + "and data_range_end_time.");
+        }
+        if (autoSegmentBuild.getLogicalDateOffsetDays() < 1) {
+            throw new KylinException(INVALID_PARAMETER, "logical_date_offset_days must be >= 1.");
+        }
+        parseSegmentBuildTime(autoSegmentBuild.getTriggerTime(), "trigger_time", false);
+        LocalTime startTime = parseSegmentBuildTime(autoSegmentBuild.getDataRangeStartTime(), "data_range_start_time",
+                false);
+        boolean endOfNextDay = END_OF_DAY.equals(autoSegmentBuild.getDataRangeEndTime());
+        LocalTime endTime = parseSegmentBuildTime(autoSegmentBuild.getDataRangeEndTime(), "data_range_end_time", true);
+        if (!endOfNextDay && !startTime.isBefore(endTime)) {
+            throw new KylinException(INVALID_PARAMETER,
+                    "data_range_start_time must be earlier than data_range_end_time.");
+        }
+        if (StringUtils.isBlank(project) || StringUtils.isBlank(modelId)) {
+            return;
+        }
+        val model = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
+        if (model == null) {
+            throw new KylinException(MODEL_ID_NOT_EXIST, modelId);
+        }
+        if (model.isStreaming()) {
+            throw new KylinException(INVALID_PARAMETER, "auto_segment_build is not supported for streaming model.");
+        }
+        if (model.isMultiPartitionModel()) {
+            throw new KylinException(INVALID_PARAMETER,
+                    "auto_segment_build is not supported for multi partition model.");
+        }
+        if (PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc())) {
+            throw new KylinException(INVALID_PARAMETER, "auto_segment_build requires a valid partition column.");
+        }
+    }
+
+    private LocalTime parseSegmentBuildTime(String value, String fieldName, boolean allow24Hour) {
+        if (allow24Hour && StringUtils.equals(value, END_OF_DAY)) {
+            return LocalTime.MIDNIGHT;
+        }
+        try {
+            return LocalTime.parse(value, SEGMENT_AUTO_BUILD_TIME_FORMATTER);
+        } catch (DateTimeParseException e) {
+            throw new KylinException(INVALID_PARAMETER,
+                    String.format(Locale.ROOT, "Invalid %s, expected HH:mm:ss.", fieldName));
+        }
+    }
+
     public ExistedDataRangeResponse getLatestDataRange(String project, String modelId, PartitionDesc desc) {
         aclEvaluate.checkProjectReadPermission(project);
         Preconditions.checkNotNull(modelId);
diff --git a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
index 55e19c629eb..17611630989 100644
--- a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
+++ b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java
@@ -125,6 +125,7 @@
 import org.apache.kylin.metadata.cube.model.RuleBasedIndex;
 import org.apache.kylin.metadata.cube.optimization.FrequencyMap;
 import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
+import org.apache.kylin.metadata.model.AutoSegmentBuildConfig;
 import org.apache.kylin.metadata.model.BadModelException;
 import org.apache.kylin.metadata.model.BadModelException.CauseType;
 import org.apache.kylin.metadata.model.ColumnDesc;
@@ -2788,6 +2789,13 @@ public void testUpdateAndGetModelConfig() {
         modelConfigRequest.setProject(project);
         modelConfigRequest.setAutoMergeEnabled(false);
         modelConfigRequest.setAutoMergeTimeRanges(Lists.newArrayList(AutoMergeTimeEnum.WEEK));
+        AutoSegmentBuildConfig autoSegmentBuildConfig = new AutoSegmentBuildConfig();
+        autoSegmentBuildConfig.setEnabled(true);
+        autoSegmentBuildConfig.setTriggerTime("01:00:00");
+        autoSegmentBuildConfig.setLogicalDateOffsetDays(1);
+        autoSegmentBuildConfig.setDataRangeStartTime("00:00:00");
+        autoSegmentBuildConfig.setDataRangeEndTime(AutoSegmentBuildConfig.END_OF_DAY);
+        modelConfigRequest.setAutoSegmentBuild(autoSegmentBuildConfig);
         modelService.updateModelConfig(project, model, modelConfigRequest);
 
         var modelConfigResponses = modelService.getModelConfig(project, null);
@@ -2795,6 +2803,8 @@ public void testUpdateAndGetModelConfig() {
             if (modelConfigResponse.getModel().equals(model)) {
                 Assert.assertEquals(false, modelConfigResponse.getAutoMergeEnabled());
                 Assert.assertEquals(1, modelConfigResponse.getAutoMergeTimeRanges().size());
+                Assert.assertNotNull(modelConfigResponse.getAutoSegmentBuild());
+                Assert.assertTrue(modelConfigResponse.getAutoSegmentBuild().isEnabled());
             }
         });
 
@@ -4330,6 +4340,114 @@ public void testCheckModelConfigParameters() {
         checkPropParameter(request);
     }
 
+    @Test
+    public void testCheckModelConfigParameters_AutoSegmentBuildInvalidConfig() {
+        ModelConfigRequest request = new ModelConfigRequest();
+        AutoSegmentBuildConfig config = new AutoSegmentBuildConfig();
+        config.setEnabled(true);
+        request.setAutoSegmentBuild(config);
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("trigger_time"));
+            Assert.assertTrue(e.getMessage().contains("logical_date_offset_days"));
+            Assert.assertTrue(e.getMessage().contains("data_range_start_time"));
+            Assert.assertTrue(e.getMessage().contains("data_range_end_time"));
+        }
+
+        config.setTriggerTime("01:00:00");
+        config.setLogicalDateOffsetDays(0);
+        config.setDataRangeStartTime("00:00:00");
+        config.setDataRangeEndTime(AutoSegmentBuildConfig.END_OF_DAY);
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("logical_date_offset_days"));
+        }
+
+        config.setLogicalDateOffsetDays(1);
+        config.setTriggerTime("25:00:00");
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("trigger_time"));
+        }
+
+        config.setTriggerTime("24:00:00");
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("trigger_time"));
+        }
+
+        config.setTriggerTime("01:00:00");
+        config.setDataRangeStartTime("24:00:00");
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("data_range_start_time"));
+        }
+
+        config.setDataRangeStartTime("10:00:00");
+        config.setDataRangeEndTime("09:00:00");
+        try {
+            modelService.checkModelConfigParameters(request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("data_range_start_time"));
+        }
+
+        config.setDataRangeStartTime("00:00:00");
+        config.setDataRangeEndTime(AutoSegmentBuildConfig.END_OF_DAY);
+        modelService.checkModelConfigParameters(request);
+    }
+
+    @Test
+    public void testCheckModelConfigParameters_AutoSegmentBuildModelConstraints() {
+        ModelConfigRequest request = new ModelConfigRequest();
+        AutoSegmentBuildConfig config = new AutoSegmentBuildConfig();
+        config.setEnabled(true);
+        config.setTriggerTime("01:00:00");
+        config.setLogicalDateOffsetDays(1);
+        config.setDataRangeStartTime("00:00:00");
+        config.setDataRangeEndTime(AutoSegmentBuildConfig.END_OF_DAY);
+        request.setAutoSegmentBuild(config);
+
+        NDataModel streamingModel = NDataModelManager.getInstance(getTestConfig(), "streaming_test").listAllModels()
+                .stream().filter(NDataModel::isStreaming).findFirst().orElse(null);
+        Assert.assertNotNull(streamingModel);
+        try {
+            modelService.checkModelConfigParameters("streaming_test", streamingModel.getId(), request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("streaming"));
+        }
+
+        NDataModel noPartitionModel = NDataModelManager.getInstance(getTestConfig(), "default").listAllModels()
+                .stream().filter(model -> PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc())).findFirst()
+                .orElse(null);
+        Assert.assertNotNull(noPartitionModel);
+        try {
+            modelService.checkModelConfigParameters("default", noPartitionModel.getId(), request);
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e instanceof KylinException);
+            Assert.assertTrue(e.getMessage().contains("partition"));
+        }
+    }
+
     @Test
     public void testBatchUpdateMultiPartition() {
         val modelId = "b780e4e4-69af-449e-b09f-05c90dfa04b6";
diff --git a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ProjectServiceTest.java b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ProjectServiceTest.java
index 4745d6c63ed..6cc4e4a2c55 100644
--- a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ProjectServiceTest.java
+++ b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ProjectServiceTest.java
@@ -474,9 +474,11 @@ public void testUpdateProjectConfig() throws IOException {
         val segmentConfigRequest = new SegmentConfigRequest();
         segmentConfigRequest.setAutoMergeEnabled(false);
         segmentConfigRequest.setAutoMergeTimeRanges(Collections.singletonList(AutoMergeTimeEnum.DAY));
+        segmentConfigRequest.setAutoSegmentBuildEnabled(true);
         projectService.updateSegmentConfig(project, segmentConfigRequest);
         response = projectService.getProjectConfig(project);
         Assert.assertFalse(response.isAutoMergeEnabled());
+        Assert.assertTrue(response.isAutoSegmentBuildEnabled());
 
         val pushDownConfigRequest = new PushDownConfigRequest();
         pushDownConfigRequest.setPushDownEnabled(false);
@@ -835,6 +837,7 @@ public void testResetProjectConfig() {
 
         response = projectService.resetProjectConfig(PROJECT, "segment_config");
         Assert.assertFalse(response.isAutoMergeEnabled());
+        Assert.assertFalse(response.isAutoSegmentBuildEnabled());
         Assert.assertEquals(4, response.getAutoMergeTimeRanges().size());
 
         response = projectService.resetProjectConfig(PROJECT, "storage_quota_config");