lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化DataJdbcSource

Browse Source
This commit is contained in:
Redkale
2022-12-29 14:56:24 +08:00
parent 8701450940
commit 2c861a5ed4
3 changed files with 291 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

317
src/main/java/org/redkale/source/DataJdbcSource.java
View File

@@ -202,66 +202,44 @@ public class DataJdbcSource extends DataSqlSource {
} }
@Override @Override
protected <T> CompletableFuture<Integer> deleteDBAsync(final EntityInfo<T> info, String[] tables, Flipper flipper, final String... sqls) { protected <T> CompletableFuture<Integer> deleteDBAsync(final EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, final String... sqls) {
return supplyAsync(() -> deleteDB(info, tables, flipper, sqls)); return supplyAsync(() -> deleteDB(info, tables, flipper, node, pkmap, sqls));
} }
@Override @Override
protected <T> int deleteDB(EntityInfo<T> info, String[] tables, Flipper flipper, String... sqls) { protected <T> int deleteDB(EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, String... sqls) {
Connection conn = null; Connection conn = null;
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setReadOnly(false); conn.setReadOnly(false);
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = 0; int c;
if (sqls.length == 1) { if (sqls.length == 1) {
String sql = sqls[0];
sql += ((flipper == null || flipper.getLimit() < 1) ? "" : (" LIMIT " + flipper.getLimit()));
if (info.isLoggable(logger, Level.FINEST, sql)) {
logger.finest(info.getType().getSimpleName() + " delete sql=" + sql);
}
final Statement stmt = conn.createStatement(); final Statement stmt = conn.createStatement();
c = stmt.executeUpdate(sql); int c1 = stmt.executeUpdate(sqls[0]);
stmt.close(); stmt.close();
c = c1;
} else { } else {
if (flipper == null || flipper.getLimit() < 1) { final Statement stmt = conn.createStatement();
if (info.isLoggable(logger, Level.FINEST, sqls[0])) { for (String sql : sqls) {
logger.finest(info.getType().getSimpleName() + " delete sqls=" + Arrays.toString(sqls)); stmt.addBatch(sql);
}
final Statement stmt = conn.createStatement();
for (String sql : sqls) {
stmt.addBatch(sql);
}
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c += cc;
}
} else {
if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " limit " + flipper.getLimit() + " delete sqls=" + Arrays.toString(sqls));
}
final Statement stmt = conn.createStatement();
for (String sql : sqls) {
stmt.addBatch(sql + " LIMIT " + flipper.getLimit());
}
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c += cc;
}
} }
int c1 = 0;
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c1 += cc;
}
c = c1;
} }
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
if (conn != null) { try {
try { conn.rollback();
conn.rollback(); } catch (SQLException se) {
} catch (SQLException se) {
}
} }
if (isTableNotExist(info, e.getSQLState())) { if (isTableNotExist(info, e.getSQLState())) {
if (info.getTableStrategy() == null) { if (info.getTableStrategy() == null) {
@@ -283,6 +261,57 @@ public class DataJdbcSource extends DataSqlSource {
throw new SourceException(e2); throw new SourceException(e2);
} }
} }
//单表结构不存在
return 0;
} else if (tables != null && tables.length == 1) {
//只查一个不存在的分表
return 0;
} else if (tables != null && tables.length > 1) {
//多分表查询中一个或多个分表不存在
String tableName = parseNotExistTableName(e);
if (tableName == null) {
throw new SourceException(e);
}
String[] oldTables = tables;
List<String> notExistTables = checkNotExistTablesNoThrows(conn, tables, tableName);
if (notExistTables.isEmpty()) {
throw new SourceException(e);
}
System.out.println(tableName + " notExistTables : " + notExistTables);
for (String t : notExistTables) {
if (pkmap != null) {
pkmap.remove(t);
} else {
tables = Utility.remove(tables, t);
}
}
if (logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "deleteDB, old-tables: " + Arrays.toString(oldTables) + ", new-tables: " + (pkmap != null ? pkmap.keySet() : Arrays.toString(tables)));
}
if ((pkmap != null ? pkmap.size() : tables.length) == 0) { //分表全部不存在
return 0;
}
sqls = pkmap != null ? deleteSql(info, pkmap) : deleteSql(info, tables, flipper, node);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
}
try {
final Statement stmt = conn.createStatement();
for (String sql : sqls) {
stmt.addBatch(sql);
}
int c = 0;
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c += cc;
}
conn.commit();
slowLog(s, sqls);
return c;
} catch (SQLException se) {
throw new SourceException(se);
}
} else { } else {
throw new SourceException(e); throw new SourceException(e);
} }
@@ -296,45 +325,96 @@ public class DataJdbcSource extends DataSqlSource {
} }
@Override @Override
protected <T> CompletableFuture<Integer> clearTableDBAsync(EntityInfo<T> info, final String[] tables, String... sqls) { protected <T> CompletableFuture<Integer> clearTableDBAsync(EntityInfo<T> info, final String[] tables, FilterNode node, String... sqls) {
return supplyAsync(() -> clearTableDB(info, tables, sqls)); return supplyAsync(() -> clearTableDB(info, tables, node, sqls));
} }
@Override @Override
protected <T> int clearTableDB(EntityInfo<T> info, final String[] tables, String... sqls) { protected <T> int clearTableDB(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
Connection conn = null; Connection conn = null;
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setReadOnly(false); conn.setReadOnly(false);
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = 0; int c;
final Statement stmt = conn.createStatement();
if (sqls.length == 1) { if (sqls.length == 1) {
String sql = sqls[0]; final Statement stmt = conn.createStatement();
c = stmt.executeUpdate(sql); int c1 = stmt.executeUpdate(sqls[0]);
stmt.close();
c = c1;
} else { } else {
final Statement stmt = conn.createStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0;
int[] cs = stmt.executeBatch(); int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) { for (int cc : cs) {
c += cc; c1 += cc;
} }
c = c1;
} }
stmt.close();
conn.commit(); conn.commit();
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
if (conn != null) { try {
try { conn.rollback();
conn.rollback(); } catch (SQLException se) {
} catch (SQLException se) {
}
} }
if (isTableNotExist(info, e.getSQLState())) { if (isTableNotExist(info, e.getSQLState())) {
return -1; if (info.getTableStrategy() == null) {
//单表结构不存在
return 0;
} else if (tables != null && tables.length == 1) {
//只查一个不存在的分表
return 0;
} else if (tables != null && tables.length > 1) {
//多分表查询中一个或多个分表不存在
String tableName = parseNotExistTableName(e);
if (tableName == null) {
throw new SourceException(e);
}
String[] oldTables = tables;
List<String> notExistTables = checkNotExistTablesNoThrows(conn, tables, tableName);
if (notExistTables.isEmpty()) {
throw new SourceException(e);
}
for (String t : notExistTables) {
tables = Utility.remove(tables, t);
}
if (logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "clearTableDB, old-tables: " + Arrays.toString(oldTables) + ", new-tables: " + Arrays.toString(tables));
}
if (tables.length == 0) { //分表全部不存在
return 0;
}
sqls = clearTableSql(info, tables, node);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls));
}
try {
final Statement stmt = conn.createStatement();
for (String sql : sqls) {
stmt.addBatch(sql);
}
int c = 0;
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c += cc;
}
conn.commit();
slowLog(s, sqls);
return c;
} catch (SQLException se) {
throw new SourceException(se);
}
} else {
throw new SourceException(e);
}
} }
throw new SourceException(e); throw new SourceException(e);
} finally { } finally {
@@ -345,51 +425,96 @@ public class DataJdbcSource extends DataSqlSource {
} }
@Override @Override
protected <T> CompletableFuture<Integer> dropTableDBAsync(EntityInfo<T> info, final String[] tables, String... sqls) { protected <T> CompletableFuture<Integer> dropTableDBAsync(EntityInfo<T> info, final String[] tables, FilterNode node, String... sqls) {
return supplyAsync(() -> dropTableDB(info, tables, sqls)); return supplyAsync(() -> dropTableDB(info, tables, node, sqls));
} }
@Override @Override
protected <T> int dropTableDB(EntityInfo<T> info, String[] tables, String... sqls) { protected <T> int dropTableDB(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
Connection conn = null; Connection conn = null;
final long s = System.currentTimeMillis(); final long s = System.currentTimeMillis();
try { try {
conn = writePool.pollConnection(); conn = writePool.pollConnection();
conn.setReadOnly(false); conn.setReadOnly(false);
conn.setAutoCommit(false); conn.setAutoCommit(false);
int c = 0; int c;
final Statement stmt = conn.createStatement();
if (sqls.length == 1) { if (sqls.length == 1) {
String sql = sqls[0]; final Statement stmt = conn.createStatement();
c = stmt.executeUpdate(sql); int c1 = stmt.executeUpdate(sqls[0]);
stmt.close();
c = c1;
} else { } else {
final Statement stmt = conn.createStatement();
for (String sql : sqls) { for (String sql : sqls) {
stmt.addBatch(sql); stmt.addBatch(sql);
} }
int c1 = 0;
int[] cs = stmt.executeBatch(); int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) { for (int cc : cs) {
c += cc; c1 += cc;
} }
c = c1;
} }
stmt.close();
conn.commit(); conn.commit();
if (info.getTableStrategy() != null) {
for (String table : tables) {
String tablekey = table.indexOf('.') > 0 ? table : (conn.getCatalog() + '.' + table);
info.removeDisTable(tablekey);
}
}
slowLog(s, sqls); slowLog(s, sqls);
return c; return c;
} catch (SQLException e) { } catch (SQLException e) {
if (conn != null) { try {
try { conn.rollback();
conn.rollback(); } catch (SQLException se) {
} catch (SQLException se) {
}
} }
if (isTableNotExist(info, e.getSQLState())) { if (isTableNotExist(info, e.getSQLState())) {
return -1; if (info.getTableStrategy() == null) {
//单表结构不存在
return 0;
} else if (tables != null && tables.length == 1) {
//只查一个不存在的分表
return 0;
} else if (tables != null && tables.length > 1) {
//多分表查询中一个或多个分表不存在
String tableName = parseNotExistTableName(e);
if (tableName == null) {
throw new SourceException(e);
}
String[] oldTables = tables;
List<String> notExistTables = checkNotExistTablesNoThrows(conn, tables, tableName);
if (notExistTables.isEmpty()) {
throw new SourceException(e);
}
for (String t : notExistTables) {
tables = Utility.remove(tables, t);
}
if (logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE, "dropTableDB, old-tables: " + Arrays.toString(oldTables) + ", new-tables: " + Arrays.toString(tables));
}
if (tables.length == 0) { //分表全部不存在
return 0;
}
sqls = dropTableSql(info, tables, node);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls));
}
try {
final Statement stmt = conn.createStatement();
for (String sql : sqls) {
stmt.addBatch(sql);
}
int c = 0;
int[] cs = stmt.executeBatch();
stmt.close();
for (int cc : cs) {
c += cc;
}
conn.commit();
slowLog(s, sqls);
return c;
} catch (SQLException se) {
throw new SourceException(se);
}
} else {
throw new SourceException(e);
}
} }
throw new SourceException(e); throw new SourceException(e);
} finally { } finally {
@@ -1565,25 +1690,35 @@ public class DataJdbcSource extends DataSqlSource {
return new String[]{listSql, countSql}; return new String[]{listSql, countSql};
} }
protected List<String> checkNotExistTablesNoThrows(Connection conn, String[] tables, String firstNotExistTable) {
try {
return checkNotExistTables(conn, tables, firstNotExistTable);
} catch (SQLException e) {
throw new SourceException(e);
}
}
protected List<String> checkNotExistTables(Connection conn, String[] tables, String firstNotExistTable) throws SQLException { protected List<String> checkNotExistTables(Connection conn, String[] tables, String firstNotExistTable) throws SQLException {
String minTableName = (firstNotExistTable.indexOf('.') > 0) ? firstNotExistTable.substring(firstNotExistTable.indexOf('.') + 1) : null; // 数据库不一定要按批量提交的SQL顺序执行， 所以第一个不存在的表不一定再tables的第一位,
List<String> maybeNoTables = new ArrayList<>(); // 比如 DELETE FROM table1; DELETE FROM table2; 如果table1、table2都不存在，SQL可能会抛出table2不存在的异常
for (String t : tables) { // List<String> maybeNoTables = new ArrayList<>();
if (!maybeNoTables.isEmpty()) { // String minTableName = (firstNotExistTable.indexOf('.') > 0) ? firstNotExistTable.substring(firstNotExistTable.indexOf('.') + 1) : null;
maybeNoTables.add(t); // for (String t : tables) {
} // if (!maybeNoTables.isEmpty()) {
if (t.equals(firstNotExistTable) || (minTableName != null && t.equals(minTableName))) { // maybeNoTables.add(t);
maybeNoTables.add(t); // }
} // if (t.equals(firstNotExistTable) || (minTableName != null && t.equals(minTableName))) {
} // maybeNoTables.add(t);
if (maybeNoTables.isEmpty()) { // }
return maybeNoTables; // }
} // if (maybeNoTables.isEmpty()) {
// return maybeNoTables;
// }
String[] tableTypes = new String[]{"TABLE"}; String[] tableTypes = new String[]{"TABLE"};
DatabaseMetaData dmd = conn.getMetaData(); DatabaseMetaData dmd = conn.getMetaData();
List<String> rs = new ArrayList<>(); List<String> rs = new ArrayList<>();
for (String t : maybeNoTables) { for (String t : tables) { //maybeNoTables
String catalog = null; String catalog = null;
String table = t; String table = t;
int pos = t.indexOf('.'); int pos = t.indexOf('.');

12
src/main/java/org/redkale/source/DataMemorySource.java
View File

@@ -122,17 +122,17 @@ public class DataMemorySource extends DataSqlSource implements SearchSource {
} }
@Override @Override
protected <T> int deleteDB(EntityInfo<T> info, String[] tables, Flipper flipper, String... sqls) { protected <T> int deleteDB(EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, String... sqls) {
return 0; return 0;
} }
@Override @Override
protected <T> int clearTableDB(EntityInfo<T> info, String[] tables, String... sqls) { protected <T> int clearTableDB(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
return 0; return 0;
} }
@Override @Override
protected <T> int dropTableDB(EntityInfo<T> info, String[] tables, String... sqls) { protected <T> int dropTableDB(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
return 0; return 0;
} }
@@ -192,17 +192,17 @@ public class DataMemorySource extends DataSqlSource implements SearchSource {
} }
@Override @Override
protected <T> CompletableFuture<Integer> deleteDBAsync(EntityInfo<T> info, String[] tables, Flipper flipper, String... sqls) { protected <T> CompletableFuture<Integer> deleteDBAsync(EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, String... sqls) {
return CompletableFuture.completedFuture(0); return CompletableFuture.completedFuture(0);
} }
@Override @Override
protected <T> CompletableFuture<Integer> clearTableDBAsync(EntityInfo<T> info, String[] tables, String... sqls) { protected <T> CompletableFuture<Integer> clearTableDBAsync(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
return CompletableFuture.completedFuture(0); return CompletableFuture.completedFuture(0);
} }
@Override @Override
protected <T> CompletableFuture<Integer> dropTableDBAsync(EntityInfo<T> info, String[] tables, String... sqls) { protected <T> CompletableFuture<Integer> dropTableDBAsync(EntityInfo<T> info, String[] tables, FilterNode node, String... sqls) {
return CompletableFuture.completedFuture(0); return CompletableFuture.completedFuture(0);
} }

114
src/main/java/org/redkale/source/DataSqlSource.java
View File

@@ -667,13 +667,13 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
protected abstract <T> CompletableFuture<Integer> insertDBAsync(final EntityInfo<T> info, T... entitys); protected abstract <T> CompletableFuture<Integer> insertDBAsync(final EntityInfo<T> info, T... entitys);
//删除记录 //删除记录
protected abstract <T> CompletableFuture<Integer> deleteDBAsync(final EntityInfo<T> info, String[] tables, Flipper flipper, final String... sqls); protected abstract <T> CompletableFuture<Integer> deleteDBAsync(final EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, final String... sqls);
//清空表 //清空表
protected abstract <T> CompletableFuture<Integer> clearTableDBAsync(final EntityInfo<T> info, String[] tables, final String... sqls); protected abstract <T> CompletableFuture<Integer> clearTableDBAsync(final EntityInfo<T> info, String[] tables, FilterNode node, final String... sqls);
//删除表 //删除表
protected abstract <T> CompletableFuture<Integer> dropTableDBAsync(final EntityInfo<T> info, String[] tables, final String... sqls); protected abstract <T> CompletableFuture<Integer> dropTableDBAsync(final EntityInfo<T> info, String[] tables, FilterNode node, final String... sqls);
//更新纪录 //更新纪录
protected abstract <T> CompletableFuture<Integer> updateEntityDBAsync(final EntityInfo<T> info, T... entitys); protected abstract <T> CompletableFuture<Integer> updateEntityDBAsync(final EntityInfo<T> info, T... entitys);
@@ -711,18 +711,18 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
//删除记录 //删除记录
protected <T> int deleteDB(final EntityInfo<T> info, String[] tables, Flipper flipper, final String... sqls) { protected <T> int deleteDB(final EntityInfo<T> info, String[] tables, Flipper flipper, FilterNode node, Map<String, List<Serializable>> pkmap, final String... sqls) {
return deleteDBAsync(info, tables, flipper, sqls).join(); return deleteDBAsync(info, tables, flipper, node, pkmap, sqls).join();
} }
//清空表 //清空表
protected <T> int clearTableDB(final EntityInfo<T> info, String[] tables, final String... sqls) { protected <T> int clearTableDB(final EntityInfo<T> info, String[] tables, FilterNode node, final String... sqls) {
return clearTableDBAsync(info, tables, sqls).join(); return clearTableDBAsync(info, tables, node, sqls).join();
} }
//删除表 //删除表
protected <T> int dropTableDB(final EntityInfo<T> info, String[] tables, final String... sqls) { protected <T> int dropTableDB(final EntityInfo<T> info, String[] tables, FilterNode node, final String... sqls) {
return dropTableDBAsync(info, tables, sqls).join(); return dropTableDBAsync(info, tables, node, sqls).join();
} }
//更新纪录 //更新纪录
@@ -974,17 +974,18 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
if (isOnlyCache(info)) { if (isOnlyCache(info)) {
return deleteCache(info, -1, pks); return deleteCache(info, -1, pks);
} }
String[] tables = new String[]{info.getTable(pks[0])}; Map<String, List<Serializable>> pkmap = info.getTableMap(pks);
String[] sqls = deleteSql(info, pks); String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
String[] sqls = deleteSql(info, pkmap);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) { if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
int rs = deleteDBAsync(info, tables, null, sqls).join(); int rs = deleteDBAsync(info, tables, null, null, pkmap, sqls).join();
deleteCache(info, rs, pks); deleteCache(info, rs, pks);
return rs; return rs;
} else { } else {
int rs = deleteDB(info, tables, null, sqls); int rs = deleteDB(info, tables, null, null, pkmap, sqls);
deleteCache(info, rs, pks); deleteCache(info, rs, pks);
return rs; return rs;
} }
@@ -999,13 +1000,14 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
if (isOnlyCache(info)) { if (isOnlyCache(info)) {
return CompletableFuture.completedFuture(deleteCache(info, -1, pks)); return CompletableFuture.completedFuture(deleteCache(info, -1, pks));
} }
String[] tables = new String[]{info.getTable(pks[0])}; Map<String, List<Serializable>> pkmap = info.getTableMap(pks);
String[] sqls = deleteSql(info, pks); String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
String[] sqls = deleteSql(info, pkmap);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) { if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
return deleteDBAsync(info, tables, null, sqls).whenComplete((rs, t) -> { return deleteDBAsync(info, tables, null, null, pkmap, sqls).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1013,7 +1015,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
}); });
} else { } else {
return supplyAsync(() -> deleteDB(info, tables, null, sqls)).whenComplete((rs, t) -> { return supplyAsync(() -> deleteDB(info, tables, null, null, pkmap, sqls)).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1045,11 +1047,11 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
int rs = deleteDBAsync(info, tables, flipper, sqls).join(); int rs = deleteDBAsync(info, tables, flipper, node, null, sqls).join();
deleteCache(info, rs, flipper, sqls); deleteCache(info, rs, flipper, sqls);
return rs; return rs;
} else { } else {
int rs = deleteDB(info, tables, flipper, sqls); int rs = deleteDB(info, tables, flipper, node, null, sqls);
deleteCache(info, rs, flipper, sqls); deleteCache(info, rs, flipper, sqls);
return rs; return rs;
} }
@@ -1067,7 +1069,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
return deleteDBAsync(info, tables, flipper, sqls).whenComplete((rs, t) -> { return deleteDBAsync(info, tables, flipper, node, null, sqls).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1075,7 +1077,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
}); });
} else { } else {
return supplyAsync(() -> deleteDB(info, tables, flipper, sqls)).whenComplete((rs, t) -> { return supplyAsync(() -> deleteDB(info, tables, flipper, node, null, sqls)).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1085,25 +1087,29 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
} }
protected <T> String[] deleteSql(final EntityInfo<T> info, final Serializable... pks) { protected <T> String[] deleteSql(final EntityInfo<T> info, final Map<String, List<Serializable>> pkmap) {
if (pks.length == 1) { List<String> sqls = new ArrayList<>();
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pks[0], sqlFormatter); final String pkSQLColumn = info.getPrimarySQLColumn();
return new String[]{sql}; pkmap.forEach((table, pks) -> {
} else { String sql;
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + " IN ("; if (pks.size() == 1) {
for (int i = 0; i < pks.length; i++) { sql = "DELETE FROM " + table + " WHERE " + pkSQLColumn + " = " + info.formatSQLValue(pkSQLColumn, pks.get(0), sqlFormatter);
if (i > 0) { } else {
sql += ','; sql = "DELETE FROM " + table + " WHERE " + pkSQLColumn + " IN (";
for (int i = 0; i < pks.size(); i++) {
if (i > 0) {
sql += ',';
}
sql += info.formatSQLValue(pkSQLColumn, pks.get(i), sqlFormatter);
} }
sql += info.formatSQLValue(info.getPrimarySQLColumn(), pks[i], sqlFormatter); sql += ")";
} }
sql += ")"; sqls.add(sql);
return new String[]{sql}; });
} return sqls.toArray(new String[sqls.size()]);
} }
protected <T> String[] deleteSql(final EntityInfo<T> info, String[] tables, final Flipper flipper, final FilterNode node) { protected <T> String[] deleteSql(final EntityInfo<T> info, String[] tables, final Flipper flipper, final FilterNode node) {
boolean pgsql = "postgresql".equals(dbtype());
Map<Class, String> joinTabalis = null; Map<Class, String> joinTabalis = null;
CharSequence join = null; CharSequence join = null;
CharSequence where = null; CharSequence where = null;
@@ -1119,25 +1125,23 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
join1 = multiSplit('[', ']', ",", new StringBuilder(), joinstr, 0); join1 = multiSplit('[', ']', ",", new StringBuilder(), joinstr, 0);
join2 = multiSplit('{', '}', " AND ", new StringBuilder(), joinstr, 0); join2 = multiSplit('{', '}', " AND ", new StringBuilder(), joinstr, 0);
} }
final String join2AndWhere = ((where == null || where.length() == 0) ? (join2 == null ? ""
: (" WHERE " + join2)) : (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
if (pgsql && flipper != null && flipper.getLimit() > 0) { if ("postgresql".equals(dbtype()) && flipper != null && flipper.getLimit() > 0) {
String wherestr = ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
List<String> sqls = new ArrayList<>(); List<String> sqls = new ArrayList<>();
for (String table : tables) { for (String table : tables) {
String sql = "DELETE FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) String sql = "DELETE FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) + " WHERE " + info.getPrimarySQLColumn() + " IN (SELECT " + info.getPrimaryColumn() + " FROM " + table
+ " WHERE " + info.getPrimarySQLColumn() + " IN (SELECT " + info.getPrimaryColumn() + " FROM " + table + join2AndWhere + info.createSQLOrderby(flipper) + " OFFSET 0 LIMIT " + flipper.getLimit() + ")";
+ wherestr + info.createSQLOrderby(flipper) + " OFFSET 0 LIMIT " + flipper.getLimit() + ")";
sqls.add(sql); sqls.add(sql);
} }
return sqls.toArray(new String[sqls.size()]); return sqls.toArray(new String[sqls.size()]);
} else { } else {
boolean mysql = "mysql".equals(dbtype());
List<String> sqls = new ArrayList<>(); List<String> sqls = new ArrayList<>();
for (String table : tables) { for (String table : tables) {
String sql = "DELETE " + ("mysql".equals(dbtype()) ? "a" : "") + " FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) String sql = "DELETE " + (mysql ? "a" : "") + " FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) + join2AndWhere + info.createSQLOrderby(flipper)
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2)) + ((mysql && flipper != null && flipper.getLimit() > 0) ? (" LIMIT " + flipper.getLimit()) : "");
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2)))) + info.createSQLOrderby(flipper)
+ (("mysql".equals(dbtype()) && flipper != null && flipper.getLimit() > 0) ? (" LIMIT " + flipper.getLimit()) : "");
sqls.add(sql); sqls.add(sql);
} }
return sqls.toArray(new String[sqls.size()]); return sqls.toArray(new String[sqls.size()]);
@@ -1162,11 +1166,11 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
int rs = clearTableDBAsync(info, tables, sqls).join(); int rs = clearTableDBAsync(info, tables, node, sqls).join();
clearTableCache(info, node); clearTableCache(info, node);
return rs; return rs;
} else { } else {
int rs = clearTableDB(info, tables, sqls); int rs = clearTableDB(info, tables, node, sqls);
clearTableCache(info, node); clearTableCache(info, node);
return rs; return rs;
} }
@@ -1184,7 +1188,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
return clearTableDBAsync(info, tables, sqls).whenComplete((rs, t) -> { return clearTableDBAsync(info, tables, node, sqls).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1192,7 +1196,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
}); });
} else { } else {
return supplyAsync(() -> clearTableDB(info, tables, sqls)).whenComplete((rs, t) -> { return supplyAsync(() -> clearTableDB(info, tables, node, sqls)).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1218,16 +1222,16 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
return dropTableCache(info, node); return dropTableCache(info, node);
} }
final String[] tables = info.getTables(node); final String[] tables = info.getTables(node);
String[] sqls = clearTableSql(info, tables, node); String[] sqls = dropTableSql(info, tables, node);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) { if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
int rs = dropTableDBAsync(info, tables, sqls).join(); int rs = dropTableDBAsync(info, tables, node, sqls).join();
dropTableCache(info, node); dropTableCache(info, node);
return rs; return rs;
} else { } else {
int rs = dropTableDB(info, tables, sqls); int rs = dropTableDB(info, tables, node, sqls);
dropTableCache(info, node); dropTableCache(info, node);
return rs; return rs;
} }
@@ -1240,12 +1244,12 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
return CompletableFuture.completedFuture(dropTableCache(info, node)); return CompletableFuture.completedFuture(dropTableCache(info, node));
} }
final String[] tables = info.getTables(node); final String[] tables = info.getTables(node);
String[] sqls = clearTableSql(info, tables, node); String[] sqls = dropTableSql(info, tables, node);
if (info.isLoggable(logger, Level.FINEST, sqls[0])) { if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls)); logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls));
} }
if (isAsync()) { if (isAsync()) {
return dropTableDBAsync(info, tables, sqls).whenComplete((rs, t) -> { return dropTableDBAsync(info, tables, node, sqls).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {
@@ -1253,7 +1257,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
} }
}); });
} else { } else {
return supplyAsync(() -> dropTableDB(info, tables, sqls)).whenComplete((rs, t) -> { return supplyAsync(() -> dropTableDB(info, tables, node, sqls)).whenComplete((rs, t) -> {
if (t != null) { if (t != null) {
errorCompleteConsumer.accept(rs, t); errorCompleteConsumer.accept(rs, t);
} else { } else {