diff --git a/src/main/java/org/redkale/source/AbstractDataSqlSource.java b/src/main/java/org/redkale/source/AbstractDataSqlSource.java
new file mode 100644
index 000000000..13ba55839
--- /dev/null
+++ b/src/main/java/org/redkale/source/AbstractDataSqlSource.java
@@ -0,0 +1,3283 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.redkale.source;
+
+import java.io.Serializable;
+import java.math.*;
+import java.sql.SQLException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.function.*;
+import java.util.logging.*;
+import java.util.stream.Stream;
+import org.redkale.annotation.AutoLoad;
+import org.redkale.annotation.*;
+import org.redkale.annotation.ResourceListener;
+import org.redkale.annotation.ResourceType;
+import static org.redkale.boot.Application.*;
+import org.redkale.net.AsyncGroup;
+import org.redkale.persistence.Table;
+import org.redkale.service.Local;
+import org.redkale.source.EntityInfo.EntityColumn;
+import org.redkale.util.*;
+
+/**
+ * DataSource的SQL抽象实现类
+ * 注意: 所有的操作只能作用在一张表上，不能同时变更多张表
+ *
+ *
+ * 详情见: https://redkale.org
+ *
+ * @author zhangjx
+ */
+@Local
+@AutoLoad(false)
+@SuppressWarnings("unchecked")
+@ResourceType(DataSource.class)
+public abstract class AbstractDataSqlSource extends AbstractDataSource implements DataSqlSource, Function {
+
+ //不存在分表时最大重试次数
+ protected static final int MAX_RETRYS = 3;
+
+ protected static final Flipper FLIPPER_ONE = new Flipper(1);
+
+ protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
+
+ protected String name;
+
+ protected boolean cacheForbidden;
+
+ protected String dbtype;
+
+ private boolean autoDDL;
+
+ protected Properties readConfProps;
+
+ protected Properties writeConfProps;
+
+ @Resource(name = RESNAME_APP_CLIENT_ASYNCGROUP, required = false)
+ protected AsyncGroup clientAsyncGroup;
+
+ //配置 APP_EXECUTOR资源为null
+ @Resource(name = RESNAME_APP_EXECUTOR, required = false)
+ protected ExecutorService workExecutor;
+
+ protected BiFunction sqlFormatter;
+
+ protected BiConsumer errorCompleteConsumer = (r, t) -> {
+ //if (t != null) logger.log(Level.INFO, "CompletableFuture complete error", (Throwable) t);
+ };
+
+ protected final BiFunction> fullloader = (s, i)
+ -> ((CompletableFuture) querySheetDBAsync(i, false, false, false, null, null, (FilterNode) null)).thenApply(e -> e == null ? new ArrayList() : e.list(true));
+
+ //超过多少毫秒视为较慢, 会打印警告级别的日志, 默认值: 2000
+ protected long slowmsWarn;
+
+ //超过多少毫秒视为很慢, 会打印错误级别的日志, 默认值: 3000
+ protected long slowmsError;
+
+ //用于反向LIKE使用
+ protected String containSQL;
+
+ //用于反向LIKE使用
+ protected String notContainSQL;
+
+ //用于判断表不存在的使用, 多个SQLState用;隔开
+ protected String tableNotExistSqlstates;
+
+ //用于复制表结构使用, sql语句必须包含IF NOT EXISTS判断，确保重复执行不会报错
+ protected String tablecopySQL;
+
+ protected AnyValue config;
+
+ private EntityInfo currEntityInfo;
+
+ public AbstractDataSqlSource() {
+ }
+
+ @Override
+ public void init(AnyValue conf) {
+ super.init(conf);
+ this.config = conf;
+ if (conf.getAnyValue("read") == null) { //没有读写分离
+ Properties rwConf = new Properties();
+ conf.forEach((k, v) -> rwConf.put(k, decryptProperty(k, v)));
+ this.dbtype = parseDbtype(rwConf.getProperty(DATA_SOURCE_URL));
+ initProperties(rwConf);
+ this.readConfProps = rwConf;
+ this.writeConfProps = rwConf;
+ } else { //读写分离
+ Properties readConf = new Properties();
+ Properties writeConf = new Properties();
+ conf.getAnyValue("read").forEach((k, v) -> readConf.put(k, decryptProperty(k, v)));
+ conf.getAnyValue("write").forEach((k, v) -> writeConf.put(k, decryptProperty(k, v)));
+ this.dbtype = parseDbtype(readConf.getProperty(DATA_SOURCE_URL));
+ initProperties(readConf);
+ initProperties(writeConf);
+ this.readConfProps = readConf;
+ this.writeConfProps = writeConf;
+ }
+ this.name = conf.getValue("name", "");
+ this.sqlFormatter = (info, val) -> formatValueToString(info, val);
+ afterResourceChange();
+ }
+
+ protected void afterResourceChange() {
+ 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");
+
+ this.tableNotExistSqlstates = ";" + readConfProps.getProperty(DATA_SOURCE_TABLENOTEXIST_SQLSTATES, "42000;42S02") + ";";
+ this.tablecopySQL = readConfProps.getProperty(DATA_SOURCE_TABLECOPY_SQLTEMPLATE, "CREATE TABLE IF NOT EXISTS ${newtable} LIKE ${oldtable}");
+
+ this.cacheForbidden = "NONE".equalsIgnoreCase(readConfProps.getProperty(DATA_SOURCE_CACHEMODE));
+ this.slowmsWarn = Integer.parseInt(readConfProps.getProperty(DATA_SOURCE_SLOWMS_WARN, "2000").trim());
+ this.slowmsError = Integer.parseInt(readConfProps.getProperty(DATA_SOURCE_SLOWMS_ERROR, "3000").trim());
+ }
+
+ @Override
+ @ResourceListener
+ public void onResourceChange(ResourceEvent[] events) {
+ if (events == null || events.length < 1) {
+ return;
+ }
+ //不支持读写分离模式的动态切换
+ if (readConfProps == writeConfProps && (events[0].name().startsWith("read.") || events[0].name().startsWith("write."))) {
+ throw new SourceException("DataSource(name=" + resourceName() + ") not support to change to read/write separation mode");
+ }
+ if (readConfProps != writeConfProps && (!events[0].name().startsWith("read.") && !events[0].name().startsWith("write."))) {
+ throw new SourceException("DataSource(name=" + resourceName() + ") not support to change to non read/write separation mode");
+ }
+
+ StringBuilder sb = new StringBuilder();
+ if (readConfProps == writeConfProps) {
+ List allEvents = new ArrayList<>();
+ Properties newProps = new Properties();
+ newProps.putAll(this.readConfProps);
+ for (ResourceEvent event : events) { //可能需要解密
+ String newValue = decryptProperty(event.name(), event.newValue().toString());
+ allEvents.add(ResourceEvent.create(event.name(), newValue, event.oldValue()));
+ newProps.put(event.name(), newValue);
+ sb.append("DataSource(name=").append(resourceName()).append(") change '")
+ .append(event.name()).append("' to '").append(event.coverNewValue()).append("'\r\n");
+ }
+ updateOneResourceChange(newProps, allEvents.toArray(new ResourceEvent[allEvents.size()]));
+ for (ResourceEvent event : allEvents) {
+ this.readConfProps.put(event.name(), event.newValue());
+ }
+ } else {
+ List readEvents = new ArrayList<>();
+ List writeEvents = new ArrayList<>();
+ Properties newReadProps = new Properties();
+ newReadProps.putAll(this.readConfProps);
+ Properties newWriteProps = new Properties();
+ newWriteProps.putAll(this.writeConfProps);
+ for (ResourceEvent event : events) {
+ if (event.name().startsWith("read.")) {
+ String newName = event.name().substring("read.".length());
+ String newValue = decryptProperty(event.name(), event.newValue().toString());
+ readEvents.add(ResourceEvent.create(newName, newValue, event.oldValue()));
+ newReadProps.put(event.name(), newValue);
+ } else {
+ String newName = event.name().substring("write.".length());
+ String newValue = decryptProperty(event.name(), event.newValue().toString());
+ writeEvents.add(ResourceEvent.create(newName, newValue, event.oldValue()));
+ newWriteProps.put(event.name(), newValue);
+ }
+ sb.append("DataSource(name=").append(resourceName()).append(") change '")
+ .append(event.name()).append("' to '").append(event.coverNewValue()).append("'\r\n");
+ }
+ if (!readEvents.isEmpty()) {
+ updateReadResourceChange(newReadProps, readEvents.toArray(new ResourceEvent[readEvents.size()]));
+ }
+ if (!writeEvents.isEmpty()) {
+ updateWriteResourceChange(newWriteProps, writeEvents.toArray(new ResourceEvent[writeEvents.size()]));
+ }
+ //更新Properties
+ if (!readEvents.isEmpty()) {
+ for (ResourceEvent event : readEvents) {
+ this.readConfProps.put(event.name(), event.newValue());
+ }
+ }
+ if (!writeEvents.isEmpty()) {
+ for (ResourceEvent event : writeEvents) {
+ this.writeConfProps.put(event.name(), event.newValue());
+ }
+ }
+ }
+ afterResourceChange();
+ if (sb.length() > 0) {
+ logger.log(Level.INFO, sb.toString());
+ }
+ }
+
+ protected void updateOneResourceChange(Properties newProps, ResourceEvent[] events) {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+
+ protected void updateReadResourceChange(Properties newReadProps, ResourceEvent[] events) {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+
+ protected void updateWriteResourceChange(Properties newWriteProps, ResourceEvent[] events) {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+
+ protected void slowLog(long startTime, String... sqls) {
+ long cost = System.currentTimeMillis() - startTime;
+ if (slowmsError > 0 && cost > slowmsError) {
+ logger.log(Level.SEVERE, DataSource.class.getSimpleName() + "(name='" + resourceName() + "') slow sql cost " + cost + " ms, content: " + Arrays.toString(sqls));
+ } else if (slowmsWarn > 0 && cost > slowmsWarn) {
+ logger.log(Level.WARNING, DataSource.class.getSimpleName() + "(name='" + resourceName() + "') very slow sql cost " + cost + " ms, content: " + Arrays.toString(sqls));
+ }
+ }
+
+ protected String parseNotExistTableName(SQLException e) {
+ String errmsg = e.getMessage();
+ char quote = '"';
+ String tableName = null;
+ int pos = errmsg.indexOf(quote);
+ if (pos < 0) {
+ quote = '\'';
+ pos = errmsg.indexOf(quote);
+ }
+ if (pos >= 0) {
+ int pos2 = errmsg.indexOf(quote, pos + 1);
+ if (pos2 > pos) {
+ tableName = errmsg.substring(pos + 1, pos2);
+ }
+ }
+ return tableName;
+ }
+
+ //解密可能存在的加密字段, 可重载
+ protected String decryptProperty(String key, String value) {
+ return value;
+ }
+
+ protected void initProperties(Properties props) {
+ if ("oracle".equals(this.dbtype)) {
+ props.setProperty(DATA_SOURCE_CONTAIN_SQLTEMPLATE, "INSTR(${keystr}, ${column}) > 0");
+ props.setProperty(DATA_SOURCE_NOTCONTAIN_SQLTEMPLATE, "INSTR(${keystr}, ${column}) = 0");
+ if (!props.containsKey(DATA_SOURCE_TABLENOTEXIST_SQLSTATES)) {
+ props.setProperty(DATA_SOURCE_TABLENOTEXIST_SQLSTATES, "42000;42S02");
+ }
+ if (!props.containsKey(DATA_SOURCE_TABLECOPY_SQLTEMPLATE)) {
+ //注意：此语句复制表结构会导致默认值和主键信息的丢失
+ props.setProperty(DATA_SOURCE_TABLECOPY_SQLTEMPLATE, "CREATE TABLE IF NOT EXISTS ${newtable} AS SELECT * FROM ${oldtable} WHERE 1=2");
+ }
+ } else if ("sqlserver".equals(this.dbtype)) {
+ props.setProperty(DATA_SOURCE_CONTAIN_SQLTEMPLATE, "CHARINDEX(${column}, ${keystr}) > 0");
+ props.setProperty(DATA_SOURCE_NOTCONTAIN_SQLTEMPLATE, "CHARINDEX(${column}, ${keystr}) = 0");
+ } else if ("postgresql".equals(this.dbtype)) {
+ if (!props.containsKey(DATA_SOURCE_TABLECOPY_SQLTEMPLATE)) { //注意：此语句复制表结构会导致默认值和主键信息的丢失
+ //注意：postgresql不支持跨库复制表结构
+ //props.setProperty(DATA_SOURCE_TABLECOPY_SQLTEMPLATE, "CREATE TABLE ${newtable} AS (SELECT * FROM ${oldtable} LIMIT 0)");
+ props.setProperty(DATA_SOURCE_TABLECOPY_SQLTEMPLATE, "CREATE TABLE IF NOT EXISTS ${newtable} (LIKE ${oldtable} INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING COMMENTS INCLUDING INDEXES)");
+ }
+ if (!props.containsKey(DATA_SOURCE_TABLENOTEXIST_SQLSTATES)) {
+ props.setProperty(DATA_SOURCE_TABLENOTEXIST_SQLSTATES, "42P01;3F000");
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ if (readConfProps == null) { //compileMode模式下会为null
+ return getClass().getSimpleName() + "{}";
+ }
+ if (readConfProps == writeConfProps) {
+ String url = readConfProps.getProperty(DATA_SOURCE_URL);
+ int pos = url.indexOf('?');
+ if (pos > 0) {
+ url = url.substring(0, pos) + "...";
+ }
+ return getClass().getSimpleName() + "{url=" + url + "}";
+ } else {
+ String readUrl = readConfProps.getProperty(DATA_SOURCE_URL);
+ int pos = readUrl.indexOf('?');
+ if (pos > 0) {
+ readUrl = readUrl.substring(0, pos) + "...";
+ }
+ String writeUrl = writeConfProps.getProperty(DATA_SOURCE_URL);
+ pos = writeUrl.indexOf('?');
+ if (pos > 0) {
+ writeUrl = writeUrl.substring(0, pos) + "...";
+ }
+ return getClass().getSimpleName() + "{readurl=" + readUrl + ",writeurl=" + writeUrl + "}";
+ }
+ }
+
+ //生成创建表的SQL
+ protected String[] createTableSqls(EntityInfo info) {
+ if (info == null || !autoDDL) {
+ return null;
+ }
+ Table table = info.getType().getAnnotation(Table.class);
+ if ("mysql".equals(dbtype())) { //mysql
+ StringBuilder sb = new StringBuilder();
+ sb.append("CREATE TABLE IF NOT EXISTS `").append(info.getOriginTable()).append("`(\n");
+ EntityColumn primary = null;
+ T one = info.constructorAttributes == null ? info.getCreator().create() : null;
+ for (EntityColumn column : info.getDDLColumns()) {
+ if (column.primary) {
+ primary = column;
+ }
+ String sqltype = "VARCHAR(" + column.length + ")";
+ String sqlnull = column.primary ? "NOT NULL" : "NULL";
+ if (column.type == boolean.class || column.type == Boolean.class) {
+ sqltype = "TINYINT(1)";
+ Boolean val = one == null ? null : (Boolean) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val != null && val ? 1 : 0);
+ } else if (column.type == byte.class || column.type == Byte.class) {
+ sqltype = "TINYINT";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val.byteValue());
+ } else if (column.type == short.class || column.type == Short.class) {
+ sqltype = "SMALLINT";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == char.class || column.type == Character.class) {
+ sqltype = "SMALLINT UNSIGNED";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val.intValue());
+ } else if (column.type == int.class || column.type == Integer.class || column.type == AtomicInteger.class) {
+ sqltype = "INT";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == long.class || column.type == Long.class || column.type == AtomicLong.class || column.type == LongAdder.class) {
+ sqltype = "BIGINT";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == float.class || column.type == Float.class) {
+ sqltype = "FLOAT";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == double.class || column.type == Double.class) {
+ sqltype = "DOUBLE";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == BigInteger.class) {
+ sqltype = "DECIMAL";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ } else {
+ sqltype += "(19,2)";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == BigDecimal.class) {
+ sqltype = "DECIMAL";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == String.class) {
+ if (column.length < 65535) {
+ String val = one == null ? null : (String) info.getAttribute(column.field).get(one);
+ if (val != null) {
+ sqlnull = "NOT NULL DEFAULT '" + val.replace('\'', '"') + "'";
+ } else if (column.primary) {
+ sqlnull = "NOT NULL DEFAULT ''";
+ }
+ } else if (column.length == 65535) {
+ sqltype = "TEXT";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ } else if (column.length <= 16777215) {
+ sqltype = "MEDIUMTEXT";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ } else {
+ sqltype = "LONGTEXT";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ }
+ } else if (column.type == byte[].class) {
+ if (column.length <= 65535) {
+ sqltype = "BLOB";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ } else if (column.length <= 16777215) {
+ sqltype = "MEDIUMBLOB";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ } else {
+ sqltype = "LONGBLOB";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ }
+ } else if (column.type == java.time.LocalDate.class || column.type == java.util.Date.class || "java.sql.Date".equals(column.type.getName())) {
+ sqltype = "DATE";
+ } else if (column.type == java.time.LocalTime.class || "java.sql.Time".equals(column.type.getName())) {
+ sqltype = "TIME";
+ } else if (column.type == java.time.LocalDateTime.class || "java.sql.Timestamp".equals(column.type.getName())) {
+ sqltype = "DATETIME";
+ } else { //JavaBean
+ sqltype = column.length >= 65535 ? "TEXT" : ("VARCHAR(" + column.length + ")");
+ sqlnull = !column.nullable ? "NOT NULL DEFAULT ''" : "NULL";
+ }
+ sb.append(" `").append(column.column).append("` ").append(sqltype).append(" ").append(sqlnull);
+ if (column.comment != null && !column.comment.isEmpty()) {
+ sb.append(" COMMENT '").append(column.comment.replace('\'', '"')).append("'");
+ }
+ sb.append(",\n");
+ }
+ sb.append(" PRIMARY KEY (`").append(primary.column).append("`)\n");
+ sb.append(")ENGINE=InnoDB DEFAULT CHARSET=UTF8MB4");
+ if (table != null && !table.comment().isEmpty()) {
+ sb.append(" COMMENT '").append(table.comment().replace('\'', '"')).append("'");
+ }
+ return Utility.ofArray(sb.toString());
+ } else if ("postgresql".equals(dbtype())) { //postgresql
+ StringBuilder sb = new StringBuilder();
+ sb.append("CREATE TABLE IF NOT EXISTS ").append(info.getOriginTable()).append("(\n");
+ EntityColumn primary = null;
+ T one = info.constructorAttributes == null ? info.getCreator().create() : null;
+ List comments = new ArrayList<>();
+ if (table != null && !table.comment().isEmpty()) {
+ comments.add("COMMENT ON TABLE " + info.getOriginTable() + " IS '" + table.comment().replace('\'', '"') + "'");
+ }
+ for (EntityColumn column : info.getDDLColumns()) {
+ if (column.primary) {
+ primary = column;
+ }
+ String sqltype = "VARCHAR(" + column.length + ")";
+ String sqlnull = column.primary ? "NOT NULL" : "NULL";
+ if (column.type == boolean.class || column.type == Boolean.class) {
+ sqltype = "BOOL";
+ Boolean val = one == null ? null : (Boolean) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val != null && val ? 1 : 0);
+ } else if (column.type == byte.class || column.type == Byte.class) {
+ sqltype = "INT2";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val.byteValue());
+ } else if (column.type == short.class || column.type == Short.class) {
+ sqltype = "INT2";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == char.class || column.type == Character.class) {
+ sqltype = "INT2 UNSIGNED";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val.intValue());
+ } else if (column.type == int.class || column.type == Integer.class || column.type == AtomicInteger.class) {
+ sqltype = "INT4";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == long.class || column.type == Long.class || column.type == AtomicLong.class || column.type == LongAdder.class) {
+ sqltype = "INT8";
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == float.class || column.type == Float.class) {
+ sqltype = "FLOAT4";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == double.class || column.type == Double.class) {
+ sqltype = "FLOAT8";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == BigInteger.class) {
+ sqltype = "NUMERIC";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ } else {
+ sqltype += "(19,2)";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == BigDecimal.class) {
+ sqltype = "NUMERIC";
+ if (column.precision > 0) {
+ sqltype += "(" + column.precision + "," + column.scale + ")";
+ }
+ Number val = one == null ? null : (Number) info.getAttribute(column.field).get(one);
+ sqlnull = "NOT NULL DEFAULT " + (val == null ? 0 : val);
+ } else if (column.type == String.class) {
+ if (column.length < 65535) {
+ String val = one == null ? null : (String) info.getAttribute(column.field).get(one);
+ if (val != null) {
+ sqlnull = "NOT NULL DEFAULT '" + val.replace('\'', '"') + "'";
+ } else if (column.primary) {
+ sqlnull = "NOT NULL DEFAULT ''";
+ }
+ } else {
+ sqltype = "TEXT";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ }
+ } else if (column.type == byte[].class) {
+ sqltype = "BYTEA";
+ if (!column.nullable) {
+ sqlnull = "NOT NULL";
+ }
+ } else if (column.type == java.time.LocalDate.class || column.type == java.util.Date.class || "java.sql.Date".equals(column.type.getName())) {
+ sqltype = "DATE";
+ } else if (column.type == java.time.LocalTime.class || "java.sql.Time".equals(column.type.getName())) {
+ sqltype = "TIME";
+ } else if (column.type == java.time.LocalDateTime.class || "java.sql.Timestamp".equals(column.type.getName())) {
+ sqltype = "TIMESTAMP";
+ } else { //JavaBean
+ sqltype = column.length >= 65535 ? "TEXT" : ("VARCHAR(" + column.length + ")");
+ sqlnull = !column.nullable ? "NOT NULL DEFAULT ''" : "NULL";
+ }
+ sb.append(" ").append(column.column).append(" ").append(sqltype).append(" ").append(sqlnull);
+ if (column.comment != null && !column.comment.isEmpty()) {
+ //postgresql不支持DDL中直接带comment
+ comments.add("COMMENT ON COLUMN " + info.getOriginTable() + "." + column.column + " IS '" + column.comment.replace('\'', '"') + "'");
+ }
+ sb.append(",\n");
+ }
+ sb.append(" PRIMARY KEY (").append(primary.column).append(")\n");
+ sb.append(")");
+ return Utility.append(Utility.ofArray(sb.toString()), comments);
+ }
+ return null;
+ }
+
+ @Local
+ protected boolean isTableNotExist(EntityInfo info, String sqlCode) {
+ return sqlCode != null && !sqlCode.isEmpty() && tableNotExistSqlstates.contains(';' + sqlCode + ';');
+ }
+
+ @Local
+ protected String getTableCopySQL(EntityInfo info, String newTable) {
+ return tablecopySQL.replace("${newtable}", newTable).replace("${oldtable}", info.table);
+ }
+
+ @Local
+ protected Serializable getSQLAttrValue(EntityInfo info, Attribute attr, Serializable val) {
+ if (val != null && !(val instanceof Number) && !(val instanceof CharSequence) && !(val instanceof java.util.Date)
+ && !val.getClass().getName().startsWith("java.sql.") && !val.getClass().getName().startsWith("java.time.")) {
+ val = info.jsonConvert.convertTo(attr.genericType(), val);
+ } else if (val == null && info.isNotNullJson(attr)) {
+ val = "";
+ }
+ return val;
+ }
+
+ @Local
+ protected Map> getInsertQuestionPrepareInfo(EntityInfo info, T... entitys) {
+ Map> map = new LinkedHashMap<>();//一定要是LinkedHashMap
+ for (T entity : entitys) {
+ String table = info.getTable(entity);
+ map.computeIfAbsent(table, t -> new PrepareInfo(info.getInsertQuestionPrepareSQL(entity))).addEntity(entity);
+ }
+ return map;
+ }
+
+ @Local
+ protected Map> getInsertDollarPrepareInfo(EntityInfo info, T... entitys) {
+ Map> map = new LinkedHashMap<>();//一定要是LinkedHashMap
+ for (T entity : entitys) {
+ String table = info.getTable(entity);
+ map.computeIfAbsent(table, t -> new PrepareInfo(info.getInsertDollarPrepareSQL(entity))).addEntity(entity);
+ }
+ return map;
+ }
+
+ @Local
+ protected Map> getUpdateQuestionPrepareInfo(EntityInfo info, T... entitys) {
+ Map> map = new LinkedHashMap<>(); //一定要是LinkedHashMap
+ for (T entity : entitys) {
+ String table = info.getTable(entity);
+ map.computeIfAbsent(table, t -> new PrepareInfo(info.getUpdateQuestionPrepareSQL(entity))).addEntity(entity);
+ }
+ return map;
+ }
+
+ @Local
+ protected Map> getUpdateDollarPrepareInfo(EntityInfo info, T... entitys) {
+ Map> map = new LinkedHashMap<>();//一定要是LinkedHashMap
+ for (T entity : entitys) {
+ String table = info.getTable(entity);
+ map.computeIfAbsent(table, t -> new PrepareInfo(info.getUpdateDollarPrepareSQL(entity))).addEntity(entity);
+ }
+ return map;
+ }
+
+ @Local
+ protected Serializable getEntityAttrValue(EntityInfo info, Attribute attr, T entity) {
+ Serializable val = info.getSQLValue(attr, entity);
+ Class clazz = attr.type();
+ if (clazz == String.class
+ || clazz == int.class || clazz == long.class
+ || clazz == Integer.class || clazz == Long.class
+ || clazz == short.class || clazz == Short.class
+ || clazz == float.class || clazz == Float.class
+ || clazz == double.class || clazz == Double.class) {
+ return val;
+ }
+ return getSQLAttrValue(info, attr, val);
+ }
+
+ @Override
+ public void destroy(AnyValue config) {
+ super.destroy(config);
+ }
+
+ @Override
+ @Local
+ public void compile(Class clazz) {
+ EntityInfo.compile(clazz, this);
+ }
+
+ @Local
+ public final String dbtype() {
+ if (dbtype == null) {
+ throw new NullPointerException("dbtype is null");
+ }
+ return dbtype;
+ }
+
+ @Local
+ public final boolean autoddl() {
+ return autoDDL;
+ }
+
+ @Local
+ public abstract int directExecute(String sql);
+
+ @Local
+ public abstract int[] directExecute(String... sqls);
+
+ @Local
+ public abstract V directQuery(String sql, Function handler);
+
+ //是否异步
+ protected abstract boolean isAsync();
+
+ //index从1开始
+ protected abstract String prepareParamSign(int index);
+
+ //插入纪录
+ protected abstract CompletableFuture insertDBAsync(final EntityInfo info, T... entitys);
+
+ //删除记录
+ protected abstract CompletableFuture deleteDBAsync(final EntityInfo info, String[] tables, Flipper flipper, FilterNode node, Map> pkmap, final String... sqls);
+
+ //清空表
+ protected abstract CompletableFuture clearTableDBAsync(final EntityInfo info, String[] tables, FilterNode node, final String... sqls);
+
+ //建表
+ protected abstract CompletableFuture createTableDBAsync(final EntityInfo info, String copyTableSql, Serializable pk, final String... sqls);
+
+ //删除表
+ protected abstract CompletableFuture dropTableDBAsync(final EntityInfo info, String[] tables, FilterNode node, final String... sqls);
+
+ //更新纪录
+ protected abstract CompletableFuture updateEntityDBAsync(final EntityInfo info, T... entitys);
+
+ //更新纪录
+ protected abstract CompletableFuture updateColumnDBAsync(final EntityInfo info, Flipper flipper, final UpdateSqlInfo sql);
+
+ //查询Number Map数据
+ protected abstract CompletableFuture> getNumberMapDBAsync(final EntityInfo info, String[] tables, final String sql, final FilterNode node, final FilterFuncColumn... columns);
+
+ //查询Number数据
+ protected abstract CompletableFuture getNumberResultDBAsync(final EntityInfo info, String[] tables, final String sql, final FilterFunc func, final Number defVal, final String column, final FilterNode node);
+
+ //查询Map数据
+ protected abstract CompletableFuture> queryColumnMapDBAsync(final EntityInfo info, String[] tables, final String sql, final String keyColumn, final FilterFunc func, final String funcColumn, FilterNode node);
+
+ //查询Map数据
+ protected abstract CompletableFuture> queryColumnMapDBAsync(final EntityInfo info, String[] tables, final String sql, final ColumnNode[] funcNodes, final String[] groupByColumns, final FilterNode node);
+
+ //查询单条记录
+ protected abstract CompletableFuture findDBAsync(final EntityInfo info, String[] tables, final String sql, final boolean onlypk, final SelectColumn selects, final Serializable pk, final FilterNode node);
+
+ //查询单条记录的单个字段
+ protected abstract CompletableFuture findColumnDBAsync(final EntityInfo info, String[] tables, final String sql, final boolean onlypk, final String column, final Serializable defValue, final Serializable pk, final FilterNode node);
+
+ //判断记录是否存在
+ protected abstract CompletableFuture existsDBAsync(final EntityInfo info, final String[] tables, final String sql, final boolean onlypk, final Serializable pk, final FilterNode node);
+
+ //查询一页数据
+ protected abstract CompletableFuture> querySheetDBAsync(final EntityInfo info, final boolean readcache, final boolean needtotal, final boolean distinct, final SelectColumn selects, final Flipper flipper, final FilterNode node);
+
+ //插入纪录
+ protected int insertDB(final EntityInfo info, T... entitys) {
+ return insertDBAsync(info, entitys).join();
+ }
+
+ //删除记录
+ protected int deleteDB(final EntityInfo info, String[] tables, Flipper flipper, FilterNode node, Map> pkmap, final String... sqls) {
+ return deleteDBAsync(info, tables, flipper, node, pkmap, sqls).join();
+ }
+
+ //清空表
+ protected int clearTableDB(final EntityInfo info, String[] tables, FilterNode node, final String... sqls) {
+ return clearTableDBAsync(info, tables, node, sqls).join();
+ }
+
+ //建表
+ protected int createTableDB(final EntityInfo info, String copyTableSql, Serializable pk, final String... sqls) {
+ return createTableDBAsync(info, copyTableSql, pk, sqls).join();
+ }
+
+ //删除表
+ protected int dropTableDB(final EntityInfo info, String[] tables, FilterNode node, final String... sqls) {
+ return dropTableDBAsync(info, tables, node, sqls).join();
+ }
+
+ //更新纪录
+ protected int updateEntityDB(final EntityInfo info, T... entitys) {
+ return updateEntityDBAsync(info, entitys).join();
+ }
+
+ //更新纪录
+ protected int updateColumnDB(final EntityInfo info, Flipper flipper, final UpdateSqlInfo sql) {
+ return updateColumnDBAsync(info, flipper, sql).join();
+ }
+
+ //查询Number Map数据
+ protected Map getNumberMapDB(final EntityInfo info, String[] tables, final String sql, final FilterNode node, final FilterFuncColumn... columns) {
+ return (Map) getNumberMapDBAsync(info, tables, sql, node, columns).join();
+ }
+
+ //查询Number数据
+ protected Number getNumberResultDB(final EntityInfo info, String[] tables, final String sql, final FilterFunc func, final Number defVal, final String column, final FilterNode node) {
+ return getNumberResultDBAsync(info, tables, sql, func, defVal, column, node).join();
+ }
+
+ //查询Map数据
+ protected Map queryColumnMapDB(final EntityInfo info, String[] tables, final String sql, final String keyColumn, final FilterFunc func, final String funcColumn, FilterNode node) {
+ return (Map) queryColumnMapDBAsync(info, tables, sql, keyColumn, func, funcColumn, node).join();
+ }
+
+ //查询Map数据
+ protected Map queryColumnMapDB(final EntityInfo info, String[] tables, final String sql, final ColumnNode[] funcNodes, final String[] groupByColumns, final FilterNode node) {
+ return (Map) queryColumnMapDBAsync(info, tables, sql, funcNodes, groupByColumns, node).join();
+ }
+
+ //查询单条记录
+ protected T findDB(final EntityInfo info, String[] tables, final String sql, final boolean onlypk, final SelectColumn selects, final Serializable pk, final FilterNode node) {
+ return findDBAsync(info, tables, sql, onlypk, selects, pk, node).join();
+ }
+
+ //查询单条记录的单个字段
+ protected Serializable findColumnDB(final EntityInfo info, String[] tables, final String sql, final boolean onlypk, final String column, final Serializable defValue, final Serializable pk, final FilterNode node) {
+ return findColumnDBAsync(info, tables, sql, onlypk, column, defValue, pk, node).join();
+ }
+
+ //判断记录是否存在
+ protected boolean existsDB(final EntityInfo info, final String[] tables, final String sql, final boolean onlypk, final Serializable pk, final FilterNode node) {
+ return existsDBAsync(info, tables, sql, onlypk, pk, node).join();
+ }
+
+ //查询一页数据
+ protected Sheet querySheetDB(final EntityInfo info, final boolean readcache, final boolean needtotal, final boolean distinct, final SelectColumn selects, final Flipper flipper, final FilterNode node) {
+ return querySheetDBAsync(info, readcache, needtotal, distinct, selects, flipper, node).join();
+ }
+
+ protected CharSequence createSQLJoin(FilterNode node, final Function func, final boolean update, final Map joinTabalis, final Set haset, final EntityInfo info) {
+ return node == null ? null : node.createSQLJoin(func, update, joinTabalis, haset, info);
+ }
+
+ protected CharSequence createSQLExpress(FilterNode node, final EntityInfo info, final Map joinTabalis) {
+ return node == null ? null : node.createSQLExpress(this, info, joinTabalis);
+ }
+
+ @Local
+ @Override
+ public String getType() {
+ return "sql";
+ }
+
+ @Override
+ public final String resourceName() {
+ return name;
+ }
+
+ @Local
+ @Override
+ public EntityInfo apply(Class t) {
+ return loadEntityInfo(t);
+ }
+
+ @Local
+ @Override
+ public void close() throws Exception {
+ }
+
+ protected EntityInfo loadEntityInfo(Class clazz) {
+ EntityInfo info = this.currEntityInfo;
+ if (info != null && info.getType() == clazz) {
+ return info;
+ }
+ info = loadEntityInfo(clazz, this.cacheForbidden, readConfProps, fullloader);
+ this.currEntityInfo = info;
+ return info;
+ }
+
+ public EntityCache loadCache(Class clazz) {
+ EntityInfo info = loadEntityInfo(clazz);
+ return info.getCache();
+ }
+
+ /**
+ * 将entity的对象全部加载到Cache中去，如果clazz没有被@org.redkale.persistence.Cacheable注解则不做任何事
+ *
+ * @param Entity类泛型
+ * @param clazz Entity类
+ */
+ public void refreshCache(Class clazz) {
+ EntityInfo info = loadEntityInfo(clazz);
+ EntityCache cache = info.getCache();
+ if (cache == null) {
+ return;
+ }
+ cache.fullLoadAsync();
+ }
+
+ protected CharSequence formatValueToString(final EntityInfo info, Object value) {
+ if (value == null) {
+ return null;
+ }
+ if ("mysql".equals(dbtype)) {
+ if (value instanceof CharSequence) {
+ return new StringBuilder().append('\'').append(value.toString().replace("\\", "\\\\").replace("'", "\\'")).append('\'').toString();
+ } else if (!(value instanceof Number) && !(value instanceof java.util.Date)
+ && !value.getClass().getName().startsWith("java.sql.") && !value.getClass().getName().startsWith("java.time.")) {
+ return new StringBuilder().append('\'').append(info.getJsonConvert().convertTo(value).replace("\\", "\\\\").replace("'", "\\'")).append('\'').toString();
+ }
+ return String.valueOf(value);
+ } else if (value != null && value instanceof CharSequence && "postgresql".equals(dbtype)) {
+ String s = String.valueOf(value);
+ int pos = s.indexOf('\'');
+ if (pos >= 0) {
+ return new StringBuilder().append("E'").append(value.toString().replace("\\", "\\\\").replace("'", "\\'")).append('\'').toString();
+ }
+ }
+ return info.formatSQLValue(value, null);
+ }
+
+ //----------------------------- insert -----------------------------
+ /**
+ * 新增对象， 必须是Entity对象
+ *
+ * @param Entity类泛型
+ * @param entitys Entity对象
+ *
+ * @return 影响的记录条数
+ */
+ @Override
+ public int insert(T... entitys) {
+ if (entitys.length == 0) {
+ return 0;
+ }
+ checkEntity("insert", entitys);
+ final EntityInfo info = loadEntityInfo((Class) entitys[0].getClass());
+ if (isOnlyCache(info)) {
+ return insertCache(info, entitys);
+ }
+ if (isAsync()) {
+ int rs = insertDBAsync(info, entitys).join();
+ insertCache(info, entitys);
+ return rs;
+ } else {
+ int rs = insertDB(info, entitys);
+ insertCache(info, entitys);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture insertAsync(T... entitys) {
+ if (entitys.length == 0) {
+ return CompletableFuture.completedFuture(0);
+ }
+ checkEntity("insert", entitys);
+ final EntityInfo info = loadEntityInfo((Class) entitys[0].getClass());
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(insertCache(info, entitys));
+ }
+ if (isAsync()) {
+ return insertDBAsync(info, entitys).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ insertCache(info, entitys);
+ }
+ });
+ } else {
+ return supplyAsync(() -> insertDB(info, entitys)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ insertCache(info, entitys);
+ }
+ });
+ }
+ }
+
+ protected int insertCache(final EntityInfo info, T... entitys) {
+ final EntityCache cache = info.getCache();
+ if (cache == null) {
+ return 0;
+ }
+ int c = 0;
+ for (final T value : entitys) {
+ c += cache.insert(value);
+ }
+ return c;
+ }
+
+ //----------------------------- deleteCompose -----------------------------
+ /**
+ * 删除对象， 必须是Entity对象
+ *
+ * @param Entity类泛型
+ * @param entitys Entity对象
+ *
+ * @return 删除的数据条数
+ */
+ @Override
+ public int delete(T... entitys) {
+ if (entitys.length == 0) {
+ return -1;
+ }
+ checkEntity("delete", entitys);
+ final Class clazz = (Class) entitys[0].getClass();
+ final EntityInfo info = loadEntityInfo(clazz);
+ final Attribute primary = info.getPrimary();
+ Serializable[] ids = new Serializable[entitys.length];
+ int i = 0;
+ for (final T value : entitys) {
+ ids[i++] = (Serializable) primary.get(value);
+ }
+ return delete(clazz, ids);
+ }
+
+ @Override
+ public CompletableFuture deleteAsync(final T... entitys) {
+ if (entitys.length == 0) {
+ return CompletableFuture.completedFuture(-1);
+ }
+ checkEntity("delete", entitys);
+ final Class clazz = (Class) entitys[0].getClass();
+ final EntityInfo info = loadEntityInfo(clazz);
+ final Attribute primary = info.getPrimary();
+ Serializable[] ids = new Serializable[entitys.length];
+ int i = 0;
+ for (final T value : entitys) {
+ ids[i++] = (Serializable) primary.get(value);
+ }
+ return deleteAsync(clazz, ids);
+ }
+
+ @Override
+ public int delete(Class clazz, Serializable... pks) {
+ if (pks.length == 0) {
+ return -1;
+ }
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return deleteCache(info, -1, pks);
+ }
+ Map> pkmap = info.getTableMap(pks);
+ String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
+ String[] sqls = deleteSql(info, pkmap);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ int rs = deleteDBAsync(info, tables, null, null, pkmap, sqls).join();
+ deleteCache(info, rs, pks);
+ return rs;
+ } else {
+ int rs = deleteDB(info, tables, null, null, pkmap, sqls);
+ deleteCache(info, rs, pks);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture deleteAsync(final Class clazz, final Serializable... pks) {
+ if (pks.length == 0) {
+ return CompletableFuture.completedFuture(-1);
+ }
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(deleteCache(info, -1, pks));
+ }
+ Map> pkmap = info.getTableMap(pks);
+ String[] tables = pkmap.keySet().toArray(new String[pkmap.size()]);
+ String[] sqls = deleteSql(info, pkmap);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ return deleteDBAsync(info, tables, null, null, pkmap, sqls).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ deleteCache(info, rs, pks);
+ }
+ });
+ } else {
+ return supplyAsync(() -> deleteDB(info, tables, null, null, pkmap, sqls)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ deleteCache(info, rs, pks);
+ }
+ });
+ }
+ }
+
+ @Override
+ public int delete(Class clazz, FilterNode node) {
+ return delete(clazz, (Flipper) null, node);
+ }
+
+ @Override
+ public CompletableFuture deleteAsync(final Class clazz, final FilterNode node) {
+ return deleteAsync(clazz, (Flipper) null, node);
+ }
+
+ @Override
+ public int delete(Class clazz, final Flipper flipper, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return deleteCache(info, -1, flipper, node);
+ }
+ String[] tables = info.getTables(node);
+ String[] sqls = deleteSql(info, tables, flipper, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ int rs = deleteDBAsync(info, tables, flipper, node, null, sqls).join();
+ deleteCache(info, rs, flipper, sqls);
+ return rs;
+ } else {
+ int rs = deleteDB(info, tables, flipper, node, null, sqls);
+ deleteCache(info, rs, flipper, sqls);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture deleteAsync(final Class clazz, final Flipper flipper, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(deleteCache(info, -1, flipper, node));
+ }
+ String[] tables = info.getTables(node);
+ String[] sqls = deleteSql(info, tables, flipper, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " delete sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ return deleteDBAsync(info, tables, flipper, node, null, sqls).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ deleteCache(info, rs, flipper, node);
+ }
+ });
+ } else {
+ return supplyAsync(() -> deleteDB(info, tables, flipper, node, null, sqls)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ deleteCache(info, rs, flipper, node);
+ }
+ });
+ }
+ }
+
+ protected String[] deleteSql(final EntityInfo info, final Map> pkmap) {
+ List sqls = new ArrayList<>();
+ final String pkSQLColumn = info.getPrimarySQLColumn();
+ pkmap.forEach((table, pks) -> {
+ String sql;
+ if (pks.size() == 1) {
+ sql = "DELETE FROM " + table + " WHERE " + pkSQLColumn + " = " + info.formatSQLValue(pkSQLColumn, pks.get(0), sqlFormatter);
+ } else {
+ 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 += ")";
+ }
+ sqls.add(sql);
+ });
+ return sqls.toArray(new String[sqls.size()]);
+ }
+
+ protected String[] deleteSql(final EntityInfo info, String[] tables, final Flipper flipper, final FilterNode node) {
+ Map joinTabalis = null;
+ CharSequence join = null;
+ CharSequence where = null;
+ if (node != null) {
+ joinTabalis = node.getJoinTabalis();
+ join = node.createSQLJoin(this, true, joinTabalis, new HashSet<>(), info);
+ where = node.createSQLExpress(this, info, joinTabalis);
+ }
+ StringBuilder join1 = null;
+ StringBuilder join2 = null;
+ if (join != null) {
+ String joinstr = join.toString();
+ join1 = multiSplit('[', ']', ",", 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 ("postgresql".equals(dbtype()) && flipper != null && flipper.getLimit() > 0) {
+ List sqls = new ArrayList<>();
+ for (String table : tables) {
+ String sql = "DELETE FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) + " WHERE " + info.getPrimarySQLColumn() + " IN (SELECT " + info.getPrimaryField() + " FROM " + table
+ + join2AndWhere + info.createSQLOrderby(flipper) + " OFFSET 0 LIMIT " + flipper.getLimit() + ")";
+ sqls.add(sql);
+ }
+ return sqls.toArray(new String[sqls.size()]);
+ } else {
+ boolean mysql = "mysql".equals(dbtype());
+ List sqls = new ArrayList<>();
+ for (String table : tables) {
+ String sql = "DELETE " + (mysql ? "a" : "") + " FROM " + table + " a" + (join1 == null ? "" : (", " + join1)) + join2AndWhere + info.createSQLOrderby(flipper)
+ + ((mysql && flipper != null && flipper.getLimit() > 0) ? (" LIMIT " + flipper.getLimit()) : "");
+ sqls.add(sql);
+ }
+ return sqls.toArray(new String[sqls.size()]);
+ }
+ }
+
+ //----------------------------- clearTableCompose -----------------------------
+ @Override
+ public int clearTable(Class clazz) {
+ return clearTable(clazz, (FilterNode) null);
+ }
+
+ @Override
+ public int clearTable(Class clazz, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return clearTableCache(info, node);
+ }
+ final String[] tables = info.getTables(node);
+ String[] sqls = clearTableSql(info, tables, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ int rs = clearTableDBAsync(info, tables, node, sqls).join();
+ clearTableCache(info, node);
+ return rs;
+ } else {
+ int rs = clearTableDB(info, tables, node, sqls);
+ clearTableCache(info, node);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture clearTableAsync(final Class clazz, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(clearTableCache(info, node));
+ }
+ final String[] tables = info.getTables(node);
+ String[] sqls = clearTableSql(info, tables, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " clearTable sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ return clearTableDBAsync(info, tables, node, sqls).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ clearTableCache(info, node);
+ }
+ });
+ } else {
+ return supplyAsync(() -> clearTableDB(info, tables, node, sqls)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ clearTableCache(info, node);
+ }
+ });
+ }
+ }
+
+ protected String[] clearTableSql(final EntityInfo info, String[] tables, final FilterNode node) {
+ List sqls = new ArrayList<>();
+ for (String table : tables) {
+ sqls.add("TRUNCATE TABLE " + table);
+ }
+ return sqls.toArray(new String[sqls.size()]);
+ }
+
+ //----------------------------- dropTable -----------------------------
+ @Override
+ public int createTable(final Class clazz, final Serializable pk) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ final String[] sqls = createTableSqls(info);
+ if (sqls == null) {
+ return -1;
+ }
+ String copyTableSql = info.getTableStrategy() == null ? null : getTableCopySQL(info, info.getTable(pk));
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " createTable sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ int rs = createTableDBAsync(info, copyTableSql, pk, sqls).join();
+ return rs;
+ } else {
+ int rs = createTableDB(info, copyTableSql, pk, sqls);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture createTableAsync(final Class clazz, final Serializable pk) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ final String[] sqls = createTableSqls(info);
+ if (sqls == null) {
+ return CompletableFuture.completedFuture(-1);
+ }
+ String copyTableSql = info.getTableStrategy() == null ? null : getTableCopySQL(info, info.getTable(pk));
+ if (copyTableSql == null) {
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " createTable sql=" + Arrays.toString(sqls));
+ }
+ } else {
+ if (info.isLoggable(logger, Level.FINEST, copyTableSql)) {
+ logger.finest(info.getType().getSimpleName() + " createTable sql=" + copyTableSql);
+ }
+ }
+ if (isAsync()) {
+ return createTableDBAsync(info, copyTableSql, pk, sqls).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ }
+ });
+ } else {
+ return supplyAsync(() -> createTableDB(info, copyTableSql, pk, sqls)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ }
+ });
+ }
+ }
+
+ //----------------------------- dropTable -----------------------------
+ @Override
+ public int dropTable(Class clazz, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return dropTableCache(info, node);
+ }
+ final String[] tables = info.getTables(node);
+ String[] sqls = dropTableSql(info, tables, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ int rs = dropTableDBAsync(info, tables, node, sqls).join();
+ dropTableCache(info, node);
+ return rs;
+ } else {
+ int rs = dropTableDB(info, tables, node, sqls);
+ dropTableCache(info, node);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture dropTableAsync(final Class clazz, FilterNode node) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(dropTableCache(info, node));
+ }
+ final String[] tables = info.getTables(node);
+ String[] sqls = dropTableSql(info, tables, node);
+ if (info.isLoggable(logger, Level.FINEST, sqls[0])) {
+ logger.finest(info.getType().getSimpleName() + " dropTable sql=" + Arrays.toString(sqls));
+ }
+ if (isAsync()) {
+ return dropTableDBAsync(info, tables, node, sqls).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ dropTableCache(info, node);
+ }
+ });
+ } else {
+ return supplyAsync(() -> dropTableDB(info, tables, node, sqls)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ dropTableCache(info, node);
+ }
+ });
+ }
+ }
+
+ protected String[] dropTableSql(final EntityInfo info, String[] tables, final FilterNode node) {
+ List sqls = new ArrayList<>();
+ for (String table : tables) {
+ sqls.add("DROP TABLE IF EXISTS " + table);
+ }
+ return sqls.toArray(new String[sqls.size()]);
+ }
+
+ protected int clearTableCache(final EntityInfo info, FilterNode node) {
+ final EntityCache cache = info.getCache();
+ if (cache == null) {
+ return -1;
+ }
+ return cache.clear();
+ }
+
+ protected int dropTableCache(final EntityInfo info, FilterNode node) {
+ final EntityCache cache = info.getCache();
+ if (cache == null) {
+ return -1;
+ }
+ return cache.drop();
+ }
+
+ protected int deleteCache(final EntityInfo info, int count, Flipper flipper, FilterNode node) {
+ final EntityCache cache = info.getCache();
+ if (cache == null) {
+ return -1;
+ }
+ Serializable[] ids = cache.delete(flipper, node);
+ return count >= 0 ? count : (ids == null ? 0 : ids.length);
+ }
+
+ protected int deleteCache(final EntityInfo info, int count, Serializable... pks) {
+ final EntityCache cache = info.getCache();
+ if (cache == null) {
+ return -1;
+ }
+ int c = 0;
+ for (Serializable key : pks) {
+ c += cache.delete(key);
+ }
+ return count >= 0 ? count : c;
+ }
+
+ protected static StringBuilder multiSplit(char ch1, char ch2, String split, StringBuilder sb, String str, int from) {
+ if (str == null) {
+ return sb;
+ }
+ int pos1 = str.indexOf(ch1, from);
+ if (pos1 < 0) {
+ return sb;
+ }
+ int pos2 = str.indexOf(ch2, from);
+ if (pos2 < 0) {
+ return sb;
+ }
+ if (sb.length() > 0) {
+ sb.append(split);
+ }
+ sb.append(str.substring(pos1 + 1, pos2));
+ return multiSplit(ch1, ch2, split, sb, str, pos2 + 1);
+ }
+
+ //---------------------------- update ----------------------------
+ /**
+ * 更新对象， 必须是Entity对象
+ *
+ * @param Entity类泛型
+ * @param entitys Entity对象
+ *
+ * @return 更新的数据条数
+ */
+ @Override
+ public int update(T... entitys) {
+ if (entitys.length == 0) {
+ return -1;
+ }
+ checkEntity("update", entitys);
+ final Class clazz = (Class) entitys[0].getClass();
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return updateCache(info, -1, entitys);
+ }
+ if (isAsync()) {
+ int rs = updateEntityDBAsync(info, entitys).join();
+ updateCache(info, rs, entitys);
+ return rs;
+ } else {
+ int rs = updateEntityDB(info, entitys);
+ updateCache(info, rs, entitys);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture updateAsync(final T... entitys) {
+ if (entitys.length == 0) {
+ return CompletableFuture.completedFuture(-1);
+ }
+ checkEntity("update", entitys);
+ final Class clazz = (Class) entitys[0].getClass();
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(updateCache(info, -1, entitys));
+ }
+ if (isAsync()) {
+ return updateEntityDBAsync(info, entitys).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ updateCache(info, rs, entitys);
+ }
+ });
+ } else {
+ return supplyAsync(() -> updateEntityDB(info, entitys)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ updateCache(info, rs, entitys);
+ }
+ });
+ }
+ }
+
+ /**
+ * 根据主键值更新对象的column对应的值， 必须是Entity Class
+ *
+ * @param Entity类的泛型
+ * @param clazz Entity类
+ * @param pk 主键值
+ * @param column 过滤字段名
+ * @param colval 过滤字段值
+ *
+ * @return 更新的数据条数
+ */
+ @Override
+ public int updateColumn(Class clazz, Serializable pk, String column, Serializable colval) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return updateCache(info, -1, pk, column, colval);
+ }
+
+ UpdateSqlInfo sql = updateColumnSql(info, pk, column, colval);
+ if (isAsync()) {
+ int rs = updateColumnDBAsync(info, null, sql).join();
+ updateCache(info, rs, pk, column, colval);
+ return rs;
+ } else {
+ int rs = updateColumnDB(info, null, sql);
+ updateCache(info, rs, pk, column, colval);
+ return rs;
+ }
+ }
+
+ @Override
+ public CompletableFuture updateColumnAsync(final Class clazz, final Serializable pk, final String column, final Serializable colval) {
+ final EntityInfo info = loadEntityInfo(clazz);
+ if (isOnlyCache(info)) {
+ return CompletableFuture.completedFuture(updateCache(info, -1, pk, column, colval));
+ }
+
+ UpdateSqlInfo sql = updateColumnSql(info, pk, column, colval);
+ if (isAsync()) {
+ return updateColumnDBAsync(info, null, sql).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ updateCache(info, rs, pk, column, colval);
+ }
+ });
+ } else {
+ return supplyAsync(() -> updateColumnDB(info, null, sql)).whenComplete((rs, t) -> {
+ if (t != null) {
+ errorCompleteConsumer.accept(rs, t);
+ } else {
+ updateCache(info, rs, pk, column, colval);
+ }
+ });
+ }
+ }
+
+ protected UpdateSqlInfo updateColumnSql(final EntityInfo 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 UpdateSqlInfo(true, "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 UpdateSqlInfo(false, "UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + "="
+ + info.formatSQLValue(column, val, sqlFormatter) + " WHERE " + info.getPrimarySQLColumn() + "=" + info.formatSQLValue(info.getPrimarySQLColumn(), pk, sqlFormatter));
+ }
+ }
+
+ /**
+ * 根据主键值更新对象的column对应的值， 必须是Entity Class
+ *
+ * @param Entity类的泛型
+ * @param clazz Entity类
+ * @param column 过滤字段名
+ * @param colval 过滤字段值
+ * @param node 过滤node 不能为null
+ *
+ * @return 更新的数据条数
+ */
+ @Override
+ public int updateColumn(Class clazz, String column, Serializable colval, FilterNode node) {
+ final EntityInfo