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

优化DataBatch

Browse Source
This commit is contained in:
Redkale
2022-12-15 13:38:27 +08:00
parent a2331ad71a
commit c805beeb34
5 changed files with 352 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

325
src/main/java/org/redkale/source/AbstractDataSource.java
View File

@@ -314,6 +314,134 @@ public abstract class AbstractDataSource extends AbstractService implements Data
return null;
}
protected DefaultBatchInfo parseBatchInfo(DefaultDataBatch batch, Function<Class, EntityInfo> func) {
final List<BatchAction> actions = ((DefaultDataBatch) batch).actions;
final Map<Class, EntityInfo> infos = new HashMap<>();
final Map<String, EntityInfo> notInsertDisTables = new HashMap<>();
final Map<String, EntityInfo> notUpDelDisTables = new HashMap<>();
for (BatchAction action : actions) {
if (action instanceof InsertBatchAction1) {
InsertBatchAction1 act = (InsertBatchAction1) action;
Object entity = act.entity;
Class clazz = entity.getClass();
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(entity);
if (!info.containsDisTable(tableKey)) {
notInsertDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof DeleteBatchAction1) {
DeleteBatchAction1 act = (DeleteBatchAction1) action;
Object entity = act.entity;
Class clazz = entity.getClass();
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(entity);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof DeleteBatchAction2) {
DeleteBatchAction2 act = (DeleteBatchAction2) action;
Class clazz = act.clazz;
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(act.pk);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof DeleteBatchAction3) {
DeleteBatchAction3 act = (DeleteBatchAction3) action;
Class clazz = act.clazz;
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(act.node);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof UpdateBatchAction1) {
UpdateBatchAction1 act = (UpdateBatchAction1) action;
Object entity = act.entity;
Class clazz = entity.getClass();
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(entity);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof UpdateBatchAction2) {
UpdateBatchAction2 act = (UpdateBatchAction2) action;
Class clazz = act.clazz;
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(act.pk);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof UpdateBatchAction3) {
UpdateBatchAction3 act = (UpdateBatchAction3) action;
Class clazz = act.clazz;
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = info.getTable(act.node);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
} else if (action instanceof UpdateBatchAction4) {
UpdateBatchAction4 act = (UpdateBatchAction4) action;
Class clazz = act.entity.getClass();
if (!infos.containsKey(clazz)) {
EntityInfo info = func.apply(clazz);
infos.put(clazz, info);
if (info.getTableStrategy() != null) {
String tableKey = act.node == null ? info.getTable(act.entity) : info.getTable(act.node);
if (!info.containsDisTable(tableKey)) {
notUpDelDisTables.put(tableKey, info);
}
}
}
}
}
return new DefaultBatchInfo(infos, notInsertDisTables, notUpDelDisTables);
}
@Override
public int batch(final DataBatch batch) {
return batchAsync(batch).join();
}
@Override
public CompletableFuture<Integer> batchAsync(final DataBatch batch) {
throw new UnsupportedOperationException("Not supported yet.");
}
@Override
public final <T> int insert(final Collection<T> entitys) {
if (entitys == null || entitys.isEmpty()) return 0;
@@ -1134,65 +1262,32 @@ public abstract class AbstractDataSource extends AbstractService implements Data
protected static class DefaultDataBatch implements DataBatch {
//-------------------- 新增操作 --------------------
@Comment("新增Entity对象")
public Map<Class, List> insertEntitys;
//-------------------- 删除操作 --------------------
@Comment("删除Entity对象")
public Map<Class, List> deleteEntitys;
@Comment("根据主键值删除")
public Map<Class, List<DeleteBatchAction1>> deleteActions1;
@Comment("根据FilterNode删除")
public Map<Class, List<DeleteBatchAction2>> deleteActions2;
//-------------------- 修改操作 --------------------
@Comment("修改Entity对象")
public Map<Class, List> updateEntitys;
@Comment("根据主键值修改部分字段")
public Map<Class, List<UpdateBatchAction1>> updateActions1;
@Comment("根据FilterNode修改部分字段")
public Map<Class, List<UpdateBatchAction2>> updateActions2;
@Comment("根据FilterNode修改Entity部分字段")
public Map<Class, List<UpdateBatchAction3>> updateActions3;
@Comment("根据FilterNode修改Entity的SelectColumn选定字段")
public Map<Class, List<UpdateBatchAction4>> updateActions4;
@Comment("操作对象")
public final List<BatchAction> actions = new ArrayList();
protected DefaultDataBatch() {
}
@Override
@Override //entitys不一定是同一表的数据
public <T> DataBatch insert(T... entitys) {
if (this.insertEntitys == null) {
this.insertEntitys = new HashMap<>();
}
for (T t : entitys) {
Objects.requireNonNull(t);
if (t.getClass().getAnnotation(Entity.class) == null) {
throw new RuntimeException("Entity Class " + t.getClass() + " must be on Annotation @Entity");
}
this.insertEntitys.computeIfAbsent(t.getClass(), c -> new ArrayList<>()).add(t);
this.actions.add(new InsertBatchAction1(t));
}
return this;
}
@Override
@Override //entitys不一定是同一表的数据
public <T> DataBatch delete(T... entitys) {
if (this.deleteEntitys == null) {
this.deleteEntitys = new HashMap<>();
}
for (T t : entitys) {
Objects.requireNonNull(t);
if (t.getClass().getAnnotation(Entity.class) == null) {
throw new RuntimeException("Entity Class " + t.getClass() + " must be on Annotation @Entity");
}
this.deleteEntitys.computeIfAbsent(t.getClass(), c -> new ArrayList<>()).add(t);
this.actions.add(new DeleteBatchAction1(t));
}
return this;
}
@@ -1208,11 +1303,8 @@ public abstract class AbstractDataSource extends AbstractService implements Data
}
for (Serializable pk : pks) {
Objects.requireNonNull(pk);
this.actions.add(new DeleteBatchAction2(clazz, pk));
}
if (this.deleteActions1 == null) {
this.deleteActions1 = new HashMap<>();
}
this.deleteActions1.computeIfAbsent(clazz, c -> new ArrayList<>()).add(new DeleteBatchAction1(clazz, pks));
return this;
}
@@ -1227,24 +1319,18 @@ public abstract class AbstractDataSource extends AbstractService implements Data
if (clazz.getAnnotation(Entity.class) == null) {
throw new RuntimeException("Entity Class " + clazz + " must be on Annotation @Entity");
}
if (this.deleteActions2 == null) {
this.deleteActions2 = new HashMap<>();
}
this.deleteActions2.computeIfAbsent(clazz, c -> new ArrayList<>()).add(new DeleteBatchAction2(clazz, node, flipper));
this.actions.add(new DeleteBatchAction3(clazz, node, flipper));
return this;
}
@Override
@Override //entitys不一定是同一表的数据
public <T> DataBatch update(T... entitys) {
if (this.updateEntitys == null) {
this.updateEntitys = new HashMap<>();
}
for (T t : entitys) {
Objects.requireNonNull(t);
if (t.getClass().getAnnotation(Entity.class) == null) {
throw new RuntimeException("Entity Class " + t.getClass() + " must be on Annotation @Entity");
}
this.updateEntitys.computeIfAbsent(t.getClass(), c -> new ArrayList<>()).add(t);
this.actions.add(new UpdateBatchAction1(t));
}
return this;
}
@@ -1267,10 +1353,7 @@ public abstract class AbstractDataSource extends AbstractService implements Data
for (ColumnValue val : values) {
Objects.requireNonNull(val);
}
if (this.updateActions1 == null) {
this.updateActions1 = new HashMap<>();
}
this.updateActions1.computeIfAbsent(clazz, c -> new ArrayList<>()).add(new UpdateBatchAction1(clazz, pk, values));
this.actions.add(new UpdateBatchAction2(clazz, pk, values));
return this;
}
@@ -1296,35 +1379,24 @@ public abstract class AbstractDataSource extends AbstractService implements Data
for (ColumnValue val : values) {
Objects.requireNonNull(val);
}
if (this.updateActions2 == null) {
this.updateActions2 = new HashMap<>();
}
this.updateActions2.computeIfAbsent(clazz, c -> new ArrayList<>()).add(new UpdateBatchAction2(clazz, node, flipper, values));
this.actions.add(new UpdateBatchAction3(clazz, node, flipper, values));
return this;
}
@Override
public <T> DataBatch updateColumn(T entity, final String... columns) {
return updateColumn(entity, (FilterNode) null, columns);
}
@Override
public <T> DataBatch updateColumn(T entity, final FilterNode node, final String... columns) {
Objects.requireNonNull(entity);
if (entity.getClass().getAnnotation(Entity.class) == null) {
throw new RuntimeException("Entity Class " + entity.getClass() + " must be on Annotation @Entity");
}
if (columns.length < 1) {
throw new RuntimeException("update column length is zero ");
}
for (String val : columns) {
Objects.requireNonNull(val);
}
if (this.updateActions3 == null) {
this.updateActions3 = new HashMap<>();
}
this.updateActions3.computeIfAbsent(entity.getClass(), c -> new ArrayList<>()).add(new UpdateBatchAction3(entity, node, columns));
return this;
return updateColumn(entity, (FilterNode) null, SelectColumn.includes(columns));
}
@Override
public <T> DataBatch updateColumn(T entity, final FilterNode node, final String... columns) {
return updateColumn(entity, node, SelectColumn.includes(columns));
}
@Override
@@ -1339,28 +1411,66 @@ public abstract class AbstractDataSource extends AbstractService implements Data
throw new RuntimeException("Entity Class " + entity.getClass() + " must be on Annotation @Entity");
}
Objects.requireNonNull(selects);
if (this.updateActions4 == null) {
this.updateActions4 = new HashMap<>();
}
this.updateActions4.computeIfAbsent(entity.getClass(), c -> new ArrayList<>()).add(new UpdateBatchAction4(entity, node, selects));
this.actions.add(new UpdateBatchAction4(entity, node, selects));
return this;
}
}
protected static class DeleteBatchAction1 {
protected static class DefaultBatchInfo {
public Class clazz;
//EntityInfo对象
public Map<Class, EntityInfo> entityInfos;
public Serializable[] pks;
//新增操作可能不存在的分表
public Map<String, EntityInfo> notInsertDisTables;
public DeleteBatchAction1(Class clazz, Serializable... pks) {
this.clazz = clazz;
this.pks = pks;
//删除修改操作可能不存在的分表
public Map<String, EntityInfo> notUpDelDisTables;
public DefaultBatchInfo(Map<Class, EntityInfo> entityInfos, Map<String, EntityInfo> notInsertDisTables, Map<String, EntityInfo> notUpDelDisTables) {
this.entityInfos = entityInfos;
this.notInsertDisTables = notInsertDisTables;
this.notUpDelDisTables = notUpDelDisTables;
}
}
protected abstract static class BatchAction {
}
protected static class InsertBatchAction1 extends BatchAction {
public Object entity;
public InsertBatchAction1(Object entity) {
this.entity = entity;
}
}
protected static class DeleteBatchAction2 {
protected static class DeleteBatchAction1 extends BatchAction {
public Object entity;
public DeleteBatchAction1(Object entity) {
this.entity = entity;
}
}
protected static class DeleteBatchAction2 extends BatchAction {
public Class clazz;
public Serializable pk;
public DeleteBatchAction2(Class clazz, Serializable pk) {
this.clazz = clazz;
this.pk = pk;
}
}
protected static class DeleteBatchAction3 extends BatchAction {
public Class clazz;
@@ -1368,19 +1478,28 @@ public abstract class AbstractDataSource extends AbstractService implements Data
public Flipper flipper;
public DeleteBatchAction2(Class clazz, FilterNode node) {
public DeleteBatchAction3(Class clazz, FilterNode node) {
this.clazz = clazz;
this.node = node;
}
public DeleteBatchAction2(Class clazz, FilterNode node, Flipper flipper) {
public DeleteBatchAction3(Class clazz, FilterNode node, Flipper flipper) {
this.clazz = clazz;
this.node = node;
this.flipper = flipper;
}
}
protected static class UpdateBatchAction1 {
protected static class UpdateBatchAction1 extends BatchAction {
public Object entity;
public UpdateBatchAction1(Object entity) {
this.entity = entity;
}
}
protected static class UpdateBatchAction2 extends BatchAction {
public Class clazz;
@@ -1388,14 +1507,14 @@ public abstract class AbstractDataSource extends AbstractService implements Data
public ColumnValue[] values;
public UpdateBatchAction1(Class clazz, Serializable pk, ColumnValue... values) {
public UpdateBatchAction2(Class clazz, Serializable pk, ColumnValue... values) {
this.clazz = clazz;
this.pk = pk;
this.values = values;
}
}
protected static class UpdateBatchAction2 {
protected static class UpdateBatchAction3 extends BatchAction {
public Class clazz;
@@ -1405,13 +1524,13 @@ public abstract class AbstractDataSource extends AbstractService implements Data
public ColumnValue[] values;
public UpdateBatchAction2(Class clazz, FilterNode node, ColumnValue... values) {
public UpdateBatchAction3(Class clazz, FilterNode node, ColumnValue... values) {
this.clazz = clazz;
this.node = node;
this.values = values;
}
public UpdateBatchAction2(Class clazz, FilterNode node, Flipper flipper, ColumnValue... values) {
public UpdateBatchAction3(Class clazz, FilterNode node, Flipper flipper, ColumnValue... values) {
this.clazz = clazz;
this.node = node;
this.flipper = flipper;
@@ -1419,27 +1538,7 @@ public abstract class AbstractDataSource extends AbstractService implements Data
}
}
protected static class UpdateBatchAction3 {
public Object entity;
public FilterNode node;
public String[] columns;
public UpdateBatchAction3(Object entity, String... columns) {
this.entity = entity;
this.columns = columns;
}
public UpdateBatchAction3(Object entity, FilterNode node, String... columns) {
this.entity = entity;
this.node = node;
this.columns = columns;
}
}
protected static class UpdateBatchAction4 {
protected static class UpdateBatchAction4 extends BatchAction {
public Object entity;

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

@@ -110,6 +110,20 @@ public class DataJdbcSource extends DataSqlSource {
return false;
}
@Override
public CompletableFuture<Integer> batchAsync(final DataBatch batch) {
return CompletableFuture.supplyAsync(() -> batch(batch), getExecutor());
}
@Override
public int batch(final DataBatch batch) {
Objects.requireNonNull(batch);
final List<BatchAction> actions = ((DefaultDataBatch) batch).actions;
final DefaultBatchInfo batchInfo = parseBatchInfo((DefaultDataBatch) batch, this);
throw new UnsupportedOperationException("Not supported yet.");
}
@Override
protected <T> CompletableFuture<Integer> insertDB(EntityInfo<T> info, T... entitys) {
Connection conn = null;
@@ -117,8 +131,6 @@ public class DataJdbcSource extends DataSqlSource {
int c = 0;
conn = writePool.pollConnection();
final String sql = info.getInsertQuestionPrepareSQL(entitys[0]);
final Class primaryType = info.getPrimary().type();
final Attribute primary = info.getPrimary();
Attribute<T, Serializable>[] attrs = info.insertAttributes;
conn.setReadOnly(false);
conn.setAutoCommit(true);
@@ -149,14 +161,14 @@ public class DataJdbcSource extends DataSqlSource {
synchronized (info.disTableLock()) {
final String catalog = conn.getCatalog();
final String newTable = info.getTable(entitys[0]);
final String tablekey = newTable.indexOf('.') > 0 ? newTable : (catalog + '.' + newTable);
if (!info.containsDisTable(tablekey)) {
final String tableKey = newTable.indexOf('.') > 0 ? newTable : (catalog + '.' + newTable);
if (!info.containsDisTable(tableKey)) {
try {
//执行一遍复制表操作
Statement st = conn.createStatement();
st.execute(getTableCopySQL(info, newTable));
st.close();
info.addDisTable(tablekey);
info.addDisTable(tableKey);
} catch (SQLException sqle) { //多进程并发时可能会出现重复建表
if (isTableNotExist(info, sqle.getSQLState())) {
if (newTable.indexOf('.') < 0) {
@@ -176,7 +188,7 @@ public class DataJdbcSource extends DataSqlSource {
st = conn.createStatement();
st.execute(getTableCopySQL(info, newTable));
st.close();
info.addDisTable(tablekey);
info.addDisTable(tableKey);
}
} else { //需要先建库
Statement st;
@@ -192,7 +204,7 @@ public class DataJdbcSource extends DataSqlSource {
st = conn.createStatement();
st.execute(getTableCopySQL(info, newTable));
st.close();
info.addDisTable(tablekey);
info.addDisTable(tableKey);
} catch (SQLException sqle2) {
if (isTableNotExist(info, sqle2.getSQLState())) {
String[] tablesqls = createTableSqls(info);
@@ -211,7 +223,7 @@ public class DataJdbcSource extends DataSqlSource {
st = conn.createStatement();
st.execute(getTableCopySQL(info, newTable));
st.close();
info.addDisTable(tablekey);
info.addDisTable(tableKey);
}
} else {
logger.log(Level.SEVERE, "create table2(" + getTableCopySQL(info, newTable) + ") error", sqle2);

8
src/main/java/org/redkale/source/DataSource.java
View File

@@ -62,20 +62,20 @@ public interface DataSource extends Resourcable {
/**
* 增删改的批量操作
*
* @param bath 批量对象
* @param batch 批量对象
*
* @return -1表示失败，正数为成功
*/
//public int batch(final DataBatch bath);
public int batch(final DataBatch batch);
/**
* 增删改的批量操作
*
* @param bath 批量对象
* @param batch 批量对象
*
* @return -1表示失败，正数为成功
*/
//public CompletableFuture<Integer> batchAsync(final DataBatch bath);
public CompletableFuture<Integer> batchAsync(final DataBatch batch);
//----------------------insertAsync-----------------------------
/**

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

@@ -7,7 +7,6 @@ package org.redkale.source;
import java.io.Serializable;
import java.math.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@@ -41,14 +40,12 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
protected String name;
protected URL persistFile;
protected boolean cacheForbidden;
protected String dbtype;
private boolean autoddl;
private boolean autoDDL;
protected Properties readConfProps;
@@ -113,7 +110,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
protected void afterResourceChange() {
this.autoddl = "true".equals(readConfProps.getProperty(DATA_SOURCE_TABLE_AUTODDL, "false").trim());
this.autoDDL = "true".equals(readConfProps.getProperty(DATA_SOURCE_TABLE_AUTODDL, "false").trim());
this.containSQL = readConfProps.getProperty(DATA_SOURCE_CONTAIN_SQLTEMPLATE, "LOCATE(${keystr}, ${column}) > 0");
this.notContainSQL = readConfProps.getProperty(DATA_SOURCE_NOTCONTAIN_SQLTEMPLATE, "LOCATE(${keystr}, ${column}) = 0");
@@ -252,7 +249,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
//生成创建表的SQL
protected <T> String[] createTableSqls(EntityInfo<T> info) {
if (info == null || !autoddl) return null;
if (info == null || !autoDDL) return null;
javax.persistence.Table table = info.getType().getAnnotation(javax.persistence.Table.class);
if ("mysql".equals(dbtype())) { //mysql
StringBuilder sb = new StringBuilder();
@@ -463,9 +460,9 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
@Local
protected boolean isTableNotExist(EntityInfo info, String code) {
if (code == null || code.isEmpty()) return false;
return tableNotExistSqlstates.contains(';' + code + ';');
protected boolean isTableNotExist(EntityInfo info, String sqlCode) {
if (sqlCode == null || sqlCode.isEmpty()) return false;
return tableNotExistSqlstates.contains(';' + sqlCode + ';');
}
@Local
@@ -509,7 +506,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
@Local
public final boolean autoddl() {
return autoddl;
return autoDDL;
}
@Local
@@ -821,9 +818,18 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
protected <T> CompletableFuture<Integer> deleteCompose(final EntityInfo<T> info, final Serializable... pks) {
String sql = deleteSql(info, pks);
if (info.isLoggable(logger, Level.FINEST, sql)) logger.finest(info.getType().getSimpleName() + " delete sql=" + sql);
return deleteDB(info, null, sql);
}
protected <T> CompletableFuture<Integer> deleteCompose(final EntityInfo<T> info, final Flipper flipper, final FilterNode node) {
return deleteDB(info, flipper, deleteSql(info, flipper, node));
}
protected <T> String deleteSql(final EntityInfo<T> info, final Serializable... pks) {
if (pks.length == 1) {
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pks[0], sqlFormatter);
return deleteDB(info, null, sql);
return "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pks[0], sqlFormatter);
}
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + " IN (";
for (int i = 0; i < pks.length; i++) {
@@ -831,11 +837,10 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
sql += info.formatSQLValue(info.getPrimarySQLColumn(), pks[i], sqlFormatter);
}
sql += ")";
if (info.isLoggable(logger, Level.FINEST, sql)) logger.finest(info.getType().getSimpleName() + " delete sql=" + sql);
return deleteDB(info, null, sql);
return sql;
}
protected <T> CompletableFuture<Integer> deleteCompose(final EntityInfo<T> info, final Flipper flipper, final FilterNode node) {
protected <T> String deleteSql(final EntityInfo<T> info, final Flipper flipper, final FilterNode node) {
boolean pgsql = "postgresql".equals(dbtype());
Map<Class, String> joinTabalis = null;
CharSequence join = null;
@@ -865,7 +870,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2)))) + info.createSQLOrderby(flipper)
+ (("mysql".equals(dbtype()) && flipper != null && flipper.getLimit() > 0) ? (" LIMIT " + flipper.getLimit()) : "");
}
return deleteDB(info, flipper, sql);
return sql;
}
//----------------------------- clearTableCompose -----------------------------
@@ -1099,13 +1104,22 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, Serializable pk, String column, final Serializable colval) {
Attribute attr = info.getAttribute(column);
Serializable val = getSQLAttrValue(info, attr, colval);
SqlInfo sql = updateSql(info, pk, column, colval);
if (val instanceof byte[]) {
String sql = "UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + "=" + prepareParamSign(1) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter);
return updateColumnDB(info, null, sql, true, val);
return updateColumnDB(info, null, sql.sql, true, val);
} else {
String sql = "UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + "="
+ info.formatSQLValue(column, val, sqlFormatter) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter);
return updateColumnDB(info, null, sql, false);
return updateColumnDB(info, null, sql.sql, false);
}
}
protected <T> SqlInfo updateSql(final EntityInfo<T> info, Serializable pk, String column, final Serializable colval) {
Attribute attr = info.getAttribute(column);
Serializable val = getSQLAttrValue(info, attr, colval);
if (val instanceof byte[]) {
return new SqlInfo("UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + "=" + prepareParamSign(1) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter), (byte[]) val);
} else {
return new SqlInfo("UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + "="
+ info.formatSQLValue(column, val, sqlFormatter) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter));
}
}
@@ -1156,6 +1170,17 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final String column, final Serializable colval, final FilterNode node) {
Attribute attr = info.getAttribute(column);
Serializable val = getSQLAttrValue(info, attr, colval);
SqlInfo sql = updateSql(info, column, colval, node);
if (val instanceof byte[]) {
return updateColumnDB(info, null, sql.sql, true, val);
} else {
return updateColumnDB(info, null, sql.sql, false);
}
}
protected <T> SqlInfo updateSql(final EntityInfo<T> info, final String column, final Serializable colval, final FilterNode node) {
Map<Class, String> joinTabalis = node.getJoinTabalis();
CharSequence join = node.createSQLJoin(this, true, joinTabalis, new HashSet<>(), info);
CharSequence where = node.createSQLExpress(this, info, joinTabalis);
@@ -1175,13 +1200,13 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
+ " SET " + info.getSQLColumn(alias, column) + "=" + prepareParamSign(1)
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
return updateColumnDB(info, null, sql, true, val);
return new SqlInfo(sql, (byte[]) val);
} else {
String sql = "UPDATE " + info.getTable(node) + " a " + (join1 == null ? "" : (", " + join1))
+ " SET " + info.getSQLColumn(alias, column) + "=" + info.formatSQLValue(val, sqlFormatter)
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
return updateColumnDB(info, null, sql, false);
return new SqlInfo(sql);
}
}
@@ -1233,6 +1258,15 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final Serializable pk, final ColumnValue... values) {
SqlInfo sql = updateSql(info, pk, values);
if (sql.blobs == null || sql.blobs.isEmpty()) {
return updateColumnDB(info, null, sql.sql, false);
} else {
return updateColumnDB(info, null, sql.sql, true, sql.blobs.toArray());
}
}
protected <T> SqlInfo updateSql(final EntityInfo<T> info, final Serializable pk, final ColumnValue... values) {
StringBuilder setsql = new StringBuilder();
List<byte[]> blobs = null;
int index = 0;
@@ -1250,10 +1284,9 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
setsql.append(sqlColumn).append("=").append(info.formatSQLValue(sqlColumn, attr, col, sqlFormatter));
}
}
if (setsql.length() < 1) return CompletableFuture.completedFuture(0);
if (setsql.length() < 1) throw new RuntimeException("update non column-value array");
String sql = "UPDATE " + info.getTable(pk) + " SET " + setsql + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter);
if (blobs == null) return updateColumnDB(info, null, sql, false);
return updateColumnDB(info, null, sql, true, blobs.toArray());
return new SqlInfo(sql, blobs);
}
@Override
@@ -1294,6 +1327,15 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final FilterNode node, final Flipper flipper, final ColumnValue... values) {
SqlInfo sql = updateSql(info, node, flipper, values);
if (sql.blobs == null || sql.blobs.isEmpty()) {
return updateColumnDB(info, flipper, sql.sql, false);
} else {
return updateColumnDB(info, flipper, sql.sql, true, sql.blobs.toArray());
}
}
protected <T> SqlInfo updateSql(final EntityInfo<T> info, final FilterNode node, final Flipper flipper, final ColumnValue... values) {
StringBuilder setsql = new StringBuilder();
List<byte[]> blobs = null;
int index = 0;
@@ -1313,7 +1355,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
setsql.append(sqlColumn).append("=").append(info.formatSQLValue(sqlColumn, attr, col, sqlFormatter));
}
}
if (setsql.length() < 1) return CompletableFuture.completedFuture(0);
if (setsql.length() < 1) throw new RuntimeException("update non column-value array");
Map<Class, String> joinTabalis = node == null ? null : node.getJoinTabalis();
CharSequence join = node == null ? null : node.createSQLJoin(this, true, joinTabalis, new HashSet<>(), info);
CharSequence where = node == null ? null : node.createSQLExpress(this, info, joinTabalis);
@@ -1338,8 +1380,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
+ info.createSQLOrderby(flipper)
+ (("mysql".equals(dbtype()) && flipper != null && flipper.getLimit() > 0) ? (" LIMIT " + flipper.getLimit()) : "");
}
if (blobs == null) return updateColumnDB(info, flipper, sql, false);
return updateColumnDB(info, flipper, sql, true, blobs.toArray());
return new SqlInfo(sql, blobs);
}
//返回不存在的字段名,null表示字段都合法;
@@ -1447,7 +1488,16 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
});
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final boolean neednode, final T entity, final FilterNode node, final SelectColumn selects) {
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final boolean needNode, final T entity, final FilterNode node, final SelectColumn selects) {
SqlInfo sql = updateSql(info, needNode, entity, node, selects);
if (sql.blobs == null || sql.blobs.isEmpty()) {
return updateColumnDB(info, null, sql.sql, false);
} else {
return updateColumnDB(info, null, sql.sql, true, sql.blobs.toArray());
}
}
protected <T> SqlInfo updateSql(final EntityInfo<T> info, final boolean needNode, final T entity, final FilterNode node, final SelectColumn selects) {
StringBuilder setsql = new StringBuilder();
List<byte[]> blobs = null;
int index = 0;
@@ -1467,7 +1517,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
setsql.append("=").append(sqlval);
}
}
if (neednode) {
if (needNode) {
Map<Class, String> joinTabalis = node.getJoinTabalis();
CharSequence join = node.createSQLJoin(this, true, joinTabalis, new HashSet<>(), info);
CharSequence where = node.createSQLExpress(this, info, joinTabalis);
@@ -1481,17 +1531,15 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
String sql = "UPDATE " + info.getTable(node) + " a " + (join1 == null ? "" : (", " + join1)) + " SET " + setsql
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
if (blobs == null) return updateColumnDB(info, null, sql, false);
return updateColumnDB(info, null, sql, true, blobs.toArray());
return new SqlInfo(sql, blobs);
} else {
final Serializable id = (Serializable) info.getSQLValue(info.getPrimary(), entity);
String sql = "UPDATE " + info.getTable(id) + " a SET " + setsql + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(id, sqlFormatter);
if (blobs == null) return updateColumnDB(info, null, sql, false);
return updateColumnDB(info, null, sql, true, blobs.toArray());
return new SqlInfo(sql, blobs);
}
}
protected <T> int updateCache(final EntityInfo<T> info, int count, final boolean neednode, final T entity, final FilterNode node, final SelectColumn selects) {
protected <T> int updateCache(final EntityInfo<T> info, int count, final boolean needNode, final T entity, final FilterNode node, final SelectColumn selects) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return count;
final List<Attribute<T, Serializable>> attrs = new ArrayList<>();
@@ -1499,7 +1547,7 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
if (!selects.test(attr.field())) continue;
attrs.add(attr);
}
if (neednode) {
if (needNode) {
T[] rs = cache.update(entity, attrs, node);
return count >= 0 ? count : (rs == null ? 0 : rs.length);
} else {
@@ -2324,8 +2372,26 @@ public abstract class DataSqlSource extends AbstractDataSource implements Functi
return querySheetDB(info, readcache, needtotal, distinct, selects, flipper, node);
}
protected static enum UpdateMode {
INSERT, DELETE, UPDATE, CLEAR, DROP, ALTER, OTHER;
}
protected static class SqlInfo {
public String sql;
public List<byte[]> blobs;
public SqlInfo(String sql, byte[]... blobs) {
this.sql = sql;
if (blobs.length > 0) {
this.blobs = new ArrayList<>();
for (byte[] bs : blobs) {
this.blobs.add(bs);
}
}
}
public SqlInfo(String sql, List<byte[]> blobs) {
this.sql = sql;
this.blobs = blobs;
}
}
}

18
src/main/java/org/redkale/source/EntityInfo.java
View File

@@ -85,8 +85,8 @@ public final class EntityInfo<T> {
//所有可更新字段，即排除了主键字段和标记为&#064;Column(updatable=false)的字段
private final Map<String, Attribute<T, Serializable>> updateAttributeMap = new HashMap<>();
//用于存在database.table_20160202类似这种分布式表
private final Set<String> tables = new CopyOnWriteArraySet<>();
//用于存在database.table_20160202类似这种分布式表, 服务分布式部署时不存在的表名不一定真实不存在
private final Set<String> disTables = new CopyOnWriteArraySet<>();
//不能为null的字段名
private final Set<String> notNullColumns = new CopyOnWriteArraySet<>();
@@ -629,19 +629,19 @@ public final class EntityInfo<T> {
}
public Object disTableLock() {
return tables;
return disTables;
}
public boolean containsDisTable(String tablekey) {
return tables.contains(tablekey);
public boolean containsDisTable(String tableKey) {
return disTables.contains(tableKey);
}
public void addDisTable(String tablekey) {
tables.add(tablekey);
public void addDisTable(String tableKey) {
disTables.add(tableKey);
}
public boolean removeDisTable(String tablekey) {
return tables.remove(tablekey);
public boolean removeDisTable(String tableKey) {
return disTables.remove(tableKey);
}
public EntityColumn[] getDDLColumns() {