Skip to content

Commit 2fd6fcf

Browse files
Chat2DB-ProChat2DB-Pro
committed
add
1 parent 23e4dcf commit 2fd6fcf

2 files changed

Lines changed: 89 additions & 39 deletions

File tree

‎chat2db-server/chat2db-server-domain/chat2db-server-domain-core/src/main/java/ai/chat2db/server/domain/core/cache/CacheKey.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -22,25 +22,25 @@ public static String getSchemasKey(Long dataSourceId, String databaseName) {
2222
}
2323

2424
public static String getTableKey(Long dataSourceId, String databaseName, String schemaName) {
25-
StringBuffer stringBuffer = new StringBuffer("tables_dataSourceId" + dataSourceId);
25+
StringBuffer stringBuffer = new StringBuffer("tables_dataSourceId_" + dataSourceId);
2626
if (!StringUtils.isEmpty(databaseName)) {
27-
stringBuffer.append("_databaseName" + databaseName);
27+
stringBuffer.append("_databaseName_" + databaseName);
2828
}
2929
if (!StringUtils.isEmpty(schemaName)) {
30-
stringBuffer.append("_schemaName" + schemaName);
30+
stringBuffer.append("_schemaName_" + schemaName);
3131
}
3232
return stringBuffer.toString();
3333
}
3434

3535
public static String getColumnKey(Long dataSourceId, String databaseName, String schemaName,String tableName) {
36-
StringBuffer stringBuffer = new StringBuffer("columns_dataSourceId" + dataSourceId);
36+
StringBuffer stringBuffer = new StringBuffer("columns_dataSourceId_" + dataSourceId);
3737
if (!StringUtils.isEmpty(databaseName)) {
38-
stringBuffer.append("_databaseName" + databaseName);
38+
stringBuffer.append("_databaseName_" + databaseName);
3939
}
4040
if (!StringUtils.isEmpty(schemaName)) {
41-
stringBuffer.append("_schemaName" + schemaName);
41+
stringBuffer.append("_schemaName_" + schemaName);
4242
}
43-
stringBuffer.append("_tableName"+tableName);
43+
stringBuffer.append("_tableName_"+tableName);
4444
return stringBuffer.toString();
4545
}
4646
}

‎chat2db-server/chat2db-server-domain/chat2db-server-domain-core/src/main/java/ai/chat2db/server/domain/core/impl/TableServiceImpl.java‎

Lines changed: 82 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -136,13 +136,45 @@ private void initOldTable(Table oldTable, Table newTable) {
136136
@Override
137137
public PageResult<Table> pageQuery(TablePageQueryParam param, TableSelector selector) {
138138
LambdaQueryWrapper<TableCacheVersionDO> queryWrapper = new LambdaQueryWrapper<>();
139-
queryWrapper.eq(TableCacheVersionDO::getKey, getTableKey(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName()));
139+
String key = getTableKey(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName());
140+
queryWrapper.eq(TableCacheVersionDO::getKey, key);
140141
TableCacheVersionDO versionDO = tableCacheVersionMapper.selectOne(queryWrapper);
142+
long total = 0;
143+
long version = 0L;
141144
if (param.isRefresh() || versionDO == null) {
142-
versionDO = addDBCache(param.getDataSourceId(),param.getDatabaseName(),param.getSchemaName(), versionDO);
145+
version = getLock(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName(), versionDO);
146+
if (version == -1) {
147+
int n = 0;
148+
while (n < 100) {
149+
try {
150+
Thread.sleep(200);
151+
} catch (InterruptedException e) {
152+
}
153+
versionDO = tableCacheVersionMapper.selectOne(queryWrapper);
154+
if (versionDO != null && "1".equals(versionDO.getStatus())) {
155+
version = versionDO.getVersion();
156+
total = versionDO.getTableCount();
157+
break;
158+
}
159+
n++;
160+
}
161+
} else {
162+
total = addDBCache(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName(), version);
163+
TableCacheVersionDO versionDO1 = new TableCacheVersionDO();
164+
versionDO1.setStatus("1");
165+
versionDO1.setTableCount(total);
166+
tableCacheVersionMapper.update(versionDO1, queryWrapper);
167+
}
168+
} else {
169+
if ("2".equals(versionDO.getStatus())) {
170+
version = versionDO.getVersion() - 1;
171+
} else {
172+
version = versionDO.getVersion();
173+
}
174+
total = versionDO.getTableCount();
143175
}
144176
LambdaQueryWrapper<TableCacheDO> query = new LambdaQueryWrapper<>();
145-
query.eq(TableCacheDO::getVersion, versionDO.getVersion());
177+
query.eq(TableCacheDO::getVersion, version);
146178
query.eq(TableCacheDO::getDataSourceId, param.getDataSourceId());
147179
if (StringUtils.isNotBlank(param.getDatabaseName())) {
148180
query.eq(TableCacheDO::getDatabaseName, param.getDatabaseName());
@@ -167,19 +199,22 @@ public PageResult<Table> pageQuery(TablePageQueryParam param, TableSelector sele
167199
tables.add(t);
168200
}
169201
}
170-
return PageResult.of(tables, versionDO.getTableCount(), param);
202+
return PageResult.of(tables, total, param);
171203
}
172204

173205
@Override
174206
public ListResult<SimpleTable> queryTables(TablePageQueryParam param) {
175207
LambdaQueryWrapper<TableCacheVersionDO> queryWrapper = new LambdaQueryWrapper<>();
176-
queryWrapper.eq(TableCacheVersionDO::getKey, getTableKey(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName()));
208+
String key = getTableKey(param.getDataSourceId(), param.getDatabaseName(), param.getSchemaName());
209+
queryWrapper.eq(TableCacheVersionDO::getKey, key);
177210
TableCacheVersionDO versionDO = tableCacheVersionMapper.selectOne(queryWrapper);
178-
if (param.isRefresh() || versionDO == null) {
179-
versionDO = addDBCache(param.getDataSourceId(),param.getDatabaseName(), param.getSchemaName(), versionDO);
211+
if(versionDO == null){
212+
return ListResult.of(Lists.newArrayList());
180213
}
214+
long version = "2".equals(versionDO.getStatus()) ? versionDO.getVersion() - 1 : versionDO.getVersion();
215+
181216
LambdaQueryWrapper<TableCacheDO> query = new LambdaQueryWrapper<>();
182-
query.eq(TableCacheDO::getVersion, versionDO.getVersion());
217+
query.eq(TableCacheDO::getVersion, version);
183218
query.eq(TableCacheDO::getDataSourceId, param.getDataSourceId());
184219
if (StringUtils.isNotBlank(param.getDatabaseName())) {
185220
query.eq(TableCacheDO::getDatabaseName, param.getDatabaseName());
@@ -189,8 +224,8 @@ public ListResult<SimpleTable> queryTables(TablePageQueryParam param) {
189224
}
190225
List<SimpleTable> tables = new ArrayList<>();
191226

192-
for (int i = 0; i < versionDO.getTableCount()/500 +1; i++) {
193-
Page<TableCacheDO> page = new Page<>(i+1, 500);
227+
for (int i = 0; i < versionDO.getTableCount() / 500 + 1; i++) {
228+
Page<TableCacheDO> page = new Page<>(i + 1, 500);
194229
IPage<TableCacheDO> iPage = tableCacheMapper.selectPage(page, query);
195230
if (CollectionUtils.isNotEmpty(iPage.getRecords())) {
196231
for (TableCacheDO tableCacheDO : iPage.getRecords()) {
@@ -204,23 +239,9 @@ public ListResult<SimpleTable> queryTables(TablePageQueryParam param) {
204239
return ListResult.of(tables);
205240
}
206241

207-
private TableCacheVersionDO addDBCache(Long dataSourceId,String databaseName,String schemaName, TableCacheVersionDO versionDO) {
242+
private long addDBCache(Long dataSourceId, String databaseName, String schemaName, long version) {
208243
String key = getTableKey(dataSourceId, databaseName, schemaName);
209-
if (versionDO == null) {
210-
versionDO = new TableCacheVersionDO();
211-
versionDO.setDatabaseName(databaseName);
212-
versionDO.setSchemaName(schemaName);
213-
versionDO.setDataSourceId(dataSourceId);
214-
versionDO.setStatus("2");
215-
versionDO.setKey(key);
216-
versionDO.setVersion(0L);
217-
versionDO.setTableCount(0L);
218-
tableCacheVersionMapper.insert(versionDO);
219-
} else {
220-
versionDO.setVersion(versionDO.getVersion() + 1);
221-
versionDO.setStatus("2");
222-
tableCacheVersionMapper.updateById(versionDO);
223-
}
244+
224245
Connection connection = Chat2DBContext.getConnection();
225246
long n = 0;
226247
try (ResultSet resultSet = connection.getMetaData().getTables(databaseName, schemaName, null,
@@ -233,7 +254,7 @@ private TableCacheVersionDO addDBCache(Long dataSourceId,String databaseName,Str
233254
tableCacheDO.setTableName(resultSet.getString("TABLE_NAME"));
234255
tableCacheDO.setExtendInfo(resultSet.getString("REMARKS"));
235256
tableCacheDO.setDataSourceId(dataSourceId);
236-
tableCacheDO.setVersion(versionDO.getVersion());
257+
tableCacheDO.setVersion(version);
237258
tableCacheDO.setKey(key);
238259
cacheDOS.add(tableCacheDO);
239260
if (cacheDOS.size() >= 500) {
@@ -245,12 +266,9 @@ private TableCacheVersionDO addDBCache(Long dataSourceId,String databaseName,Str
245266
if (!CollectionUtils.isEmpty(cacheDOS)) {
246267
tableCacheMapper.batchInsert(cacheDOS);
247268
}
248-
versionDO.setStatus("1");
249-
versionDO.setTableCount(n);
250-
tableCacheVersionMapper.updateById(versionDO);
251269
LambdaQueryWrapper<TableCacheDO> q = new LambdaQueryWrapper();
252270
q.eq(TableCacheDO::getDataSourceId, dataSourceId);
253-
q.lt(TableCacheDO::getVersion, versionDO.getVersion());
271+
q.lt(TableCacheDO::getVersion, version);
254272
if (StringUtils.isNotBlank(databaseName)) {
255273
q.eq(TableCacheDO::getDatabaseName, databaseName);
256274
}
@@ -261,8 +279,40 @@ private TableCacheVersionDO addDBCache(Long dataSourceId,String databaseName,Str
261279
} catch (SQLException e) {
262280
throw new RuntimeException(e);
263281
}
264-
return versionDO;
282+
return n;
283+
}
265284

285+
private Long getLock(Long dataSourceId, String databaseName, String schemaName, TableCacheVersionDO versionDO) {
286+
String key = getTableKey(dataSourceId, databaseName, schemaName);
287+
if (versionDO == null) {
288+
versionDO = new TableCacheVersionDO();
289+
versionDO.setDatabaseName(databaseName);
290+
versionDO.setSchemaName(schemaName);
291+
versionDO.setDataSourceId(dataSourceId);
292+
versionDO.setStatus("2");
293+
versionDO.setKey(key);
294+
versionDO.setVersion(0L);
295+
versionDO.setTableCount(0L);
296+
try {
297+
tableCacheVersionMapper.insert(versionDO);
298+
return 0L;
299+
} catch (Exception e) {
300+
return -1L;
301+
}
302+
} else {
303+
long version = versionDO.getVersion() + 1;
304+
LambdaQueryWrapper<TableCacheVersionDO> queryWrapper = new LambdaQueryWrapper();
305+
queryWrapper.eq(TableCacheVersionDO::getId, versionDO.getId());
306+
queryWrapper.eq(TableCacheVersionDO::getVersion, versionDO.getVersion());
307+
versionDO.setVersion(version);
308+
versionDO.setStatus("2");
309+
int n = tableCacheVersionMapper.update(versionDO, queryWrapper);
310+
if (n == 1) {
311+
return version;
312+
} else {
313+
return -1L;
314+
}
315+
}
266316
}
267317

268318

0 commit comments

Comments
 (0)