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

feat(api): 对于需要记录最后时间的字段使用比较方式更新

Browse Source
This commit is contained in:
v-zhangjc9
2024-05-09 18:22:42 +08:00
parent bfdc5743ab
commit e35733a2d8
1 changed files with 41 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
service-api/src/main/java/com/lanyuanxiaoyao/service/api/service/SyncStateService.java
View File

@@ -3,6 +3,7 @@ package com.lanyuanxiaoyao.service.api.service;
import club.kingon.sql.builder.SqlBuilder;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.api.entity.HoodieCommitMetadata;
import com.lanyuanxiaoyao.service.common.Constants;
import java.time.Instant;
@@ -119,9 +120,18 @@ public class SyncStateService {
)
.values()
.addValue(null, null)
.onDuplicateKeyUpdateColumn(TbAppHudiSyncState.SOURCE_PUBLISH_TIME_O)
.onDuplicateKeyUpdateSetter(
StrUtil.format(
"{} = if({} is null, ?, greatest({}, ?))",
TbAppHudiSyncState.SOURCE_PUBLISH_TIME_O,
TbAppHudiSyncState.SOURCE_PUBLISH_TIME_O,
TbAppHudiSyncState.SOURCE_PUBLISH_TIME_O
)
)
.precompileSql(),
syncStateId(flinkJobId, alias),
publishDate,
publishDate,
publishDate
);
}
@@ -137,9 +147,18 @@ public class SyncStateService {
)
.values()
.addValue(null, null)
.onDuplicateKeyUpdateColumn(TbAppHudiSyncState.SOURCE_RECEIVE_TIME_O)
.onDuplicateKeyUpdateSetter(
StrUtil.format(
"{} = if({} is null, ?, greatest({}, ?))",
TbAppHudiSyncState.SOURCE_RECEIVE_TIME_O,
TbAppHudiSyncState.SOURCE_RECEIVE_TIME_O,
TbAppHudiSyncState.SOURCE_RECEIVE_TIME_O
)
)
.precompileSql(),
syncStateId(flinkJobId, alias),
receiveDate,
receiveDate,
receiveDate
);
}
@@ -155,9 +174,18 @@ public class SyncStateService {
)
.values()
.addValue(null, null)
.onDuplicateKeyUpdateColumn(TbAppHudiSyncState.SOURCE_OP_TIME_O)
.onDuplicateKeyUpdateSetter(
StrUtil.format(
"{} = if({} is null, ?, greatest({}, ?))",
TbAppHudiSyncState.SOURCE_OP_TIME_O,
TbAppHudiSyncState.SOURCE_OP_TIME_O,
TbAppHudiSyncState.SOURCE_OP_TIME_O
)
)
.precompileSql(),
syncStateId(flinkJobId, alias),
operationDate,
operationDate,
operationDate
);
}
@@ -299,9 +327,18 @@ public class SyncStateService {
)
.values()
.addValue(null, null)
.onDuplicateKeyUpdateColumn(TbAppHudiSyncState.COMPACTION_LATEST_OP_TS_O)
.onDuplicateKeyUpdateSetter(
StrUtil.format(
"{} = if({} is null, ?, greatest({}, ?))",
TbAppHudiSyncState.COMPACTION_LATEST_OP_TS_O,
TbAppHudiSyncState.COMPACTION_LATEST_OP_TS_O,
TbAppHudiSyncState.COMPACTION_LATEST_OP_TS_O
)
)
.precompileSql(),
syncStateId(flinkJobId, alias),
operationDate,
operationDate,
operationDate
);
}