Skip to content

fix(semantic): deduplicate synced semantic metadata #90

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mengnankkkk merged 2 commits into from
Aug 9, 2026
Merged

fix(semantic): deduplicate synced semantic metadata #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
87c9b3f
fix(semantic): deduplicate synced semantic metadata
mengnankkkk Aug 9, 2026
83cd153
fix: ci
mengnankkkk Aug 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 34 additions & 14 deletions ...nd/src/main/java/io/github/malonetalk/service/semantic/sync/SemanticSyncApplyService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,7 @@ public List<SyncTableResult> applyTableSync(
Map<String, List<ColumnInfo>> columnsByTableName =
loadSemanticColumnsByTable(datasourceId, selectedTableNames);

// 表和列分别执行一次批量 upsert，不按表或列逐条写库。
batchUpsertPresentSchema(datasourceId, presentTables);
batchSavePresentSchema(datasourceId, presentTables, existingTableIndex, columnsByTableName);

// 基于写入前快照生成统计结果，同时收集需要批量标记缺失的列 ID。
List<String> missingTableNames = new ArrayList<>();
Expand Down Expand Up @@ -151,25 +150,46 @@ public List<SyncTableResult> refreshPhysicalStatus(
return results;
}

private void batchUpsertPresentSchema(
Integer datasourceId, List<TableSyncSource> presentTables) {
private void batchSavePresentSchema(
Integer datasourceId,
List<TableSyncSource> presentTables,
Map<String, TableInfo> existingTableIndex,
Map<String, List<ColumnInfo>> columnsByTableName) {
if (presentTables.isEmpty()) {
return;
}

tableInfoMapper.batchUpsertPhysicalCache(
presentTables.stream()
.map(table -> buildPhysicalTableInfo(datasourceId, table))
.toList());
List<ColumnInfo> physicalColumns = new ArrayList<>();
List<TableInfo> newTables = new ArrayList<>();
List<ColumnInfo> newColumns = new ArrayList<>();
for (TableSyncSource table : presentTables) {
TableInfo tableInfo = buildPhysicalTableInfo(datasourceId, table);
TableInfo existingTable = existingTableIndex.get(table.tableName());
if (existingTable == null) {
newTables.add(tableInfo);
} else {
tableInfo.setId(existingTable.getId());
tableInfoMapper.updatePhysicalCacheFields(tableInfo);
}

Map<String, ColumnInfo> existingColumnIndex =
loadColumnIndex(columnsByTableName.getOrDefault(table.tableName(), List.of()));
for (ColumnSyncSource column : table.columns()) {
physicalColumns.add(
buildPhysicalColumnInfo(datasourceId, table.tableName(), column));
ColumnInfo columnInfo =
buildPhysicalColumnInfo(datasourceId, table.tableName(), column);
ColumnInfo existingColumn = existingColumnIndex.get(column.columnName());
if (existingColumn == null) {
newColumns.add(columnInfo);
} else {
columnInfo.setId(existingColumn.getId());
columnSemanticInfoMapper.updatePhysicalCacheFields(columnInfo);
}
}
}
if (!physicalColumns.isEmpty()) {
columnSemanticInfoMapper.batchUpsertPhysicalCache(physicalColumns);
if (!newTables.isEmpty()) {
tableInfoMapper.batchUpsertPhysicalCache(newTables);
}
if (!newColumns.isEmpty()) {
columnSemanticInfoMapper.batchUpsertPhysicalCache(newColumns);
}
}

Expand Down Expand Up @@ -287,7 +307,7 @@ private Map<String, List<ColumnInfo>> loadSemanticColumnsByTable(
private Map<String, ColumnInfo> loadColumnIndex(List<ColumnInfo> columns) {
Map<String, ColumnInfo> index = new LinkedHashMap<>();
for (ColumnInfo column : columns) {
index.put(
index.putIfAbsent(
SemanticUtils.normalizeObjectName(
column.getColumnName(), "Missing semantic columnName."),
column);
Expand Down
43 changes: 37 additions & 6 deletions data-agent-backend/src/main/resources/mapper/ColumnSemanticInfoMapper.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,21 +17,49 @@
<result column="update_time" property="updateTime"/>
</resultMap>

<sql id="DeduplicatedColumns">
SELECT *
FROM (
SELECT
column_info.*,
ROW_NUMBER() OVER (
PARTITION BY datasource_id, LOWER(table_name), LOWER(column_name)
ORDER BY
CASE
WHEN column_description IS NOT NULL AND column_description &lt;&gt; '' THEN 0
WHEN is_visible = 0 THEN 0
ELSE 1
END ASC,
update_time DESC,
create_time DESC,
id DESC
) AS row_num
FROM column_info
) deduplicated_columns
WHERE row_num = 1
</sql>

<select id="selectByDatasourceId" resultMap="BaseResultMap">
SELECT * FROM column_info
SELECT * FROM (
<include refid="DeduplicatedColumns"/>
) column_info
WHERE datasource_id = #{datasourceId}
ORDER BY table_name ASC, column_name ASC, id ASC
</select>

<select id="selectByDatasourceIdAndTableName" resultMap="BaseResultMap">
SELECT * FROM column_info
SELECT * FROM (
<include refid="DeduplicatedColumns"/>
) column_info
WHERE datasource_id = #{datasourceId}
AND LOWER(table_name) = LOWER(#{tableName})
ORDER BY column_name ASC, id ASC
</select>

<select id="selectByDatasourceIdAndTableNames" resultMap="BaseResultMap">
SELECT * FROM column_info
SELECT * FROM (
<include refid="DeduplicatedColumns"/>
) column_info
WHERE datasource_id = #{datasourceId}
AND LOWER(table_name) IN
<foreach collection="tableNames" item="tableName" open="(" separator="," close=")">
Expand All @@ -41,7 +69,9 @@
</select>

<select id="selectPageByDatasourceIdAndTableName" resultMap="BaseResultMap">
SELECT * FROM column_info
SELECT * FROM (
<include refid="DeduplicatedColumns"/>
) column_info
WHERE datasource_id = #{query.datasourceId}
AND LOWER(table_name) = LOWER(#{query.tableName})
<if test="query.keyword != null and query.keyword != ''">
Expand All @@ -58,11 +88,12 @@
</select>

<select id="selectByDatasourceIdAndTableNameAndColumnName" resultMap="BaseResultMap">
SELECT * FROM column_info
SELECT * FROM (
<include refid="DeduplicatedColumns"/>
) column_info
WHERE datasource_id = #{datasourceId}
AND LOWER(table_name) = LOWER(#{tableName})
AND LOWER(column_name) = LOWER(#{columnName})
ORDER BY update_time DESC, create_time DESC, id DESC
LIMIT 1
</select>

Expand Down
46 changes: 40 additions & 6 deletions data-agent-backend/src/main/resources/mapper/LogicalTableRelationMapper.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,25 +18,56 @@
<result column="update_time" property="updateTime"/>
</resultMap>

<sql id="DeduplicatedRelations">
SELECT *
FROM (
SELECT
logical_table_relation.*,
ROW_NUMBER() OVER (
PARTITION BY
datasource_id,
LOWER(source_table_name),
source_column_signature
ORDER BY
CASE
WHEN description IS NOT NULL AND description &lt;&gt; '' THEN 0
WHEN is_enabled = 0 THEN 0
ELSE 1
END ASC,
update_time DESC,
create_time DESC,
id DESC
) AS row_num
FROM logical_table_relation
) deduplicated_relations
WHERE row_num = 1
</sql>

<select id="selectById" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation WHERE id = #{id}
</select>

<select id="selectByDatasourceId" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation
SELECT * FROM (
<include refid="DeduplicatedRelations"/>
) logical_table_relation
WHERE datasource_id = #{datasourceId}
ORDER BY create_time DESC, id DESC
</select>

<select id="selectByDatasourceIdAndSourceTable" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation
SELECT * FROM (
<include refid="DeduplicatedRelations"/>
) logical_table_relation
WHERE datasource_id = #{datasourceId}
AND LOWER(source_table_name) = LOWER(#{sourceTableName})
ORDER BY id DESC
</select>

<select id="selectByDatasourceIdAndSourceTables" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation
SELECT * FROM (
<include refid="DeduplicatedRelations"/>
) logical_table_relation
WHERE datasource_id = #{datasourceId}
AND LOWER(source_table_name) IN
<foreach collection="sourceTableNames" item="sourceTableName" open="(" separator="," close=")">
Expand All @@ -46,7 +77,9 @@
</select>

<select id="selectPageByDatasourceIdAndSourceTable" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation
SELECT * FROM (
<include refid="DeduplicatedRelations"/>
) logical_table_relation
WHERE datasource_id = #{query.datasourceId}
AND LOWER(source_table_name) = LOWER(#{query.tableName})
<if test="query.keyword != null and query.keyword != ''">
Expand All @@ -66,11 +99,12 @@
</select>

<select id="selectByUniqueSourceKey" resultMap="BaseResultMap">
SELECT * FROM logical_table_relation
SELECT * FROM (
<include refid="DeduplicatedRelations"/>
) logical_table_relation
WHERE datasource_id = #{datasourceId}
AND LOWER(source_table_name) = LOWER(#{sourceTableName})
AND source_column_signature = #{sourceColumnSignature}
ORDER BY update_time DESC, create_time DESC, id DESC
LIMIT 1
</select>

Expand Down
47 changes: 40 additions & 7 deletions data-agent-backend/src/main/resources/mapper/TableInfoMapper.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,40 @@
<result column="update_time" property="updateTime"/>
</resultMap>

<sql id="DeduplicatedTables">
SELECT *
FROM (
SELECT
table_info.*,
ROW_NUMBER() OVER (
PARTITION BY datasource_id, LOWER(table_name)
ORDER BY
CASE
WHEN table_description IS NOT NULL AND table_description &lt;&gt; '' THEN 0
WHEN is_visible = 0 THEN 0
ELSE 1
END ASC,
update_time DESC,
create_time DESC,
id DESC
) AS row_num
FROM table_info
) deduplicated_tables
WHERE row_num = 1
</sql>

<select id="selectByDatasourceId" resultMap="BaseResultMap">
SELECT * FROM table_info
SELECT * FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE datasource_id = #{datasourceId}
ORDER BY table_name ASC, id ASC
</select>

<select id="selectPageByDatasourceId" resultMap="BaseResultMap">
SELECT * FROM table_info
SELECT * FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE datasource_id = #{query.datasourceId}
<if test="query.keyword != null and query.keyword != ''">
AND LOWER(table_name) LIKE CONCAT('%', LOWER(#{query.keyword}), '%')
Expand All @@ -38,15 +64,18 @@
</select>

<select id="selectByDatasourceIdAndTableName" resultMap="BaseResultMap">
SELECT * FROM table_info
SELECT * FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE datasource_id = #{datasourceId}
AND LOWER(table_name) = LOWER(#{tableName})
ORDER BY update_time DESC, create_time DESC, id DESC
LIMIT 1
</select>

<select id="selectByDatasourceIdAndTableNames" resultMap="BaseResultMap">
SELECT * FROM table_info
SELECT * FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE datasource_id = #{datasourceId}
AND LOWER(table_name) IN
<foreach collection="tableNames" item="tableName" open="(" separator="," close=")">
Expand Down Expand Up @@ -122,7 +151,9 @@
</update>

<select id="selectByDatasourceIdAndDomains" resultMap="BaseResultMap">
SELECT * FROM table_info
SELECT * FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE datasource_id = #{datasourceId}
AND domain IN
<foreach collection="domains" item="domain" open="(" separator="," close=")">
Expand All @@ -132,7 +163,9 @@
</select>

<select id="countByDomain" resultType="int">
SELECT COUNT(1) FROM table_info
SELECT COUNT(1) FROM (
<include refid="DeduplicatedTables"/>
) table_info
WHERE LOWER(domain) = LOWER(#{domain})
</select>

Expand Down
Loading