fix(compactor): serialize the dynamic merge of a day with its hourly jobs

The 00:00 job of a closed day now waits in run_merge until no other job of
the day is pending or running, checked in file_list_jobs, so it never merges
the hours another job is merging. This replaces the per-node routing and the
in-process running guard, which an open-day job of the same stream could
clear while the day job ran.

Old-data jobs of a day already merged by day go to the day's 00:00 job, so
late data joins DD/00 instead of leaving an hourly file next to the merged
day. The day job either merges the day into DD/00 or leaves it alone; the
hour-00 fallback that merged hour 00 from the day job is gone.

src/compaction/src/lib.rs

@@ -325,10 +325,16 @@ pub async fn run_merge(job_tx: mpsc::Sender<worker::MergeJob>) -> Result<(), any
             need_done_ids.push(job.id); // the data will be deleted by retention, just skip
             continue;
         }
-        // every job of a dynamic stream's closed day reads the files the day's job merges
-        let dynamic_day = is_past_day(job.offsets)
-            && get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.clone()));
-        if partition_time_level == PartitionTimeLevel::Daily || dynamic_day {
+        // the day's merge job reads every hour of the day: the day's other jobs go first
+        if is_dynamic_day_job(
+            job.offsets,
+            get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.as_ref())),
+        ) && day_jobs_unfinished(&org_id, stream_type, &stream_name, job.offsets).await
+        {
+            need_release_ids.push(job.id);
+            continue;
+        }
+        if partition_time_level == PartitionTimeLevel::Daily {
             // check if this stream need process by this node
             let Some(node_name) =
                 get_node_from_consistent_hash(&stream_name, &Role::Compactor, None).await
@@ -460,6 +466,30 @@ pub async fn run_delay_deletion() -> Result<(), anyhow::Error> {
     Ok(())
 }
 
+/// Whether a job of the day past its 00:00 is pending or running; a failed lookup counts as one.
+async fn day_jobs_unfinished(
+    org_id: &str,
+    stream_type: StreamType,
+    stream_name: &str,
+    day_start: i64,
+) -> bool {
+    let range = (day_start + 1, day_start + day_micros(1) - 1);
+    match infra_file_list::has_unfinished_jobs(org_id, stream_type, stream_name, range).await {
+        Ok(unfinished) => unfinished,
+        Err(e) => {
+            log::error!(
+                "[COMPACTOR] has_unfinished_jobs [{org_id}/{stream_type}/{stream_name}] failed: {e}"
+            );
+            true
+        }
+    }
+}
+
+/// The 00:00 job of an enabled stream's closed day: the one that merges the day into `DD/00`.
+pub(crate) fn is_dynamic_day_job(offset: i64, dynamic_merge: bool) -> bool {
+    dynamic_merge && offset % day_micros(1) == 0 && is_past_day(offset)
+}
+
 /// A day is closed once its last hour is past: no hour of it can receive files any more.
 pub(crate) fn is_past_day(offset: i64) -> bool {
     let day = day_micros(1);
@@ -472,3 +502,33 @@ pub(crate) fn is_past_hour(offset: i64) -> bool {
     let offset_hour_end = offset - offset % hour + hour;
     now - offset_hour_end > second_micros(get_config().limit.max_file_retention_time as i64) * 3
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The day's merge job waits for the day's other jobs, never for its own row or a done one.
+    #[tokio::test]
+    async fn test_day_jobs_unfinished_sees_only_the_day_past_its_00_job() {
+        infra_file_list::create_table().await.unwrap();
+        // the sqlite file outlives a test run
+        let stream = &format!("day_jobs_{}", config::ider::generate());
+        let day = day_micros(20_000);
+        let unfinished = || day_jobs_unfinished("o", StreamType::Logs, stream, day);
+
+        infra_file_list::add_job("o", StreamType::Logs, stream, day)
+            .await
+            .unwrap();
+        infra_file_list::add_job("o", StreamType::Logs, stream, day + day_micros(1))
+            .await
+            .unwrap();
+        assert!(!unfinished().await);
+
+        let hour23 = infra_file_list::add_job("o", StreamType::Logs, stream, day + hour_micros(23))
+            .await
+            .unwrap();
+        assert!(unfinished().await);
+        infra_file_list::set_job_done(&[hour23]).await.unwrap();
+        assert!(!unfinished().await);
+    }
+}

src/compaction/src/merge/dynamic.rs

@@ -15,7 +15,10 @@
 
 //! Dynamic merge: what the 00:00 job of a closed day does with one partition of that day.
 
-use config::meta::stream::{FileKey, MergeStrategy};
+use config::{
+    meta::stream::{FileKey, MergeStrategy},
+    utils::time::{day_micros, hour_micros},
+};
 use search::datafusion::merge::MergeMode;
 
 use super::{
@@ -30,14 +33,6 @@ const DAY_MERGE_SLACK: usize = 4;
 /// Position of the hour directory in `files/{org}/{type}/{stream}/YYYY/MM/DD/HH/...`.
 const HOUR_SEGMENT: usize = 7;
 
-/// What the dynamic merge job does with one partition of its day.
-pub(super) enum DayPlan {
-    /// The day holds many more files than its size needs: these batches go to `DD/00`.
-    Merge(Vec<PlannedBatch>),
-    /// The day stays; its hour-00 files still get their hourly merge.
-    Hour00(Vec<FileKey>),
-}
-
 /// `prefix` with its hour directory replaced by `00`, where a merged day is written.
 pub(super) fn day_prefix(prefix: &str) -> String {
     prefix
@@ -48,9 +43,26 @@ pub(super) fn day_prefix(prefix: &str) -> String {
         .join("/")
 }
 
-pub(super) fn plan_day(files: Vec<FileKey>, mode: &MergeMode, limits: &BatchLimits<'_>) -> DayPlan {
-    if !should_merge_day(&files, limits.max_file_size) {
-        return DayPlan::Hour00(hour00_files(files, mode, limits));
+/// Whether the day was merged by day: only that writes a file of `DD/00` reaching past the hour.
+pub(super) fn merged_by_day(files: &[FileKey]) -> bool {
+    files.iter().any(day_output)
+}
+
+/// Batches of one partition of the day, all written to `DD/00`; none when the day stays as it is.
+pub(super) fn plan_day(
+    files: Vec<FileKey>,
+    mode: &MergeMode,
+    limits: &BatchLimits<'_>,
+) -> Vec<PlannedBatch> {
+    let outputs = files.iter().filter(|f| day_output(f)).count();
+    let act = if outputs > 0 {
+        // a merged day stays in one place: what arrived since joins it, whatever its count
+        outputs < files.len()
+    } else {
+        should_merge_day(&files, limits.max_file_size)
+    };
+    if !act {
+        return Vec::new();
     }
     let (mode_files, legacy_files) = split_legacy_metrics(files, mode);
     let mut batches: Vec<PlannedBatch> = time_ordered_groups(legacy_files, limits)
@@ -69,31 +81,19 @@ pub(super) fn plan_day(files: Vec<FileKey>, mode: &MergeMode, limits: &BatchLimi
                 .map(|group| (group, mode.clone())),
         );
     }
-    DayPlan::Merge(batches)
+    batches
+}
+
+/// A file of `DD/00` reaching past its hour; no file reaches past its day.
+fn day_output(file: &FileKey) -> bool {
+    file.key.split('/').nth(HOUR_SEGMENT) == Some("00")
+        && file.meta.max_ts % day_micros(1) >= hour_micros(1)
 }
 
 fn should_merge_day(files: &[FileKey], max_file_size: usize) -> bool {
     files.len() > DAY_MERGE_FACTOR * ideal_file_count(files, max_file_size) + DAY_MERGE_SLACK
 }
 
-/// The files an hourly merge of hour 00 would have queried.
-fn hour00_files(
-    mut files: Vec<FileKey>,
-    mode: &MergeMode,
-    limits: &BatchLimits<'_>,
-) -> Vec<FileKey> {
-    let max_original_size = if mode.merges_whole_batch() {
-        i64::MAX
-    } else {
-        limits.merge_max_original_size
-    };
-    files.retain(|f| {
-        f.key.split('/').nth(HOUR_SEGMENT) == Some("00")
-            && f.meta.original_size <= max_original_size
-    });
-    files
-}
-
 /// Size-bounded groups in time order, so each output covers a contiguous range of the day.
 fn time_ordered_groups(mut files: Vec<FileKey>, limits: &BatchLimits<'_>) -> Vec<Vec<FileKey>> {
     files.retain(|f| f.meta.original_size <= limits.merge_max_original_size);
@@ -108,7 +108,7 @@ fn time_ordered_groups(mut files: Vec<FileKey>, limits: &BatchLimits<'_>) -> Vec
 
 #[cfg(test)]
 mod tests {
-    use config::{meta::stream::FileMeta, utils::time::hour_micros};
+    use config::meta::stream::FileMeta;
 
     use super::*;
 
@@ -146,18 +146,19 @@ mod tests {
             .collect()
     }
 
-    fn merged(plan: DayPlan) -> Vec<PlannedBatch> {
-        match plan {
-            DayPlan::Merge(batches) => batches,
-            DayPlan::Hour00(_) => panic!("the day was not merged"),
-        }
-    }
-
-    fn hour00(plan: DayPlan) -> Vec<FileKey> {
-        match plan {
-            DayPlan::Hour00(files) => files,
-            DayPlan::Merge(_) => panic!("the day was merged"),
-        }
+    /// What a merged day looks like: each batch became one file in `DD/00` spanning its hours.
+    fn outputs(batches: &[PlannedBatch]) -> Vec<FileKey> {
+        batches
+            .iter()
+            .enumerate()
+            .map(|(i, (files, _))| {
+                let mut out = file("logs", 0, &format!("merged{i}"), 0);
+                out.meta.min_ts = files.iter().map(|f| f.meta.min_ts).min().unwrap();
+                out.meta.max_ts = files.iter().map(|f| f.meta.max_ts).max().unwrap();
+                out.meta.original_size = files.iter().map(|f| f.meta.original_size).sum();
+                out
+            })
+            .collect()
     }
 
     #[test]
@@ -172,6 +173,18 @@ mod tests {
         );
     }
 
+    /// Only a file of hour 00 reaching past the hour marks a merged day.
+    #[test]
+    fn test_merged_by_day_needs_a_file_of_hour_00_past_the_hour() {
+        assert!(!merged_by_day(&day("logs", "f", MB)));
+        let mut spanning = file("logs", 5, "late", MB);
+        spanning.meta.max_ts = hour_micros(23);
+        assert!(!merged_by_day(&[spanning]));
+        let mut merged = file("logs", 0, "merged", MB);
+        merged.meta.max_ts = hour_micros(23);
+        assert!(merged_by_day(&[merged]));
+    }
+
     /// The design's table: one file per hour against the default 2 GB target.
     #[test]
     fn test_should_merge_day_follows_the_data_volume() {
@@ -205,7 +218,7 @@ mod tests {
         let mut files = day("logs", "f", 300 * MB);
         files.reverse();
         files.swap(3, 17);
-        let batches = merged(plan_day(files, &MergeMode::Classic, &limits));
+        let batches = plan_day(files, &MergeMode::Classic, &limits);
         assert_eq!(batches.len(), 4);
         let ranges: Vec<(i64, i64)> = batches
             .iter()
@@ -220,47 +233,62 @@ mod tests {
         assert!(batches.iter().all(|(files, _)| files.len() == 6));
     }
 
+    /// A day with the files its size needs stays as it is, late data or not.
+    #[test]
+    fn test_plan_day_leaves_a_large_day_alone() {
+        let strategy = MergeStrategy::FileTime;
+        let limits = limits(&strategy);
+        let mut files = day("logs", "f", GB);
+        files.push(file("logs", 0, "late", MB));
+        files.push(file("logs", 5, "late", MB));
+        assert!(plan_day(files, &MergeMode::Classic, &limits).is_empty());
+    }
+
     /// Running the job again on the merged day does nothing: the policy is idempotent.
     #[test]
     fn test_plan_day_second_run_leaves_a_merged_day_alone() {
         let strategy = MergeStrategy::FileTime;
         let limits = limits(&strategy);
-        let batches = merged(plan_day(
+        let batches = plan_day(day("logs", "f", 300 * MB), &MergeMode::Classic, &limits);
+        let merged = outputs(&batches);
+        assert_eq!(merged.len(), 4);
+        assert!(plan_day(merged, &MergeMode::Classic, &limits).is_empty());
+    }
+
+    /// A merged day takes its late files without the threshold; they join an unfilled output.
+    #[test]
+    fn test_plan_day_merged_day_takes_late_files_into_hour_00() {
+        let strategy = MergeStrategy::FileTime;
+        let limits = limits(&strategy);
+        let mut merged = outputs(&plan_day(
             day("logs", "f", 300 * MB),
             &MergeMode::Classic,
             &limits,
         ));
-        // each batch becomes one file in DD/00 that spans the batch's hours
-        let outputs: Vec<FileKey> = batches
-            .iter()
-            .enumerate()
-            .map(|(i, (files, _))| {
-                let mut out = file("logs", 0, &format!("merged{i}"), 0);
-                out.meta.min_ts = files.iter().map(|f| f.meta.min_ts).min().unwrap();
-                out.meta.max_ts = files.iter().map(|f| f.meta.max_ts).max().unwrap();
-                out.meta.original_size = files.iter().map(|f| f.meta.original_size).sum();
-                out
-            })
-            .collect();
-        let left = hour00(plan_day(outputs, &MergeMode::Classic, &limits));
-        assert_eq!(left.len(), 4);
-    }
-
-    #[test]
-    fn test_plan_day_skip_keeps_only_the_hour_00_merge_candidates() {
-        let strategy = MergeStrategy::FileTime;
-        let limits = limits(&strategy);
-        // a large stream, plus late data and a full-size file in hour 00
-        let mut files = day("logs", "f", GB);
-        files.push(file("logs", 0, "late", MB));
-        files.push(file("logs", 0, "full", 2 * GB));
-        files.push(file("logs", 5, "late", MB));
-        let files = hour00(plan_day(files, &MergeMode::Classic, &limits));
-        let names: Vec<&str> = files
+        // the first three outputs are full, the last has room
+        for out in merged.iter_mut().take(3) {
+            out.meta.original_size = 2 * GB;
+        }
+        let mut files = merged.clone();
+        files.push(file("logs", 5, "late5", MB));
+        files.push(file("logs", 9, "late9", MB));
+        let batches = plan_day(files, &MergeMode::Classic, &limits);
+        assert_eq!(batches.len(), 1);
+        let names: Vec<&str> = batches[0]
+            .0
             .iter()
             .map(|f| f.key.rsplit('/').next().unwrap())
             .collect();
-        assert_eq!(names, vec!["f0.parquet", "late.parquet"]);
+        assert_eq!(
+            names,
+            vec!["late5.parquet", "late9.parquet", "merged3.parquet"]
+        );
+        // a single late file with no output to join stays where it is
+        for out in merged.iter_mut() {
+            out.meta.original_size = 2 * GB;
+        }
+        merged.push(file("logs", 5, "late5", MB));
+        assert!(plan_day(merged, &MergeMode::Classic, &limits).is_empty());
     }
 
     /// Indexed metrics keep every series in one batch; legacy files never join it.
@@ -271,7 +299,7 @@ mod tests {
         let mut files = day("metrics", "indexed-v1-", 100 * MB);
         files.push(file("metrics", 3, "legacy-a", MB));
         files.push(file("metrics", 9, "legacy-b", MB));
-        let batches = merged(plan_day(files, &MergeMode::MetricsIndexed, &limits));
+        let batches = plan_day(files, &MergeMode::MetricsIndexed, &limits);
         assert_eq!(batches.len(), 2);
         let (legacy, indexed): (Vec<_>, Vec<_>) = batches
             .iter()

src/compaction/src/merge/job.rs

@@ -13,19 +13,26 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::collections::{BTreeSet, HashMap};
+
 use chrono::{DateTime, Datelike, Duration, TimeZone, Utc};
 use config::{
     cluster::LOCAL_NODE,
     get_config,
     meta::stream::{PartitionTimeLevel, StreamType},
-    utils::time::{day_micros, hour_micros},
+    utils::time::{HourFormat, day_micros, get_ymdh_from_micros, hour_micros},
 };
 use infra::{
     cluster::get_node_by_uuid,
     dist_lock, file_list as infra_file_list,
-    schema::{get_dynamic_merge, get_partition_time_level, unwrap_stream_created_at},
+    schema::{
+        get_dynamic_merge, get_partition_time_level, get_stream_setting_dynamic_merge,
+        unwrap_stream_created_at,
+    },
 };
 
+use super::dynamic::merged_by_day;
+
 /// What the merge job at `offset` covers.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(super) enum JobScope {
@@ -245,6 +252,10 @@ pub async fn generate_old_data_job_by_stream(
     .await?;
 
     // generate merging job
+    let dynamic_merge =
+        get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.as_ref()));
+    let mut merged_days = HashMap::new();
+    let mut offsets = BTreeSet::new();
     for hour in hours {
         let column = hour.split('/').collect::<Vec<_>>();
         if column.len() != 4 {
@@ -257,10 +268,18 @@ pub async fn generate_old_data_job_by_stream(
             column[0], column[1], column[2], column[3]
         ))?
         .with_timezone(&Utc);
-        let offset = offset.timestamp_micros();
+        let mut offset = offset.timestamp_micros();
+        if dynamic_merge {
+            offset =
+                merged_day_job_offset(org_id, stream_type, stream_name, offset, &mut merged_days)
+                    .await?;
+        }
         log::debug!(
             "[COMPACTOR] generate_old_data_job_by_stream [{org_id}/{stream_type}/{stream_name}] hours: {hour}, offset: {offset}"
         );
+        offsets.insert(offset);
+    }
+    for offset in offsets {
         if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await {
             return Err(anyhow::anyhow!(
                 "[COMPACTOR] add file_list_jobs for old data failed: {e}"
@@ -391,8 +410,40 @@ fn dynamic_merge_offset(offset: i64) -> Option<i64> {
     (offset % day_micros(1) == 0).then(|| offset - day_micros(1))
 }
 
+/// An old hour of a closed day merged by day re-arms the day's 00:00 job: its files go to `DD/00`.
+async fn merged_day_job_offset(
+    org_id: &str,
+    stream_type: StreamType,
+    stream_name: &str,
+    offset: i64,
+    merged_days: &mut HashMap<i64, bool>,
+) -> Result<i64, anyhow::Error> {
+    let day = offset - offset % day_micros(1);
+    if offset == day || !crate::is_past_day(offset) {
+        return Ok(offset);
+    }
+    let merged = match merged_days.get(&day) {
+        Some(merged) => *merged,
+        None => {
+            let hour00 = get_ymdh_from_micros(day, HourFormat::Zero);
+            let files = infra_file_list::query_for_merge(
+                org_id,
+                stream_type,
+                stream_name,
+                (hour00.clone(), hour00),
+                i64::MAX,
+            )
+            .await?;
+            *merged_days.entry(day).or_insert(merged_by_day(&files))
+        }
+    };
+    Ok(if merged { day } else { offset })
+}
+
 #[cfg(test)]
 mod tests {
+    use config::meta::stream::{FileKey, FileMeta};
+
     use super::*;
 
     #[test]
@@ -471,4 +522,45 @@ mod tests {
         assert_eq!(dynamic_merge_offset(day), Some(day - day_micros(1)));
         assert_eq!(dynamic_merge_offset(day + hour_micros(1)), None);
     }
+
+    /// An old hour of a day merged by day re-arms the day's 00:00 job instead of its own.
+    #[tokio::test]
+    async fn test_merged_day_job_offset_follows_the_files_of_hour_00() {
+        infra_file_list::create_table().await.unwrap();
+        // the sqlite file outlives a test run
+        let stream = &format!("merged_day_{}", config::ider::generate());
+        let day = day_start();
+        let hour15 = day + hour_micros(15);
+        let hour00_file = |name: &str, max_ts: i64| {
+            FileKey::new(
+                0,
+                "default".to_string(),
+                format!("files/old/logs/{stream}/2026/08/18/00/{name}.parquet"),
+                FileMeta {
+                    min_ts: day + 1,
+                    max_ts,
+                    records: 1,
+                    original_size: 1,
+                    compressed_size: 1,
+                    ..Default::default()
+                },
+                false,
+            )
+        };
+        let offset = |offset: i64| async move {
+            merged_day_job_offset("old", StreamType::Logs, stream, offset, &mut HashMap::new())
+                .await
+                .unwrap()
+        };
+
+        // an hourly file of hour 00 does not make a merged day
+        let hourly = hour00_file("hourly", day + hour_micros(1) - 2);
+        infra_file_list::batch_add(&[hourly]).await.unwrap();
+        assert_eq!(offset(hour15).await, hour15);
+
+        let merged = hour00_file("merged", day + hour_micros(24) - 2);
+        infra_file_list::batch_add(&[merged]).await.unwrap();
+        assert_eq!(offset(hour15).await, day);
+        assert_eq!(offset(day).await, day);
+    }
 }

src/compaction/src/merge/stream.rs

@@ -19,10 +19,7 @@ use config::{
     cluster, get_config,
     meta::stream::{FileKey, FileListDeleted, MergeStrategy, PartitionTimeLevel, StreamType},
     metrics,
-    utils::{
-        parquet::parse_file_key_columns,
-        time::{day_micros, hour_micros},
-    },
+    utils::{parquet::parse_file_key_columns, time::hour_micros},
 };
 use hashbrown::{HashMap, HashSet};
 use infra::{cache::file_data, file_list as infra_file_list};
@@ -36,7 +33,7 @@ use tokio::{
 };
 
 use super::{
-    dynamic::{DayPlan, day_prefix, plan_day},
+    dynamic::{day_prefix, plan_day},
     job::{JobScope, job_range_end, job_scope},
     plan::{BatchLimits, PlannedBatch, plan_batches},
 };
@@ -89,9 +86,10 @@ pub async fn merge_by_stream(
     let is_incremental = !crate::is_past_hour(offset);
 
     // check offset
-    let dynamic_day = offset % day_micros(1) == 0
-        && crate::is_past_day(offset)
-        && infra::schema::get_dynamic_merge(org_id, stream_name, stream_type).await;
+    let dynamic_day = crate::is_dynamic_day_job(
+        offset,
+        infra::schema::get_dynamic_merge(org_id, stream_name, stream_type).await,
+    );
     let scope = job_scope(
         stream_type,
         offset,
@@ -443,27 +441,22 @@ fn plan_partition(
     stream: &str,
     prefix: &str,
 ) -> Vec<PlannedBatch> {
-    let hourly = |files| {
-        plan_batches(
+    if scope != JobScope::DynamicDay {
+        return plan_batches(
             sort_by_strategy(files, limits.strategy),
             mode,
             limits,
             stream,
-        )
-    };
-    if scope != JobScope::DynamicDay {
-        return hourly(files);
+        );
     }
     let day_files = files.len();
-    match plan_day(files, mode, limits) {
-        DayPlan::Merge(batches) => {
-            log::info!(
-                "[COMPACTOR] merge_by_stream [{stream}] dynamic merge of {day_files} files into {prefix}"
-            );
-            batches
-        }
-        DayPlan::Hour00(files) => hourly(files),
+    let batches = plan_day(files, mode, limits);
+    if !batches.is_empty() {
+        log::info!(
+            "[COMPACTOR] merge_by_stream [{stream}] dynamic merge of {day_files} files into {prefix}"
+        );
     }
+    batches
 }
 
 /// The order the hourly planner groups files in.

src/infra/src/file_list/mod.rs

@@ -277,6 +277,13 @@ pub trait FileList: Sync + Send + 'static {
         fast_mode: bool,
     ) -> Result<Vec<MergeJobRecord>>;
     async fn get_pending_jobs_count(&self) -> Result<stdHashMap<String, stdHashMap<String, i64>>>;
+    async fn has_unfinished_jobs(
+        &self,
+        org_id: &str,
+        stream_type: StreamType,
+        stream: &str,
+        offset_range: (i64, i64),
+    ) -> Result<bool>;
     async fn set_job_pending(&self, ids: &[i64], offsets: i64, stream: Option<&str>)
     -> Result<u64>;
     async fn set_job_done(&self, ids: &[i64]) -> Result<()>;
@@ -777,6 +784,19 @@ pub async fn get_pending_jobs_count() -> Result<stdHashMap<String, stdHashMap<St
     CLIENT.get_pending_jobs_count().await
 }
 
+/// Whether a job of the stream with `offsets` in the inclusive range is pending or running.
+#[inline]
+pub async fn has_unfinished_jobs(
+    org_id: &str,
+    stream_type: StreamType,
+    stream: &str,
+    offset_range: (i64, i64),
+) -> Result<bool> {
+    CLIENT
+        .has_unfinished_jobs(org_id, stream_type, stream, offset_range)
+        .await
+}
+
 #[inline]
 pub async fn set_job_pending(ids: &[i64], offsets: i64, stream: Option<&str>) -> Result<u64> {
     CLIENT.set_job_pending(ids, offsets, stream).await

src/infra/src/file_list/postgres.rs

@@ -1564,6 +1564,30 @@ DO UPDATE SET
         Ok(ret)
     }
 
+    async fn has_unfinished_jobs(
+        &self,
+        org_id: &str,
+        stream_type: StreamType,
+        stream: &str,
+        offset_range: (i64, i64),
+    ) -> Result<bool> {
+        let stream_key = format!("{org_id}/{stream_type}/{stream}");
+        let pool = CLIENT_RO.clone();
+        DB_QUERY_NUMS
+            .with_label_values(&["query", "file_list_jobs"])
+            .inc();
+        let ret = sqlx::query(
+            "SELECT id FROM file_list_jobs WHERE stream = $1 AND offsets >= $2 AND offsets <= $3 AND status <> $4 LIMIT 1;",
+        )
+        .bind(stream_key)
+        .bind(offset_range.0)
+        .bind(offset_range.1)
+        .bind(super::FileListJobStatus::Done)
+        .fetch_optional(&pool)
+        .await?;
+        Ok(ret.is_some())
+    }
+
     async fn set_job_pending(
         &self,
         ids: &[i64],

src/infra/src/file_list/sqlite.rs

@@ -1230,6 +1230,27 @@ SELECT stream, max(id) as id, COUNT(*) AS num
         Ok(ret)
     }
 
+    async fn has_unfinished_jobs(
+        &self,
+        org_id: &str,
+        stream_type: StreamType,
+        stream: &str,
+        offset_range: (i64, i64),
+    ) -> Result<bool> {
+        let stream_key = format!("{org_id}/{stream_type}/{stream}");
+        let client = CLIENT_RO.clone();
+        let ret = sqlx::query(
+            "SELECT id FROM file_list_jobs WHERE stream = $1 AND offsets >= $2 AND offsets <= $3 AND status <> $4 LIMIT 1;",
+        )
+        .bind(stream_key)
+        .bind(offset_range.0)
+        .bind(offset_range.1)
+        .bind(super::FileListJobStatus::Done)
+        .fetch_optional(&client)
+        .await?;
+        Ok(ret.is_some())
+    }
+
     async fn set_job_pending(
         &self,
         ids: &[i64],