apache · jlfsdtc · Jun 11, 2026 · Jun 8, 2026
diff --git a/src/core-job/src/main/java/org/apache/kylin/job/common/SegmentUtil.java b/src/core-job/src/main/java/org/apache/kylin/job/common/SegmentUtil.java
@@ -180,10 +180,8 @@ private static <T extends ISegment> boolean isAnyPartitionRefreshing(T segment)
     public static Segments<NDataSegment> getValidSegments(String modelId, String project) {
         val df = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project).getDataflow(modelId);
 
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
-        jobMapperFilter.setModelIds(Lists.newArrayList(modelId));
-        jobMapperFilter.setStatuses(ExecutableState.getNotFinalStates());
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project)
+                .modelIds(Lists.newArrayList(modelId)).statuses(ExecutableState.getNotFinalStates()).build();
         List<JobInfo> runningJobInfoList = JobContextUtil.getJobInfoDao(KylinConfig.getInstanceFromEnv())
                 .getJobInfoListByFilter(jobMapperFilter);
         ExecutableManager executableManager = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), project);

diff --git a/src/core-job/src/main/java/org/apache/kylin/job/dao/JobInfoDao.java b/src/core-job/src/main/java/org/apache/kylin/job/dao/JobInfoDao.java
@@ -25,17 +25,20 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil;
 import org.apache.kylin.common.util.CompressionUtils;
 import org.apache.kylin.common.util.JsonUtil;
 import org.apache.kylin.guava30.shaded.common.base.Preconditions;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
 import org.apache.kylin.job.domain.JobInfo;
 import org.apache.kylin.job.domain.JobLock;
 import org.apache.kylin.job.exception.ExecuteRuntimeException;
@@ -73,18 +76,24 @@ public class JobInfoDao {
     @Setter
     private JobLockMapper jobLockMapper;
 
+    public List<JobInfo> getJobInfoListByProjectFilter(final JobMapperFilter jobMapperFilter) {
+        if (StringUtils.isBlank(jobMapperFilter.getProject())
+                && CollectionUtils.isEmpty(jobMapperFilter.getProjects())) {
+            return Collections.emptyList();
+        }
+        return getJobInfoListByFilter(jobMapperFilter);
+    }
+
     public List<JobInfo> getJobInfoListByFilter(final JobMapperFilter jobMapperFilter) {
-        List<JobInfo> jobInfoList = jobInfoMapper.selectByJobFilter(jobMapperFilter);
-        return jobInfoList;
+        return jobInfoMapper.selectByJobFilter(jobMapperFilter);
     }
 
     public long countByFilter(JobMapperFilter jobMapperFilter) {
         return jobInfoMapper.countByJobFilter(jobMapperFilter);
     }
 
     public List<ExecutablePO> getJobs(String project) {
-        JobMapperFilter filter = new JobMapperFilter();
-        filter.setProject(project);
+        JobMapperFilter filter = JobMapperFilter.builder().project(project).build();
         return jobInfoMapper.selectByJobFilter(filter).stream().map(JobInfoUtil::deserializeExecutablePO)
                 .collect(Collectors.toList());
     }
@@ -136,12 +145,20 @@ public ExecutablePO getExecutablePOByUuid(String uuid) {
         return null;
     }
 
+    public ExecutablePO getExecutablePoByUuidWithProject(String project, String jobId) {
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project).jobIds(Lists.newArrayList(jobId))
+                .build();
+        List<JobInfo> jobInfoList = jobInfoMapper.selectByJobFilter(jobMapperFilter);
+        if (CollectionUtils.isEmpty(jobInfoList) || jobInfoList.size() != 1) {
+            return null;
+        }
+        return JobInfoUtil.deserializeExecutablePO(jobInfoList.get(0));
+    }
+
     public List<ExecutablePO> getExecutablePoByStatus(String project, List<String> jobIds,
             List<ExecutableState> filterStatuses) {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
-        jobMapperFilter.setStatuses(filterStatuses);
-        jobMapperFilter.setJobIds(jobIds);
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project).statuses(filterStatuses)
+                .jobIds(jobIds).build();
         List<JobInfo> jobInfoList = jobInfoMapper.selectByJobFilter(jobMapperFilter);
         if (CollectionUtils.isEmpty(jobInfoList)) {
             return new ArrayList<>();

diff --git a/src/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java b/src/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableManager.java
@@ -1103,8 +1103,7 @@ public Map<String, String> getWaiteTime(ExecutablePO executablePO, AbstractExecu
                     if (task instanceof ChainedStageExecutable) {
                         final ChainedStageExecutable stageExecutable = (ChainedStageExecutable) task;
                         Map<String, List<StageExecutable>> stageMap = Optional
-                                .ofNullable(stageExecutable.getStagesMap())
-                                .orElse(Maps.newHashMap());
+                                .ofNullable(stageExecutable.getStagesMap()).orElse(Maps.newHashMap());
                         val taskStartTime = task.getStartTime();
                         for (Map.Entry<String, List<StageExecutable>> entry : stageMap.entrySet()) {
                             final String segmentId = entry.getKey();
@@ -1287,8 +1286,7 @@ public List<AbstractExecutable> getPartialExecutables(String modelId) {
     }
 
     private List<ExecutablePO> getExecutablePOByModelId(String project, String modelId) {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project).build();
         if (null != modelId) {
             jobMapperFilter.setModelIds(Lists.newArrayList(modelId));
         }
@@ -1385,8 +1383,7 @@ public List<AbstractExecutable> listMultiPartitionModelExec(String model, Predic
     }
 
     public List<ExecutablePO> getExecutablePOsByStatus(List<String> jobIds, List<ExecutableState> executableStates) {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project).build();
         if (CollectionUtils.isNotEmpty(jobIds)) {
             jobMapperFilter.setJobIds(jobIds);
         }
@@ -1411,16 +1408,6 @@ public List<AbstractExecutable> getExecutablesByStatus(ExecutableState status) {
         return getExecutablesByStatus(null, Lists.newArrayList(status));
     }
 
-    public List<AbstractExecutable> getExecutablesByJobType(Set<JobTypeEnum> RELATED_JOBS) {
-        List<String> jobTypeNames = RELATED_JOBS.stream().map(jobTypeEnum -> jobTypeEnum.name())
-                .collect(Collectors.toList());
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setJobNames(jobTypeNames);
-        List<JobInfo> jobInfoList = jobInfoDao.getJobInfoListByFilter(jobMapperFilter);
-        return jobInfoList.stream().map(jobInfo -> JobInfoUtil.deserializeExecutablePO(jobInfo)).map(this::fromPO)
-                .collect(Collectors.toList());
-    }
-
     public ExecutablePO getExecutablePO(String jobId) {
         return jobInfoDao.getExecutablePOByUuid(jobId);
     }
@@ -1672,15 +1659,14 @@ public List<AbstractExecutable> getRunningExecutables(String project, String mod
         if (StringUtils.isNotBlank(model)) {
             return listExecByModelAndStatus(model, ExecutableState::isRunning, null);
         } else {
-            JobMapperFilter jobMapperFilter = new JobMapperFilter();
             List<ExecutableState> runningStates = Lists.newArrayList();
             for (ExecutableState executableState : ExecutableState.values()) {
                 if (executableState.isRunning()) {
                     runningStates.add(executableState);
                 }
             }
-            jobMapperFilter.setStatuses(runningStates);
-            jobMapperFilter.setProject(project);
+            JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project).statuses(runningStates)
+                    .build();
             return jobInfoDao.getJobInfoListByFilter(jobMapperFilter).stream()
                     .map(jobInfo -> JobInfoUtil.deserializeExecutablePO(jobInfo)).map(this::fromPO)
                     .collect(Collectors.toList());
@@ -1752,9 +1738,8 @@ public void suicideRunningJobByJobType(String project, String targetModelId, Lis
     }
 
     public void checkSuicideJobOfModel(String project, String modelId) {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
-        jobMapperFilter.setModelIds(Lists.newArrayList(modelId));
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project)
+                .modelIds(Lists.newArrayList(modelId)).build();
         jobMapperFilter.setStatuses(ExecutableState.ERROR, ExecutableState.PAUSED);
         List<JobInfo> errorJobInfoList = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), project)
                 .fetchJobsByFilter(jobMapperFilter);
@@ -2013,12 +1998,10 @@ public List<JobInfo> fetchJobsByTypesAndStates(String project, List<String> jobT
     }
 
     public List<AbstractExecutable> getNotFinalExecutablesByType(List<JobTypeEnum> jobTypeEnums) {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setProject(project);
-        jobMapperFilter.setStatuses(ExecutableState.getNotFinalStates());
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().project(project)
+                .statuses(ExecutableState.getNotFinalStates()).build();
         if (CollectionUtils.isNotEmpty(jobTypeEnums)) {
-            jobMapperFilter.setJobNames(
-                    jobTypeEnums.stream().map(jobTypeEnum -> jobTypeEnum.name()).collect(Collectors.toList()));
+            jobMapperFilter.setJobNames(jobTypeEnums.stream().map(Enum::name).collect(Collectors.toList()));
         }
         List<JobInfo> jobInfoList = jobInfoDao.getJobInfoListByFilter(jobMapperFilter);
         return jobInfoList.stream().map(jobInfo -> fromPO(JobInfoUtil.deserializeExecutablePO(jobInfo)))

diff --git a/src/core-job/src/main/java/org/apache/kylin/job/rest/JobMapperFilter.java b/src/core-job/src/main/java/org/apache/kylin/job/rest/JobMapperFilter.java
@@ -58,6 +58,8 @@ public class JobMapperFilter {
 
     private String project;
 
+    private List<String> projects;
+
     private String orderByFiled;
 
     private String orderType;
@@ -76,8 +78,32 @@ public class JobMapperFilter {
     public void setStatuses(List<ExecutableState> stateList) {
         statuses = stateList;
     }
-    
+
     public void setStatuses(ExecutableState... states) {
         statuses = Lists.newArrayList(states);
     }
+
+    public void setProjects(List<String> projects) {
+        this.project = null;
+        this.projects = projects;
+    }
+
+    public void setProject(String project) {
+        this.project = project;
+        this.projects = null;
+    }
+
+    public static class JobMapperFilterBuilder {
+        public JobMapperFilterBuilder projects(List<String> projects) {
+            this.project = null;
+            this.projects = projects;
+            return this;
+        }
+
+        public JobMapperFilterBuilder project(String project) {
+            this.project = project;
+            this.projects = null;
+            return this;
+        }
+    }
 }
diff --git a/src/core-job/src/main/java/org/apache/kylin/job/runners/JobCheckRunner.java b/src/core-job/src/main/java/org/apache/kylin/job/runners/JobCheckRunner.java
@@ -124,8 +124,8 @@ private static boolean markSuicideJobWithTransaction(AbstractExecutable jobExecu
     }
 
     private void markSuicideForErrorOrPausedJobs() {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
-        jobMapperFilter.setStatuses(Lists.newArrayList(ExecutableState.ERROR, ExecutableState.PAUSED));
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder()
+                .statuses(Lists.newArrayList(ExecutableState.ERROR, ExecutableState.PAUSED)).build();
         jobMapperFilter.setLimit(10);
         jobMapperFilter.setOffset(0);
         List<JobInfo> jobInfoList = jobContext.getJobInfoMapper().selectByJobFilter(jobMapperFilter);

diff --git a/src/core-job/src/main/java/org/apache/kylin/job/scheduler/JdbcJobScheduler.java b/src/core-job/src/main/java/org/apache/kylin/job/scheduler/JdbcJobScheduler.java
@@ -360,10 +360,9 @@ private static boolean markSuicideJobWithTransaction(JobInfo jobInfo) {
     }
 
     private List<JobInfo> getProcessingJobInfoWithOrder() {
-        JobMapperFilter jobMapperFilter = new JobMapperFilter();
+        JobMapperFilter jobMapperFilter = JobMapperFilter.builder().orderByFiled("priority,create_time")
+                .orderType("ASC").build();
         jobMapperFilter.setStatuses(ExecutableState.READY, ExecutableState.PENDING, ExecutableState.RUNNING);
-        jobMapperFilter.setOrderByFiled("priority,create_time");
-        jobMapperFilter.setOrderType("ASC");
         return jobContext.getJobInfoMapper().selectByJobFilter(jobMapperFilter);
     }
 
@@ -407,12 +406,11 @@ private Map<String, Integer> getProjectProduceCount(Map<String, Integer> project
 
     private void releaseExpiredLock() {
         int batchSize = jobContext.getKylinConfig().getJobSchedulerMasterPollBatchSize();
-        JobMapperFilter filter = new JobMapperFilter();
         List<String> jobIds = jobContext.getJobLockMapper().findExpiredORNonLockIdList(batchSize);
         if (jobIds.isEmpty()) {
             return;
         }
-        filter.setJobIds(jobIds);
+        JobMapperFilter filter = JobMapperFilter.builder().jobIds(jobIds).build();
         List<JobInfo> jobs = jobContext.getJobInfoMapper().selectByJobFilter(filter);
         List<String> jobInfoIds = jobs.stream().map(JobInfo::getJobId).collect(Collectors.toList());
         List<String> toRemoveLocks = Lists.newArrayList(jobIds).stream().filter(jobId -> !jobInfoIds.contains(jobId))

diff --git a/src/core-job/src/main/resources/mybatis-mapper/JobInfoMapper.xml b/src/core-job/src/main/resources/mybatis-mapper/JobInfoMapper.xml
@@ -214,6 +214,11 @@
       <if test="project != null">
         project = #{project}
       </if>
+      <if test="projects != null">
+        <foreach close=")" collection="projects" index="index" item="item" open="project in (" separator=",">
+          #{item}
+        </foreach>
+      </if>
       <if test="queryStartTime != null">
         AND update_time &gt;= #{queryStartTime}
       </if>
@@ -275,6 +280,11 @@
       <if test="project != null">
         project = #{project}
       </if>
+      <if test="projects != null">
+        <foreach close=")" collection="projects" index="index" item="item" open="project in (" separator=",">
+            #{item}
+        </foreach>
+      </if>
       <if test="queryStartTime != null">
         AND update_time &gt;= #{queryStartTime}
       </if>

diff --git a/src/core-job/src/test/java/org/apache/kylin/job/dao/JobInfoDaoTest.java b/src/core-job/src/test/java/org/apache/kylin/job/dao/JobInfoDaoTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.job.dao;
+
+import java.util.List;
+
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
+import org.apache.kylin.job.domain.JobInfo;
+import org.apache.kylin.job.mapper.JobInfoMapper;
+import org.apache.kylin.job.rest.JobMapperFilter;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+import org.springframework.test.util.ReflectionTestUtils;
+
+public class JobInfoDaoTest {
+
+    private JobInfoMapper jobInfoMapper;
+    private JobInfoDao jobInfoDao;
+
+    @Before
+    public void setUp() {
+        jobInfoMapper = Mockito.mock(JobInfoMapper.class);
+        jobInfoDao = new JobInfoDao();
+        ReflectionTestUtils.setField(jobInfoDao, "jobInfoMapper", jobInfoMapper);
+    }
+
+    private JobInfo mockJobInfo(String jobId) {
+        JobInfo jobInfo = new JobInfo();
+        jobInfo.setJobId(jobId);
+        return jobInfo;
+    }
+
+    @Test
+    public void testGetJobInfoListByProjectFilterReturnsEmptyWhenNoProjectScope() {
+        JobMapperFilter filter = JobMapperFilter.builder().build();
+
+        List<JobInfo> result = jobInfoDao.getJobInfoListByProjectFilter(filter);
+
+        Assert.assertTrue(result.isEmpty());
+        // must short-circuit without hitting the mapper to avoid a cross-project scan
+        Mockito.verify(jobInfoMapper, Mockito.never()).selectByJobFilter(ArgumentMatchers.any());
+    }
+
+    @Test
+    public void testGetJobInfoListByProjectFilterReturnsEmptyWhenProjectIsBlank() {
+        JobMapperFilter filter = JobMapperFilter.builder().project("   ").build();
+
+        List<JobInfo> result = jobInfoDao.getJobInfoListByProjectFilter(filter);
+
+        Assert.assertTrue(result.isEmpty());
+        Mockito.verify(jobInfoMapper, Mockito.never()).selectByJobFilter(ArgumentMatchers.any());
+    }
+
+    @Test
+    public void testGetJobInfoListByProjectFilterDelegatesWhenSingleProjectPresent() {
+        JobMapperFilter filter = JobMapperFilter.builder().project("default").build();
+        List<JobInfo> expected = Lists.newArrayList(mockJobInfo("job-1"));
+        Mockito.when(jobInfoMapper.selectByJobFilter(filter)).thenReturn(expected);
+
+        List<JobInfo> result = jobInfoDao.getJobInfoListByProjectFilter(filter);
+
+        Assert.assertSame(expected, result);
+        Mockito.verify(jobInfoMapper, Mockito.times(1)).selectByJobFilter(filter);
+    }
+
+    @Test
+    public void testGetJobInfoListByProjectFilterDelegatesWhenProjectsPresent() {
+        JobMapperFilter filter = JobMapperFilter.builder().projects(Lists.newArrayList("p1", "p2")).build();
+        List<JobInfo> expected = Lists.newArrayList(mockJobInfo("job-1"), mockJobInfo("job-2"));
+        Mockito.when(jobInfoMapper.selectByJobFilter(filter)).thenReturn(expected);
+
+        List<JobInfo> result = jobInfoDao.getJobInfoListByProjectFilter(filter);
+
+        Assert.assertEquals(2, result.size());
+        Mockito.verify(jobInfoMapper, Mockito.times(1)).selectByJobFilter(filter);
+    }
+
+    @Test
+    public void testGetJobInfoListByFilterAlwaysDelegates() {
+        JobMapperFilter filter = JobMapperFilter.builder().build();
+        List<JobInfo> expected = Lists.newArrayList(mockJobInfo("job-1"));
+        Mockito.when(jobInfoMapper.selectByJobFilter(filter)).thenReturn(expected);
+
+        List<JobInfo> result = jobInfoDao.getJobInfoListByFilter(filter);
+
+        Assert.assertSame(expected, result);
+        Mockito.verify(jobInfoMapper, Mockito.times(1)).selectByJobFilter(filter);
+    }
+}