lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(info-query): 增加单表跨天查询

Browse Source
This commit is contained in:
v-zhangjc9
2024-07-05 11:38:42 +08:00
parent db38686957
commit d99e5b261b
3 changed files with 30 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/InfoService.java
View File

@@ -116,6 +116,9 @@ public interface InfoService {
@Get("/info/updated_version_tables")
ImmutableList<String> updatedVersionTables();
@Get("/info/is_version_updated")
Boolean isVersionUpdated(@Query("flink_job_id") Long flinkJobId, @Query("alias") String alias);
@Get("/info/exists_table")
Boolean existsTable(@Query("flink_job_id") Long flinkJobId, @Query("alias") String alias);

5
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/controller/VersionController.java
View File

@@ -103,4 +103,9 @@ public class VersionController {
public Long unReceiveVersionFocusTableCount(@RequestParam("version") String version) {
return versionService.unReceiveVersionFocusTableCount(version);
}
@GetMapping("/is_version_updated")
public Boolean isVersionUpdated(Long flinkJobId, String alias) {
return versionService.isVersionUpdated(flinkJobId, alias);
}
}

24
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/service/VersionService.java
View File

@@ -66,8 +66,8 @@ public class VersionService extends BaseService {
return builder
.from(TbAppFlinkJobConfig._alias_, TbAppCollectTableInfo._alias_, TbAppCollectTableVersion._alias_)
.whereEq(TbAppFlinkJobConfig.ID_A, Column.as(TbAppCollectTableInfo.FLINK_JOB_ID_A))
.andEq(TbAppFlinkJobConfig.STATUS_A, "y")
.andEq(TbAppCollectTableInfo.STATUS_A, "y")
.andEq(TbAppFlinkJobConfig.STATUS_A, STATUS_Y)
.andEq(TbAppCollectTableInfo.STATUS_A, STATUS_Y)
.andEq(TbAppFlinkJobConfig.ID_A, Column.as(TbAppCollectTableVersion.FLINK_JOB_ID_A))
.andEq(TbAppCollectTableInfo.ALIAS_A, Column.as(TbAppCollectTableVersion.ALIAS_A))
.andLt(TbAppCollectTableInfo.PRIORITY_A, 10000)
@@ -351,4 +351,24 @@ public class VersionService extends BaseService {
Long.class
);
}
@Cacheable(value = "is-version-updated", sync = true)
@Retryable(Throwable.class)
public Boolean isVersionUpdated(Long flinkJobId, String alias) {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(*) > 0")
.from(TbAppFlinkJobConfig._alias_, TbAppCollectTableInfo._alias_, TbAppCollectTableVersion._alias_)
.whereEq(TbAppFlinkJobConfig.ID_A, Column.as(TbAppCollectTableInfo.FLINK_JOB_ID_A))
.andEq(TbAppFlinkJobConfig.STATUS_A, STATUS_Y)
.andEq(TbAppCollectTableInfo.STATUS_A, STATUS_Y)
.andEq(TbAppFlinkJobConfig.ID_A, Column.as(TbAppCollectTableVersion.FLINK_JOB_ID_A))
.andEq(TbAppCollectTableInfo.ALIAS_A, Column.as(TbAppCollectTableVersion.ALIAS_A))
.andEq(TbAppCollectTableInfo.FLINK_JOB_ID_A, flinkJobId)
.andEq(TbAppCollectTableInfo.ALIAS_A, alias)
.andEq(TbAppCollectTableVersion.VERSION_A, Column.as("date_format(subdate(current_date(), 1), '%Y%m%d')"))
.andEq(TbAppCollectTableVersion.SCHEDULED_A, true)
.build(),
Boolean.class
);
}
}