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

feat(info-query): 补充查询table meta的方法统一使用同一种判断

Browse Source
This commit is contained in:
兰缘小妖
2024-01-24 18:24:08 +08:00
parent 8ba4296670
commit 526f29d87e
2 changed files with 64 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/service/InfoService.java
View File

@@ -58,20 +58,12 @@ public class InfoService extends BaseService {
) {
int limit = Math.max(count, 1);
int offset = limit * Math.max(page - 1, 0);
return builder.from(
TbAppFlinkJobConfig._alias_,
TbAppCollectTableInfo._alias_,
TbAppHudiSyncState._alias_
)
.whereEq(TbAppFlinkJobConfig.ID_A, Column.as(TbAppCollectTableInfo.FLINK_JOB_ID_A))
.andEq(TbAppHudiSyncState.ID_A, Column.as(StrUtil.format("concat({}, '-', {})", TbAppFlinkJobConfig.ID_A, TbAppCollectTableInfo.ALIAS_A)))
return TableMetaService.generateSimpleTableMetaList(builder)
.andLike(ObjectUtil.isNotNull(flinkJobId), TbAppFlinkJobConfig.ID_A, flinkJobId)
.andLike(ObjectUtil.isNotNull(alias), TbAppCollectTableInfo.ALIAS_A, alias)
.andIn(ObjectUtil.isNotEmpty(selectHudiTableType), TbAppCollectTableInfo.TGT_TABLE_A, selectHudiTableType)
.andIn(ObjectUtil.isNotEmpty(selectedRunMode), TbAppFlinkJobConfig.RUN_MODE_A, selectedRunMode)
.andIn(ObjectUtil.isNotEmpty(selectedCompactionStatus), TbAppHudiSyncState.COMPACTION_STATUS_A, selectedCompactionStatus)
.andEq(TbAppFlinkJobConfig.STATUS_A, STATUS_Y)
.andEq(TbAppCollectTableInfo.STATUS_A, STATUS_Y)
.orderBy(StrUtil.isNotBlank(order) && StrUtil.isNotBlank(direction), () -> StrUtil.format("{} {}", order, direction))
.limit(limited, offset, limit);
}

93
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/service/TableMetaService.java
View File

@@ -1,7 +1,9 @@
package com.lanyuanxiaoyao.service.info.service;
import club.kingon.sql.builder.ConditionSqlBuilder;
import club.kingon.sql.builder.SelectSqlBuilder;
import club.kingon.sql.builder.SqlBuilder;
import club.kingon.sql.builder.WhereSqlBuilder;
import club.kingon.sql.builder.entry.Alias;
import club.kingon.sql.builder.entry.Column;
import cn.hutool.core.collection.IterUtil;
@@ -67,28 +69,28 @@ public class TableMetaService extends BaseService {
return tableMetaList(flinkJobId, null);
}
private SqlBuilder generateTableMetaList(SelectSqlBuilder builder) {
return generateTableMetaList(builder, null, null, null, null);
public static ConditionSqlBuilder<WhereSqlBuilder> generateSimpleTableMetaList(SelectSqlBuilder builder) {
return generateSimpleTableMetaList(builder, null, null, null, null);
}
private SqlBuilder generateTableMetaList(SelectSqlBuilder builder, Integer priority) {
return generateTableMetaList(builder, null, null, priority, null);
public static ConditionSqlBuilder<WhereSqlBuilder> generateSimpleTableMetaList(SelectSqlBuilder builder, Integer priority) {
return generateSimpleTableMetaList(builder, null, null, priority, null);
}
private SqlBuilder generateTableMetaList(SelectSqlBuilder builder, Long flinkJobId, String aliasText) {
return generateTableMetaList(builder, flinkJobId, aliasText, null, null);
public static ConditionSqlBuilder<WhereSqlBuilder> generateSimpleTableMetaList(SelectSqlBuilder builder, Long flinkJobId, String aliasText) {
return generateSimpleTableMetaList(builder, flinkJobId, aliasText, null, null);
}
private SqlBuilder generateTableMetaList(SelectSqlBuilder builder, String hdfs) {
return generateTableMetaList(builder, null, null, null, hdfs);
public static ConditionSqlBuilder<WhereSqlBuilder> generateSimpleTableMetaList(SelectSqlBuilder builder, String hdfs) {
return generateSimpleTableMetaList(builder, null, null, null, hdfs);
}
private SqlBuilder generateTableMetaList(SelectSqlBuilder builder, Long flinkJobId, String aliasText, Integer priority, String hdfs) {
// 相比正式的table meta查询少了表字段的匹配，大幅提高了查询速度
public static ConditionSqlBuilder<WhereSqlBuilder> generateSimpleTableMetaList(SelectSqlBuilder builder, Long flinkJobId, String aliasText, Integer priority, String hdfs) {
return builder
.from(
DataSource._alias_,
DataSourceTable._alias_,
DataSourceTableField._alias_,
TbAppFlinkJobConfig._alias_,
TbAppHudiJobConfig._alias_,
Alias.of(TbAppYarnJobConfig._origin_, "tayjc_sync"),
@@ -102,8 +104,6 @@ public class TableMetaService extends BaseService {
.andEq(DataSource.RECORD_STATE_A, STATUS_Y)
.andEq(DataSourceTable.DS_ID_A, Column.as(DataSource.DS_ID_A))
.andEq(DataSourceTable.RECORD_STATE_A, STATUS_Y)
.andEq(DataSourceTableField.TABLE_ID_A, Column.as(DataSourceTable.TABLE_ID_A))
.andEq(DataSourceTableField.RECORD_STATE_A, STATUS_Y)
.andIn(DataSource.DS_TYPE_A, "udal", "telepg")
.andEq(DataSource.DS_NAME_A, Column.as(TbAppCollectTableInfo.SRC_DB_A))
.andEq(DataSource.SCHEMA_NAME_A, Column.as(TbAppCollectTableInfo.SRC_SCHEMA_A))
@@ -122,15 +122,13 @@ public class TableMetaService extends BaseService {
.andEq(TbAppFlinkJobConfig.STATUS_A, STATUS_Y)
.andEq(TbAppHudiJobConfig.STATUS_A, STATUS_Y)
.andEq("tayjc_sync.status", STATUS_Y)
.andEq("tayjc_compaction.status", STATUS_Y)
.orderBy(DataSourceTableField.FIELD_SEQ_A);
.andEq("tayjc_compaction.status", STATUS_Y);
}
private ImmutableList<TableMeta> tableMetaList(Long flinkJobId, String aliasText) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateTableMetaList(
SqlBuilder.select(
SqlBuilder.select(
DataSource.DS_NAME_A,
DataSource.SCHEMA_NAME_A,
DataSourceTable.TABLE_NAME_A,
@@ -190,10 +188,45 @@ public class TableMetaService extends BaseService {
TbAppGlobalConfig.ZK_URL_A,
TbAppCollectTableInfo.VERSION_A,
DataSourceTableField.SCALE_A
),
flinkJobId,
aliasText
).build(),
)
.from(
DataSource._alias_,
DataSourceTable._alias_,
DataSourceTableField._alias_,
TbAppFlinkJobConfig._alias_,
TbAppHudiJobConfig._alias_,
Alias.of(TbAppYarnJobConfig._origin_, "tayjc_sync"),
Alias.of(TbAppYarnJobConfig._origin_, "tayjc_compaction"),
TbAppGlobalConfig._alias_,
TbAppCollectTableInfo._alias_,
TbAppHudiSyncState._alias_
)
.whereEq(DataSource.DS_ROLE_A, "src")
.andEq(DataSource.DS_STATE_A, STATUS_Y)
.andEq(DataSource.RECORD_STATE_A, STATUS_Y)
.andEq(DataSourceTable.DS_ID_A, Column.as(DataSource.DS_ID_A))
.andEq(DataSourceTable.RECORD_STATE_A, STATUS_Y)
.andEq(DataSourceTableField.TABLE_ID_A, Column.as(DataSourceTable.TABLE_ID_A))
.andEq(DataSourceTableField.RECORD_STATE_A, STATUS_Y)
.andIn(DataSource.DS_TYPE_A, "udal", "telepg")
.andEq(DataSource.DS_NAME_A, Column.as(TbAppCollectTableInfo.SRC_DB_A))
.andEq(DataSource.SCHEMA_NAME_A, Column.as(TbAppCollectTableInfo.SRC_SCHEMA_A))
.andEq(DataSourceTable.TABLE_NAME_A, Column.as(TbAppCollectTableInfo.SRC_TABLE_A))
.andEq(TbAppCollectTableInfo.FLINK_JOB_ID_A, Column.as(TbAppFlinkJobConfig.ID_A))
.andEq(TbAppCollectTableInfo.HUDI_JOB_ID_A, Column.as(TbAppHudiJobConfig.ID_A))
.andEq(TbAppCollectTableInfo.SYNC_YARN_JOB_ID_A, Column.as("tayjc_sync.id"))
.andEq(TbAppCollectTableInfo.COMPACTION_YARN_JOB_ID_A, Column.as("tayjc_compaction.id"))
.andEq(TbAppCollectTableInfo.CONFIG_ID_A, Column.as(TbAppGlobalConfig.ID_A))
.andEq(TbAppHudiSyncState.ID_A, Column.as(StrUtil.format("concat({}, '-', {})", TbAppFlinkJobConfig.ID_A, TbAppCollectTableInfo.ALIAS_A)))
.andEq(ObjectUtil.isNotNull(flinkJobId), TbAppFlinkJobConfig.ID_A, flinkJobId)
.andEq(StrUtil.isNotBlank(aliasText), TbAppCollectTableInfo.ALIAS_A, aliasText)
.andEq(TbAppCollectTableInfo.STATUS_A, STATUS_Y)
.andEq(TbAppFlinkJobConfig.STATUS_A, STATUS_Y)
.andEq(TbAppHudiJobConfig.STATUS_A, STATUS_Y)
.andEq("tayjc_sync.status", STATUS_Y)
.andEq("tayjc_compaction.status", STATUS_Y)
.orderBy(DataSourceTableField.FIELD_SEQ_A)
.build(),
(rs, row) -> TableMeta.RowMeta.builder()
.dsName(rs.getString(1))
.schemaName(rs.getString(2))
@@ -483,7 +516,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long tableCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct concat({}, {}))", TbAppCollectTableInfo.SRC_SCHEMA_A, TbAppCollectTableInfo.SRC_TABLE_A))
).build(),
Long.class
@@ -494,7 +527,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long tableFocusCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct concat({}, {}))", TbAppCollectTableInfo.SRC_SCHEMA_A, TbAppCollectTableInfo.SRC_TABLE_A)),
10000
).build(),
@@ -506,7 +539,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long hudiCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct {}) as count", TbAppCollectTableInfo.TGT_HDFS_PATH_A))
).build(),
Long.class
@@ -517,7 +550,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long hudiFocusCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct {}) as count", TbAppCollectTableInfo.TGT_HDFS_PATH_A)),
10000
).build(),
@@ -529,7 +562,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long hiveCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct concat({}, {})) as count", TbAppCollectTableInfo.HIVE_DB_A, TbAppCollectTableInfo.HIVE_TABLE_A))
).build(),
Long.class
@@ -540,7 +573,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Long hiveFocusCount() {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(StrUtil.format("count(distinct concat({}, {})) as count", TbAppCollectTableInfo.HIVE_DB_A, TbAppCollectTableInfo.HIVE_TABLE_A)),
10000
).build(),
@@ -553,7 +586,7 @@ public class TableMetaService extends BaseService {
public ImmutableList<SimpleTableMeta> simpleTableMetas(Long flinkJobId, String alias) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(
TbAppFlinkJobConfig.ID_A,
TbAppFlinkJobConfig.NAME_A,
@@ -585,7 +618,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Boolean existsTable(Long flinkJobId, String alias) {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select("count(*) > 0"),
flinkJobId,
alias
@@ -598,7 +631,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public Boolean existsTableByHdfs(String hdfs) {
return mysqlJdbcTemplate.queryForObject(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select("count(*) > 0"),
hdfs
).build(),
@@ -682,7 +715,7 @@ public class TableMetaService extends BaseService {
@Retryable(Throwable.class)
public ImmutableList<TableInfoSearchCache> allTableInfoSearchCache() {
return Lists.immutable.ofAll(mysqlJdbcTemplate.query(
generateTableMetaList(
generateSimpleTableMetaList(
SqlBuilder.select(
TbAppFlinkJobConfig.ID_A,
TbAppCollectTableInfo.ALIAS_A,