[Improve] mybatis-plus query improvements

apache · Jan 16, 2025 · 8a51037 · 8a51037
1 parent c1eee11
commit 8a51037
Show file tree

Hide file tree

Showing 25 changed files with 81 additions and 282 deletions.
diff --git a/...rvice/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java b/...rvice/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java
@@ -37,8 +37,6 @@ public interface FlinkApplicationMapper extends BaseMapper<FlinkApplication> {
 
     List<FlinkApplication> selectAppsByTeamId(@Param("teamId") Long teamId);
 
-    boolean mapping(@Param("app") FlinkApplication appParam);
-
     List<String> selectRecentK8sNamespaces(@Param("limitSize") Integer limit);
 
     List<String> selectRecentK8sClusterIds(
@@ -53,10 +51,6 @@ List<String> selectRecentK8sClusterIds(
 
     List<String> selectRecentK8sTmPodTemplates(@Param("limitSize") Integer limit);
 
-    List<FlinkApplication> selectAppsByProjectId(@Param("projectId") Long id);
-
-    boolean existsRunningJobByClusterId(@Param("clusterId") Long clusterId);
-
     Integer countAffectedByClusterId(
                                      @Param("clusterId") Long clusterId, @Param("dbType") String dbType);
 }
diff --git a/...e-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkCatalogMapper.java b/...e-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkCatalogMapper.java
@@ -19,18 +19,9 @@
 
 import org.apache.streampark.console.core.entity.FlinkCatalog;
 
-import org.apache.ibatis.annotations.Param;
-
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 
 /** catalog mapper */
 public interface FlinkCatalogMapper extends BaseMapper<FlinkCatalog> {
 
-    boolean existsByCatalogName(@Param("catalogName") String catalogName);
-
-    FlinkCatalog selectByCatalogName(@Param("catalogName") String catalogName);
-
-    IPage<FlinkCatalog> selectPage(Page<FlinkCatalog> page, @Param("catalog") FlinkCatalog catalog);
 }
diff --git a/...e-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkClusterMapper.java b/...e-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkClusterMapper.java
@@ -19,14 +19,8 @@
 
 import org.apache.streampark.console.core.entity.FlinkCluster;
 
-import org.apache.ibatis.annotations.Param;
-
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 
 public interface FlinkClusterMapper extends BaseMapper<FlinkCluster> {
 
-    boolean existsByClusterId(@Param("clusterId") String clusterId, @Param("id") Long id);
-
-    boolean existsByClusterName(@Param("clusterName") String clusterName, @Param("id") Long id);
-
 }
diff --git a/...onsole-service/src/main/java/org/apache/streampark/console/core/mapper/ProjectMapper.java b/...onsole-service/src/main/java/org/apache/streampark/console/core/mapper/ProjectMapper.java
@@ -25,15 +25,8 @@
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 
-import java.util.List;
-
 public interface ProjectMapper extends BaseMapper<Project> {
 
     IPage<Project> selectPage(Page<Project> page, @Param("project") Project project);
 
-    boolean existsByTeamId(@Param("teamId") Long teamId);
-
-    List<Project> selectProjectsByTeamId(@Param("teamId") Long teamId);
-
-    Long getBuildingCount();
 }
diff --git a/...nsole-service/src/main/java/org/apache/streampark/console/core/mapper/ResourceMapper.java b/...nsole-service/src/main/java/org/apache/streampark/console/core/mapper/ResourceMapper.java
@@ -28,6 +28,4 @@
 public interface ResourceMapper extends BaseMapper<Resource> {
 
     IPage<Resource> selectPage(Page<Resource> page, @Param("resource") Resource resource);
-
-    boolean existsByUserId(@Param("userId") Long userId);
 }
diff --git a/...rvice/src/main/java/org/apache/streampark/console/core/mapper/SparkApplicationMapper.java b/...rvice/src/main/java/org/apache/streampark/console/core/mapper/SparkApplicationMapper.java
@@ -38,19 +38,8 @@ IPage<SparkApplication> selectPage(
 
     void persistMetrics(@Param("app") SparkApplication application);
 
-    boolean mapping(@Param("app") SparkApplication appParam);
-
     List<String> selectRecentK8sNamespaces(@Param("limitSize") Integer limit);
 
-    List<String> selectRecentK8sClusterIds(
-                                           @Param("deployMode") Integer deployMode,
-                                           @Param("limitSize") Integer limit);
-
     List<String> selectRecentK8sPodTemplates(@Param("limitSize") Integer limit);
 
-    List<String> selectRecentK8sJmPodTemplates(@Param("limitSize") Integer limit);
-
-    List<String> selectRecentK8sTmPodTemplates(@Param("limitSize") Integer limit);
-
-    List<SparkApplication> selectAppsByProjectId(@Param("projectId") Long id);
 }
diff --git a/...nsole-service/src/main/java/org/apache/streampark/console/core/mapper/VariableMapper.java b/...nsole-service/src/main/java/org/apache/streampark/console/core/mapper/VariableMapper.java
@@ -25,13 +25,8 @@
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 
-import java.util.List;
-
 public interface VariableMapper extends BaseMapper<Variable> {
 
     IPage<Variable> selectPage(Page<Variable> page, @Param("variable") Variable variable);
 
-    List<Variable> selectVarsByTeamId(@Param("teamId") Long teamId, @Param("keyword") String keyword);
-
-    boolean existsByTeamId(@Param("teamId") Long teamId);
 }
diff --git a/...sole-service/src/main/java/org/apache/streampark/console/core/mapper/YarnQueueMapper.java b/...sole-service/src/main/java/org/apache/streampark/console/core/mapper/YarnQueueMapper.java
@@ -19,12 +19,8 @@
 
 import org.apache.streampark.console.core.entity.YarnQueue;
 
-import org.apache.ibatis.annotations.Param;
-
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 
 /** Yarn queue mapper definition. */
 public interface YarnQueueMapper extends BaseMapper<YarnQueue> {
-
-    boolean existsByQueueLabel(@Param("yarnQueue") YarnQueue yarnQueue);
 }
diff --git a/...che/streampark/console/core/service/application/impl/FlinkApplicationInfoServiceImpl.java b/...che/streampark/console/core/service/application/impl/FlinkApplicationInfoServiceImpl.java
@@ -260,12 +260,16 @@ public boolean existsByUserId(Long userId) {
 
     @Override
     public boolean existsRunningByClusterId(Long clusterId) {
-        return baseMapper.existsRunningJobByClusterId(clusterId)
-            || FlinkAppHttpWatcher.getWatchingApps().stream()
-                .anyMatch(
-                    application -> clusterId.equals(application.getFlinkClusterId())
-                        && FlinkAppStateEnum.RUNNING == application
-                            .getStateEnum());
+        boolean exists = this.lambdaQuery()
+            .eq(FlinkApplication::getFlinkClusterId, clusterId)
+            .eq(FlinkApplication::getState, FlinkAppStateEnum.RUNNING.getValue())
+            .exists();
+
+        return exists || FlinkAppHttpWatcher.getWatchingApps().stream()
+            .anyMatch(
+                application -> clusterId.equals(application.getFlinkClusterId())
+                    && FlinkAppStateEnum.RUNNING == application
+                        .getStateEnum());
     }
 
     @Override

diff --git a/...e/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java b/...e/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
@@ -179,15 +179,23 @@ public void persistMetrics(FlinkApplication appParam) {
 
     @Override
     public boolean mapping(FlinkApplication appParam) {
-        boolean mapping = this.baseMapper.mapping(appParam);
+        boolean result = this.lambdaUpdate()
+            .eq(FlinkApplication::getId, appParam.getId())
+            .set(appParam.getClusterId() != null, FlinkApplication::getClusterId, appParam.getClusterId())
+            .set(appParam.getJobId() != null, FlinkApplication::getJobId, appParam.getJobId())
+            .set(FlinkApplication::getEndTime, null)
+            .set(FlinkApplication::getState, FlinkAppStateEnum.MAPPING.getValue())
+            .set(FlinkApplication::getTracking, 1)
+            .update();
+
         FlinkApplication application = getById(appParam.getId());
         if (application.isKubernetesModeJob()) {
             // todo mark
             k8SFlinkTrackMonitor.doWatching(k8sWatcherWrapper.toTrackId(application));
         } else {
             FlinkAppHttpWatcher.doWatching(application);
         }
-        return mapping;
+        return result;
     }
 
     @Override
@@ -674,7 +682,7 @@ public void updateRelease(FlinkApplication appParam) {
 
     @Override
     public List<FlinkApplication> listByProjectId(Long id) {
-        return baseMapper.selectAppsByProjectId(id);
+        return this.lambdaQuery().eq(FlinkApplication::getProjectId, id).list();
     }
 
     @Override

diff --git a/...e/streampark/console/core/service/application/impl/SparkApplicationManageServiceImpl.java b/...e/streampark/console/core/service/application/impl/SparkApplicationManageServiceImpl.java
@@ -152,7 +152,13 @@ public void persistMetrics(SparkApplication appParam) {
 
     @Override
     public boolean mapping(SparkApplication appParam) {
-        return this.baseMapper.mapping(appParam);
+        this.lambdaUpdate()
+            .set(SparkApplication::getClusterId, appParam.getClusterId())
+            .set(SparkApplication::getEndTime, null)
+            .set(SparkApplication::getState, SparkAppStateEnum.MAPPING.getValue())
+            .set(SparkApplication::getTracking, 1)
+            .eq(SparkApplication::getId, appParam.getId())
+            .update();
     }
 
     @Override
@@ -566,7 +572,7 @@ public void updateRelease(SparkApplication appParam) {
 
     @Override
     public List<SparkApplication> listByProjectId(Long id) {
-        return baseMapper.selectAppsByProjectId(id);
+        return this.lambdaQuery().eq(SparkApplication::getProjectId, id).list();
     }
 
     @Override

diff --git a/...rc/main/java/org/apache/streampark/console/core/service/impl/FlinkCatalogServiceImpl.java b/...rc/main/java/org/apache/streampark/console/core/service/impl/FlinkCatalogServiceImpl.java
@@ -26,6 +26,8 @@
 import org.apache.streampark.console.core.mapper.FlinkCatalogMapper;
 import org.apache.streampark.console.core.service.FlinkCatalogService;
 
+import org.apache.commons.lang3.StringUtils;
+
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -79,7 +81,14 @@ public IPage<FlinkCatalogParams> page(FlinkCatalogParams catalog, RestRequest re
             catalog.getTeamId(), "The teamId can't be null. List catalog failed.");
 
         Page<FlinkCatalog> page = MybatisPager.getPage(request);
-        this.baseMapper.selectPage(page, FlinkCatalog.of(catalog));
+
+        this.lambdaQuery()
+            .eq(FlinkCatalog::getTeamId, catalog.getTeamId())
+            .eq(StringUtils.isNotBlank(catalog.getCatalogName()), FlinkCatalog::getCatalogName,
+                catalog.getCatalogName())
+            .eq(catalog.getUserId() != null, FlinkCatalog::getUserId, catalog.getUserId())
+            .page(page);
+
         Page<FlinkCatalogParams> paramsPage = new Page<>();
         BeanUtils.copyProperties(page, paramsPage, "records");
         List<FlinkCatalogParams> paramList = new ArrayList<>();
@@ -99,7 +108,7 @@ public FlinkCatalog getCatalog(Long catalogId) {
 
     @Override
     public FlinkCatalog getCatalog(String catalogName) {
-        return this.baseMapper.selectByCatalogName(catalogName);
+        return this.lambdaQuery().eq(FlinkCatalog::getCatalogName, catalogName).one();
     }
 
     @Override
@@ -123,7 +132,9 @@ public boolean update(FlinkCatalogParams catalogParam, Long userId) {
     }
 
     public Boolean existsByCatalogName(String catalogName) {
-        return this.baseMapper.existsByCatalogName(catalogName);
+        return this.lambdaQuery()
+            .eq(FlinkCatalog::getCatalogName, catalogName)
+            .exists();
     }
 
     /** validate catalog name */

diff --git a/...rc/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java b/...rc/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
@@ -277,12 +277,18 @@ public Boolean allowShutdownCluster(FlinkCluster cluster) {
 
     @Override
     public Boolean existsByClusterId(String clusterId, Long id) {
-        return this.baseMapper.existsByClusterId(clusterId, id);
+        return this.lambdaQuery()
+            .eq(FlinkCluster::getClusterId, clusterId)
+            .ne(id != null, FlinkCluster::getId, id)
+            .exists();
     }
 
     @Override
     public Boolean existsByClusterName(String clusterName, Long id) {
-        return this.baseMapper.existsByClusterName(clusterName, id);
+        return this.lambdaQuery()
+            .eq(FlinkCluster::getClusterName, clusterName)
+            .ne(id != null, FlinkCluster::getId, id)
+            .exists();
     }
 
     @Override

diff --git a/...ice/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java b/...ice/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
@@ -44,6 +44,7 @@
 import org.apache.streampark.console.core.task.ProjectBuildTask;
 import org.apache.streampark.console.core.watcher.FlinkAppHttpWatcher;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.configuration.MemorySize;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
@@ -195,22 +196,34 @@ public boolean removeById(Long id) {
     @Override
     public IPage<Project> getPage(Project project, RestRequest request) {
         Page<Project> page = MybatisPager.getPage(request);
-        return this.baseMapper.selectPage(page, project);
+        this.lambdaQuery()
+            .eq(Project::getTeamId, project.getTeamId())
+            .like(StringUtils.isNotBlank(project.getName()), Project::getName, project.getName())
+            .eq(project.getBuildState() != null, Project::getBuildState, project.getBuildState())
+            .page(page);
+        return page;
     }
 
     @Override
     public Boolean existsByTeamId(Long teamId) {
-        return this.baseMapper.existsByTeamId(teamId);
+        return this.lambdaQuery()
+            .eq(Project::getTeamId, teamId)
+            .exists();
     }
 
     @Override
     public List<Project> listByTeamId(Long teamId) {
-        return this.baseMapper.selectProjectsByTeamId(teamId);
+        return this.lambdaQuery().eq(Project::getTeamId, teamId)
+            .list();
     }
 
     @Override
     public void build(Long id) throws Exception {
-        Long currentBuildCount = this.baseMapper.getBuildingCount();
+
+        Long currentBuildCount = this.lambdaQuery()
+            .eq(Project::getBuildState, BuildStateEnum.BUILDING.get())
+            .count();
+
         ApiAlertException.throwIfTrue(
             maxProjectBuildNum > -1 && currentBuildCount > maxProjectBuildNum,
             String.format(

diff --git a/...ce/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java b/...ce/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java
@@ -123,7 +123,8 @@ public IPage<Resource> getPage(Resource resource, RestRequest request) {
      */
     @Override
     public boolean existsByUserId(Long userId) {
-        return this.baseMapper.existsByUserId(userId);
+        return this.lambdaQuery().eq(Resource::getCreatorId, userId)
+            .exists();
     }
 
     @Override

diff --git a/...ce/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java b/...ce/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
@@ -169,7 +169,11 @@ public List<Variable> listByTeamId(Long teamId) {
      */
     @Override
     public List<Variable> listByTeamId(Long teamId, String keyword) {
-        return baseMapper.selectVarsByTeamId(teamId, keyword);
+        return this.lambdaQuery().eq(Variable::getTeamId, teamId)
+            .and(StringUtils.isNotBlank(keyword), c -> c.like(Variable::getVariableCode, keyword)
+                .or()
+                .like(Variable::getDescription, keyword))
+            .list();
     }
 
     /**
@@ -255,6 +259,7 @@ private String getCodeFromPlaceholder(String placeholder) {
 
     @Override
     public boolean existsByTeamId(Long teamId) {
-        return this.baseMapper.existsByTeamId(teamId);
+        return this.lambdaQuery().eq(Variable::getTeamId, teamId)
+            .exists();
     }
 }
diff --git a/...e/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java b/...e/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java
@@ -112,7 +112,10 @@ public ResponseResult<String> checkYarnQueue(YarnQueue yarnQueue) {
             return responseResult;
         }
 
-        boolean existed = this.baseMapper.existsByQueueLabel(yarnQueue);
+        boolean existed = this.lambdaQuery().eq(YarnQueue::getTeamId, yarnQueue.getTeamId())
+            .eq(YarnQueue::getQueueLabel, yarnQueue.getQueueLabel())
+            .ne(YarnQueue::getId, yarnQueue.getId())
+            .exists();
 
         if (existed) {
             responseResult.setStatus(1);