Skip to content

fix: Use native metadata #121

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 1 commit into from
Aug 15, 2026
+64 −62
Merged

fix: Use native metadata #121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
41 changes: 3 additions & 38 deletions data-agent-backend/src/main/java/io/github/malonetalk/agent/datasource/SchemaReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import lombok.AllArgsConstructor;
Expand Down Expand Up @@ -64,9 +63,8 @@ public List<PhysicalColumnInfo> getTableSchema(Datasource datasource, String tab
javax.sql.DataSource ds = dynamicDataSourceManager.getOrCreateDataSource(datasource);

try (Connection conn = ds.getConnection()) {
String metadataTableName = resolveMetadataTableName(conn, tableName);
Set<String> primaryKeys = getPrimaryKeys(conn, metadataTableName);
return getColumns(conn, metadataTableName, primaryKeys);
Set<String> primaryKeys = getPrimaryKeys(conn, tableName);
return getColumns(conn, tableName, primaryKeys);
} catch (SQLException e) {
log.error("Failed to read schema for table {}: {}", tableName, e.getMessage(), e);
throw BusinessException.of(
Expand All @@ -82,10 +80,8 @@ public Map<String, Set<String>> getTableColumnNames(
return Map.of();
}
Map<String, Set<String>> columnNamesByTable = new LinkedHashMap<>();
Map<String, String> tableNameLookup = new LinkedHashMap<>();
for (String tableName : tableNames) {
columnNamesByTable.putIfAbsent(tableName, new LinkedHashSet<>());
tableNameLookup.putIfAbsent(caseInsensitiveKey(tableName), tableName);
}

javax.sql.DataSource ds = dynamicDataSourceManager.getOrCreateDataSource(datasource);
Expand All @@ -95,12 +91,7 @@ public Map<String, Set<String>> getTableColumnNames(
try (ResultSet rs =
metaData.getColumns(conn.getCatalog(), conn.getSchema(), "%", null)) {
while (rs.next()) {
String metadataTableName = rs.getString("TABLE_NAME");
String tableName =
columnNamesByTable.containsKey(metadataTableName)
? metadataTableName
: tableNameLookup.get(caseInsensitiveKey(metadataTableName));
Set<String> columnNames = columnNamesByTable.get(tableName);
Set<String> columnNames = columnNamesByTable.get(rs.getString("TABLE_NAME"));
if (columnNames != null) {
columnNames.add(rs.getString("COLUMN_NAME"));
}
Expand Down Expand Up @@ -132,32 +123,6 @@ private List<PhysicalTableInfo> getTables(Connection conn) throws SQLException {
return tables;
}

private String resolveMetadataTableName(Connection conn, String tableName) throws SQLException {
DatabaseMetaData metaData = conn.getMetaData();
String catalog = conn.getCatalog();
String schema = conn.getSchema();
String[] types = new String[] {"TABLE"};

try (ResultSet rs = metaData.getTables(catalog, schema, tableName, types)) {
if (rs.next()) {
return rs.getString("TABLE_NAME");
}
}
try (ResultSet rs = metaData.getTables(catalog, schema, "%", types)) {
while (rs.next()) {
String metadataTableName = rs.getString("TABLE_NAME");
if (metadataTableName != null && metadataTableName.equalsIgnoreCase(tableName)) {
return metadataTableName;
}
}
}
return tableName;
}

private String caseInsensitiveKey(String value) {
return value == null ? null : value.toLowerCase(Locale.ROOT);
}

private Set<String> getPrimaryKeys(Connection conn, String tableName) throws SQLException {
Set<String> pkColumns = new HashSet<>();
DatabaseMetaData metaData = conn.getMetaData();
Expand Down
35 changes: 22 additions & 13 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 @@ -68,19 +68,21 @@ public List<SyncTableResult> applyTableSync(
List<Integer> missingColumnIds = new ArrayList<>();
List<SyncTableResult> results = new ArrayList<>();
for (TableSyncSource table : presentTables) {
String tableKey = tableKey(table.tableName());
results.add(
buildPresentTableResult(
table,
existingTableIndex.get(table.tableName()),
columnsByTableName.getOrDefault(table.tableName(), List.of()),
existingTableIndex.get(tableKey),
columnsByTableName.getOrDefault(tableKey, List.of()),
missingColumnIds));
}
for (String missingTable : missingTables) {
String tableKey = tableKey(missingTable);
SyncTableResult result =
buildMissingTableResult(
missingTable,
existingTableIndex.get(missingTable),
columnsByTableName.getOrDefault(missingTable, List.of()),
existingTableIndex.get(tableKey),
columnsByTableName.getOrDefault(tableKey, List.of()),
missingColumnIds);
results.add(result);
if (result.tableMarkedAsMissing()) {
Expand Down Expand Up @@ -162,8 +164,9 @@ private void batchSavePresentSchema(
List<TableInfo> newTables = new ArrayList<>();
List<ColumnInfo> newColumns = new ArrayList<>();
for (TableSyncSource table : presentTables) {
String tableKey = tableKey(table.tableName());
TableInfo tableInfo = buildPhysicalTableInfo(datasourceId, table);
TableInfo existingTable = existingTableIndex.get(table.tableName());
TableInfo existingTable = existingTableIndex.get(tableKey);
if (existingTable == null) {
newTables.add(tableInfo);
} else {
Expand All @@ -172,11 +175,11 @@ private void batchSavePresentSchema(
}

Map<String, ColumnInfo> existingColumnIndex =
loadColumnIndex(columnsByTableName.getOrDefault(table.tableName(), List.of()));
loadColumnIndex(columnsByTableName.getOrDefault(tableKey, List.of()));
for (ColumnSyncSource column : table.columns()) {
ColumnInfo columnInfo =
buildPhysicalColumnInfo(datasourceId, table.tableName(), column);
ColumnInfo existingColumn = existingColumnIndex.get(column.columnName());
ColumnInfo existingColumn = existingColumnIndex.get(columnKey(column.columnName()));
if (existingColumn == null) {
newColumns.add(columnInfo);
} else {
Expand Down Expand Up @@ -218,12 +221,13 @@ private SyncTableResult buildPresentTableResult(
Set<String> physicalColumnNames =
table.columns().stream()
.map(ColumnSyncSource::columnName)
.map(this::columnKey)
.collect(Collectors.toSet());
int addedColumns = 0;
int reactivatedColumns = 0;
int updatedColumns = 0;
for (ColumnSyncSource column : table.columns()) {
ColumnInfo existingColumn = existingColumnIndex.get(column.columnName());
ColumnInfo existingColumn = existingColumnIndex.get(columnKey(column.columnName()));
if (existingColumn == null) {
addedColumns++;
continue;
Expand All @@ -240,7 +244,7 @@ private SyncTableResult buildPresentTableResult(
int missingColumnsMarked =
collectMissingColumnIds(existingColumns, physicalColumnNames, missingColumnIds);
return SyncTableResult.builder()
.tableName(table.displayName())
.tableName(table.tableName())
.physicalTableFound(true)
.tableAdded(existingTable == null)
.tableReactivated(
Expand Down Expand Up @@ -277,6 +281,14 @@ private SyncTableResult buildMissingTableResult(
.build();
}

private String tableKey(String tableName) {
return SemanticUtils.normalizeObjectName(tableName, "Missing tableName.");
}

private String columnKey(String columnName) {
return SemanticUtils.normalizeObjectName(columnName, "Missing columnName.");
}

private Map<String, TableInfo> loadSemanticTableIndex(
Integer datasourceId, List<String> tableNames) {
Map<String, TableInfo> result = new LinkedHashMap<>();
Expand Down Expand Up @@ -362,10 +374,7 @@ private ColumnInfo buildPhysicalColumnInfo(
}

public record TableSyncSource(
String tableName,
String displayName,
String description,
List<ColumnSyncSource> columns) {}
String tableName, String description, List<ColumnSyncSource> columns) {}

public record ColumnSyncSource(
String columnName, String description, String typeName, Boolean primaryKey) {}
Expand Down
47 changes: 36 additions & 11 deletions ...end/src/main/java/io/github/malonetalk/service/semantic/sync/SemanticSyncServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import io.github.malonetalk.dto.semantic.SyncTableSemanticsRequest;
import io.github.malonetalk.dto.semantic.SyncTableSemanticsResponse;
import io.github.malonetalk.entity.Datasource;
import io.github.malonetalk.entity.TableInfo;
import io.github.malonetalk.exception.BusinessException;
import io.github.malonetalk.mapper.TableInfoMapper;
import io.github.malonetalk.service.DatasourceService;
Expand Down Expand Up @@ -70,7 +71,7 @@ public PageResponse<PhysicalTableCandidateResponse> getPhysicalTableCandidates(
}
// SchemaReader 一次性读取物理表，筛选、排序和分页统一在内存中完成。
List<PhysicalTableInfo> filteredTables =
schemaReader.getTables(datasource).stream()
loadCandidateTables(datasource).stream()
.filter(
table ->
keyword == null
Expand Down Expand Up @@ -99,26 +100,51 @@ public PageResponse<PhysicalTableCandidateResponse> getPhysicalTableCandidates(
return PageResponse.of(responses, filteredTables.size(), pageNumber, pageSize);
}

private List<PhysicalTableInfo> loadCandidateTables(Datasource datasource) {
Map<String, PhysicalTableInfo> candidates = new LinkedHashMap<>();
for (PhysicalTableInfo table : schemaReader.getTables(datasource)) {
candidates.putIfAbsent(
SemanticUtils.normalizeObjectName(
table.tableName(), "Missing physical tableName."),
table);
}
for (TableInfo table : tableInfoMapper.selectByDatasourceId(datasource.getId())) {
String tableName =
SemanticUtils.requireTrimmed(
table.getTableName(), "Missing semantic tableName.");
candidates.putIfAbsent(
SemanticUtils.normalizeObjectName(tableName, "Missing semantic tableName."),
new PhysicalTableInfo(
tableName,
SemanticUtils.trimToNull(table.getPhysicalTableDescription())));
}
return List.copyOf(candidates.values());
}

@Override
public SyncTableSemanticsResponse syncTables(SyncTableSemanticsRequest request) {
Datasource datasource = requireDatasource(request.datasourceId());
Map<String, PhysicalTableInfo> physicalTables = loadPhysicalTableIndex(datasource);
Set<String> selectedTableNames = new LinkedHashSet<>();
Map<String, String> selectedTableNames = new LinkedHashMap<>();
for (String tableName : request.tableNames()) {
selectedTableNames.add(
SemanticUtils.normalizeObjectName(tableName, "Missing physical tableName."));
String trimmedTableName =
SemanticUtils.requireTrimmed(tableName, "Missing physical tableName.");
selectedTableNames.putIfAbsent(
SemanticUtils.normalizeObjectName(
trimmedTableName, "Missing physical tableName."),
trimmedTableName);
}

List<TableSyncSource> presentTables = new ArrayList<>();
List<String> missingTableNames = new ArrayList<>();
// 先读取并整理全部物理结构，再交给事务方法批量写库，避免事务中访问外部数据源。
for (String normalizedTableName : selectedTableNames) {
PhysicalTableInfo physicalTable = physicalTables.get(normalizedTableName);
for (Map.Entry<String, String> selectedTable : selectedTableNames.entrySet()) {
PhysicalTableInfo physicalTable = physicalTables.get(selectedTable.getKey());
if (physicalTable == null) {
missingTableNames.add(normalizedTableName);
missingTableNames.add(selectedTable.getValue());
continue;
}
presentTables.add(readTableSyncSource(datasource, physicalTable, normalizedTableName));
presentTables.add(readTableSyncSource(datasource, physicalTable));
}

List<SyncTableResult> results =
Expand Down Expand Up @@ -169,21 +195,20 @@ private Set<String> loadSyncedTableNames(
}

private TableSyncSource readTableSyncSource(
Datasource datasource, PhysicalTableInfo physicalTable, String normalizedTableName) {
Datasource datasource, PhysicalTableInfo physicalTable) {
// 在开启写事务前读取物理列快照，避免 JDBC 元数据读取延长事务时间。
List<ColumnSyncSource> columns = new ArrayList<>();
for (PhysicalColumnInfo column :
schemaReader.getTableSchema(datasource, physicalTable.tableName())) {
columns.add(
new ColumnSyncSource(
SemanticUtils.normalizeObjectName(
SemanticUtils.requireTrimmed(
column.columnName(), "Missing physical columnName."),
SemanticUtils.trimToNull(column.remarks()),
SemanticUtils.trimToNull(column.typeName()),
column.primaryKey()));
}
return new TableSyncSource(
normalizedTableName,
physicalTable.tableName(),
SemanticUtils.trimToNull(physicalTable.remarks()),
columns);
Expand Down
2 changes: 2 additions & 0 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 @@ -146,6 +146,8 @@
<update id="updatePhysicalCacheFields" parameterType="io.github.malonetalk.entity.ColumnInfo">
UPDATE column_info
<set>
<if test="tableName != null">table_name = #{tableName},</if>
<if test="columnName != null">column_name = #{columnName},</if>
physical_column_description = #{physicalColumnDescription},
type_name = #{typeName},
primary_key = #{primaryKey},
Expand Down
1 change: 1 addition & 0 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 @@ -130,6 +130,7 @@
<update id="updatePhysicalCacheFields" parameterType="io.github.malonetalk.entity.TableInfo">
UPDATE table_info
<set>
<if test="tableName != null">table_name = #{tableName},</if>
physical_table_description = #{physicalTableDescription},
<if test="physicalStatus != null">physical_status = #{physicalStatus},</if>
<if test="updateTime != null">update_time = #{updateTime},</if>
Expand Down
Loading