EasyFlow-Tech-Dev/EasyFlow
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: 收敛工作流状态与高 IO 节点开销

Browse Source 
- 落地 Redis 版本状态、触发租约和定义缓存

- 优化数据批写、插件请求、文件下载与审计日志

- 补齐循环范围校验、轮询兼容和专项测试
This commit is contained in:
陈子默
2026-07-29 00:47:48 +08:00
parent 5ee6065017
commit 1ae8a22afe
103 changed files with 15600 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
easyflow-modules/easyflow-module-datacenter/pom.xml
View File

@@ -45,6 +45,18 @@
<groupId>tech.easyflow</groupId>
<artifactId>easyflow-common-web</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>5.12.0</version>
<scope>test</scope>
</dependency>
</dependencies>
<profiles>

36
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/QueryExecutor.java
View File

@@ -7,9 +7,45 @@ import tech.easyflow.datacenter.execution.model.DatacenterQueryRequest;
import tech.easyflow.datacenter.meta.entity.DatacenterSource;
import java.util.List;
import java.util.function.Consumer;
public interface QueryExecutor {
/**
* 分页查询结构化数据集。
*
* @param source 数据源
* @param table 数据表
* @param request 查询请求
* @return 分页结果
*/
Page<Row> queryPage(DatacenterSource source, DatacenterTable table, DatacenterQueryRequest request);
/**
* 执行原生 SQL 并返回完整结果。
*
* @param source 数据源
* @param sql 已校验 SQL
* @return 完整结果
*/
List<Row> queryBySql(DatacenterSource source, String sql);
/**
* 在单次查询中按结果顺序消费原生 SQL 返回行。
*
* <p>缺省实现保持第三方连接器兼容；JDBC 连接器应覆盖此方法并使用单连接、
* 单 ResultSet 流式读取。</p>
*
* @param source 数据源
* @param sql 已校验 SQL
* @param fetchSize JDBC 建议拉取行数
* @param consumer 单行消费者
*/
default void consumeBySql(
DatacenterSource source,
String sql,
int fetchSize,
Consumer<Row> consumer) {
queryBySql(source, sql).forEach(consumer);
}
}

50
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/WriteExecutor.java
View File

@@ -6,9 +6,59 @@ import tech.easyflow.datacenter.entity.DatacenterTable;
import tech.easyflow.datacenter.meta.entity.DatacenterSource;
import java.math.BigInteger;
import java.util.List;
public interface WriteExecutor {
void saveRow(DatacenterSource source, DatacenterTable table, JSONObject data, LoginAccount account);
/**
* 批量保存数据行。
* <p>
* 缺省实现保持逐行语义；支持批处理的连接器应覆盖此方法。
*
* @param source 数据源
* @param table 数据表
* @param rows 待保存数据行
* @param account 当前操作账号
* @param batchSize 单批最大行数
*/
default void saveRows(DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize) {
for (JSONObject row : rows) {
saveRow(source, table, row, account);
}
}
/**
* 在目标数据库中以唯一回执和业务写入同事务保存数据。
*
* <p>缺省实现用于不支持目标库事务回执的连接器，仍保持普通批量写入语义。支持写能力
* 的连接器应覆盖该方法。</p>
*
* @param source 数据源
* @param table 数据表
* @param rows 待保存行
* @param account 操作账号
* @param batchSize 单批最大行数
* @param receiptKey 有界幂等回执键
* @param payloadHash 输入负载摘要
* @return 本次实际写入时为 {@code true}，同负载回执已存在时为 {@code false}
*/
default boolean saveRowsIdempotently(
DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize,
String receiptKey,
String payloadHash) {
saveRows(source, table, rows, account, batchSize);
return true;
}
void deleteRow(DatacenterSource source, DatacenterTable table, BigInteger id, LoginAccount account);
}

5
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/impl/ExcelConnector.java
View File

@@ -5,11 +5,12 @@ import tech.easyflow.datacenter.connector.support.AbstractInternalTableConnector
import tech.easyflow.datacenter.meta.enums.DatacenterCapability;
import tech.easyflow.datacenter.meta.enums.DatacenterSourceType;
import javax.sql.DataSource;
import java.util.EnumSet;
@Component
public class ExcelConnector extends AbstractInternalTableConnector {
public ExcelConnector() {
public ExcelConnector(DataSource dataSource) {
super(DatacenterSourceType.EXCEL, EnumSet.of(
DatacenterCapability.TEST_CONNECTION,
DatacenterCapability.BROWSE_METADATA,
@@ -17,6 +18,6 @@ public class ExcelConnector extends AbstractInternalTableConnector {
DatacenterCapability.WRITE_MUTATION,
DatacenterCapability.MATERIALIZE,
DatacenterCapability.EXPORT
));
), dataSource);
}
}

6
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/impl/ExcelMaterializedConnector.java
View File

@@ -5,11 +5,13 @@ import tech.easyflow.datacenter.connector.support.AbstractInternalTableConnector
import tech.easyflow.datacenter.meta.enums.DatacenterCapability;
import tech.easyflow.datacenter.meta.enums.DatacenterSourceType;
import javax.sql.DataSource;
import java.util.EnumSet;
@Component
public class ExcelMaterializedConnector extends AbstractInternalTableConnector {
public ExcelMaterializedConnector() {
public ExcelMaterializedConnector(
DataSource dataSource) {
super(DatacenterSourceType.EXCEL_MATERIALIZED, EnumSet.of(
DatacenterCapability.TEST_CONNECTION,
DatacenterCapability.BROWSE_METADATA,
@@ -17,6 +19,6 @@ public class ExcelMaterializedConnector extends AbstractInternalTableConnector {
DatacenterCapability.WRITE_MUTATION,
DatacenterCapability.MATERIALIZE,
DatacenterCapability.EXPORT
));
), dataSource);
}
}

445
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/impl/ProjectMysqlConnector.java
View File

@@ -9,6 +9,7 @@ import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.datacenter.connector.dialect.MysqlSqlDialect;
import tech.easyflow.datacenter.connector.support.AbstractJdbcConnector;
import tech.easyflow.datacenter.connector.support.WriteReceiptSupport;
import tech.easyflow.datacenter.entity.DatacenterTable;
import tech.easyflow.datacenter.entity.DatacenterTableField;
import tech.easyflow.datacenter.execution.model.DatacenterQueryRequest;
@@ -20,9 +21,15 @@ import javax.sql.DataSource;
import java.math.BigInteger;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@Component
@@ -116,6 +123,444 @@ public class ProjectMysqlConnector extends AbstractJdbcConnector {
}
}
/**
* 使用单个连接按相同 SQL 连续分组批量保存数据。
*
* @param source 数据源
* @param table 数据表
* @param rows 待保存数据行
* @param account 当前操作账号
* @param batchSize 单批最大行数
*/
@Override
public void saveRows(DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize) {
if (rows == null || rows.isEmpty()) {
return;
}
try (Connection connection = dataSource.getConnection()) {
saveRows(connection, source, table, rows, batchSize);
} catch (Exception ex) {
throw new BusinessException("项目 MySQL 批量写入失败: " + ex.getMessage());
}
}
/**
* {@inheritDoc}
*/
@Override
public boolean saveRowsIdempotently(
DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize,
String receiptKey,
String payloadHash) {
if (StrUtil.isBlank(receiptKey)) {
saveRows(source, table, rows, account, batchSize);
return true;
}
try (Connection connection = dataSource.getConnection()) {
boolean originalAutoCommit = connection.getAutoCommit();
connection.setAutoCommit(false);
try {
if (hasMatchingReceipt(connection, receiptKey, payloadHash)) {
connection.rollback();
return false;
}
int effectiveBatchSize = Math.max(1, batchSize);
for (int offset = 0; offset < rows.size(); offset += effectiveBatchSize) {
int end = Math.min(rows.size(), offset + effectiveBatchSize);
List<PendingRow> batch = pendingRows(
connection,
rows,
offset,
end,
receiptKey,
payloadHash);
if (batch.isEmpty()) {
connection.rollback();
continue;
}
try {
insertReceiptsBatch(connection, batch, payloadHash);
saveRows(
connection,
source,
table,
batch.stream()
.map(PendingRow::row)
.collect(Collectors.toList()),
effectiveBatchSize);
connection.commit();
} catch (Exception batchError) {
connection.rollback();
/*
* 批失败才逐行回放，精确保留旧实现“失败前行已提交、失败后不再执行”
* 的可观察语义，同时让正常路径按 batchSize 真正批量提交。
*/
replayRowsIndividually(
connection,
source,
table,
batch,
payloadHash);
}
}
if (!insertReceipt(connection, receiptKey, payloadHash)) {
connection.rollback();
return false;
}
connection.commit();
return true;
} catch (Exception error) {
connection.rollback();
throw error;
} finally {
connection.setAutoCommit(originalAutoCommit);
}
} catch (Exception error) {
throw new BusinessException("项目 MySQL 幂等批量写入失败: " + error.getMessage());
}
}
/**
* 批量读取子回执并筛出尚未写入的行。
*
* @param connection JDBC 连接
* @param rows 全部数据行
* @param startInclusive 批起始下标
* @param endExclusive 批结束下标
* @param receiptKey 根回执键
* @param payloadHash 负载摘要
* @return 尚未提交的行
* @throws SQLException 查询失败或回执负载冲突
*/
private List<PendingRow> pendingRows(
Connection connection,
List<JSONObject> rows,
int startInclusive,
int endExclusive,
String receiptKey,
String payloadHash) throws SQLException {
List<PendingRow> candidates =
new ArrayList<>(endExclusive - startInclusive);
for (int rowIndex = startInclusive;
rowIndex < endExclusive;
rowIndex++) {
candidates.add(new PendingRow(
WriteReceiptSupport.childKey(receiptKey, rowIndex),
rows.get(rowIndex)));
}
String placeholders = candidates.stream()
.map(candidate -> "?")
.collect(Collectors.joining(","));
String sql = "SELECT idempotency_key, payload_hash "
+ "FROM tb_datacenter_write_receipt "
+ "WHERE idempotency_key IN (" + placeholders + ")";
Map<String, String> existing = new HashMap<>();
try (PreparedStatement statement = connection.prepareStatement(sql)) {
for (int index = 0; index < candidates.size(); index++) {
statement.setString(index + 1, candidates.get(index).receiptKey());
}
try (ResultSet resultSet = statement.executeQuery()) {
while (resultSet.next()) {
existing.put(resultSet.getString(1), resultSet.getString(2));
}
}
}
List<PendingRow> pending = new ArrayList<>(candidates.size());
for (PendingRow candidate : candidates) {
String existingHash = existing.get(candidate.receiptKey());
if (existingHash == null) {
pending.add(candidate);
} else if (!java.util.Objects.equals(payloadHash, existingHash)) {
throw new SQLException(
"相同幂等键对应的数据内容不一致", "23000");
}
}
return pending;
}
/**
* 在当前事务中批量创建子回执。
*
* @param connection JDBC 连接
* @param rows 待写行
* @param payloadHash 负载摘要
* @throws SQLException 回执批写失败
*/
private void insertReceiptsBatch(
Connection connection,
List<PendingRow> rows,
String payloadHash) throws SQLException {
String sql = "INSERT INTO tb_datacenter_write_receipt "
+ "(idempotency_key, payload_hash, created) "
+ "VALUES (?, ?, CURRENT_TIMESTAMP)";
try (PreparedStatement statement = connection.prepareStatement(sql)) {
for (PendingRow row : rows) {
statement.setString(1, row.receiptKey());
statement.setString(2, payloadHash);
statement.addBatch();
}
statement.executeBatch();
}
}
/**
* 批失败后逐行回放，定位首个业务失败并保留旧部分成功边界。
*
* @param connection JDBC 连接
* @param source 数据源
* @param table 数据表
* @param rows 本批待写行
* @param payloadHash 负载摘要
* @throws Exception 首个真实行写入失败
*/
private void replayRowsIndividually(
Connection connection,
DatacenterSource source,
DatacenterTable table,
List<PendingRow> rows,
String payloadHash) throws Exception {
for (PendingRow pendingRow : rows) {
if (!insertReceipt(
connection,
pendingRow.receiptKey(),
payloadHash)) {
connection.rollback();
continue;
}
try {
saveRows(
connection,
source,
table,
List.of(pendingRow.row()),
1);
connection.commit();
} catch (Exception rowError) {
connection.rollback();
throw rowError;
}
}
}
/**
* 待写行及其稳定子回执键。
*
* @param receiptKey 子回执键
* @param row 数据行
*/
private record PendingRow(String receiptKey, JSONObject row) {
}
/**
* 检查整次写入回执是否已经存在并校验负载。
*
* @param connection JDBC 连接
* @param receiptKey 回执键
* @param payloadHash 负载摘要
* @return 相同负载的回执存在时为 {@code true}
* @throws SQLException 数据库访问失败或负载冲突
*/
private boolean hasMatchingReceipt(
Connection connection, String receiptKey, String payloadHash)
throws SQLException {
String querySql = "SELECT payload_hash FROM tb_datacenter_write_receipt "
+ "WHERE idempotency_key = ?";
try (PreparedStatement statement = connection.prepareStatement(querySql)) {
statement.setString(1, receiptKey);
try (ResultSet resultSet = statement.executeQuery()) {
if (!resultSet.next()) {
return false;
}
if (!java.util.Objects.equals(payloadHash, resultSet.getString(1))) {
throw new SQLException(
"相同幂等键对应的数据内容不一致", "23000");
}
return true;
}
}
}
/**
* 在当前连接中写入唯一回执。
*
* @param connection JDBC 连接
* @param receiptKey 回执键
* @param payloadHash 负载摘要
* @return 新建回执时为 {@code true}，相同负载回执已存在时为 {@code false}
* @throws SQLException 数据库访问失败或负载摘要冲突
*/
private boolean insertReceipt(
Connection connection, String receiptKey, String payloadHash) throws SQLException {
String insertSql = "INSERT INTO tb_datacenter_write_receipt "
+ "(idempotency_key, payload_hash, created) VALUES (?, ?, CURRENT_TIMESTAMP)";
try (PreparedStatement statement = connection.prepareStatement(insertSql)) {
statement.setString(1, receiptKey);
statement.setString(2, payloadHash);
statement.executeUpdate();
return true;
} catch (SQLException duplicate) {
if (!"23000".equals(duplicate.getSQLState()) && duplicate.getErrorCode() != 1062) {
throw duplicate;
}
String querySql = "SELECT payload_hash FROM tb_datacenter_write_receipt "
+ "WHERE idempotency_key = ?";
try (PreparedStatement statement = connection.prepareStatement(querySql)) {
statement.setString(1, receiptKey);
try (ResultSet resultSet = statement.executeQuery()) {
if (!resultSet.next()) {
throw duplicate;
}
if (!java.util.Objects.equals(payloadHash, resultSet.getString(1))) {
throw new SQLException("相同幂等键对应的数据内容不一致", "23000");
}
return false;
}
}
}
}
/**
* 在给定连接上执行完整批量，供普通和事务幂等写入复用。
*
* @param connection JDBC 连接
* @param source 数据源
* @param table 数据表
* @param rows 待保存行
* @param batchSize 单批最大行数
* @throws Exception JDBC 批处理失败
*/
private void saveRows(
Connection connection,
DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
int batchSize) throws Exception {
List<DatacenterTableField> writableFields = table.getFields().stream()
.filter(field -> field.getWritable() == null || field.getWritable() == 1)
.collect(Collectors.toList());
int effectiveBatchSize = Math.max(1, batchSize);
List<SqlMutation> batch = new ArrayList<>(Math.min(rows.size(), effectiveBatchSize));
String batchSql = null;
for (JSONObject row : rows) {
SqlMutation mutation = buildMutation(source, table, writableFields, row);
if (mutation == null) {
continue;
}
if (batchSql != null
&& (!batchSql.equals(mutation.sql) || batch.size() >= effectiveBatchSize)) {
executeBatch(connection, batchSql, batch);
batch.clear();
}
batchSql = mutation.sql;
batch.add(mutation);
}
if (!batch.isEmpty()) {
executeBatch(connection, batchSql, batch);
}
}
/**
* 构建单行参数化写入。
*
* @param source 数据源
* @param table 数据表
* @param writableFields 可写字段
* @param data 数据行
* @return SQL 与参数；无可更新字段时返回 null
*/
private SqlMutation buildMutation(DatacenterSource source,
DatacenterTable table,
List<DatacenterTableField> writableFields,
JSONObject data) {
Object id = data.get("id");
if (id == null) {
List<String> columns = new ArrayList<>();
List<Object> values = new ArrayList<>();
for (DatacenterTableField field : writableFields) {
Object value = data.get(field.getFieldName());
if (value != null) {
columns.add(field.getFieldName());
values.add(value);
}
}
if (columns.isEmpty()) {
throw new BusinessException("没有可写字段");
}
String sql = "INSERT INTO "
+ dialect.qualifyTable(source.getDatabaseName(), resolvePhysicalTableName(table))
+ " (" + columns.stream().map(dialect::quoteIdentifier).collect(Collectors.joining(","))
+ ") VALUES (" + columns.stream().map(item -> "?").collect(Collectors.joining(",")) + ")";
return new SqlMutation(sql, values);
}
List<String> setClauses = new ArrayList<>();
List<Object> values = new ArrayList<>();
for (DatacenterTableField field : writableFields) {
if (!data.containsKey(field.getFieldName())) {
continue;
}
setClauses.add(dialect.quoteIdentifier(field.getFieldName()) + " = ?");
values.add(data.get(field.getFieldName()));
}
if (setClauses.isEmpty()) {
return null;
}
String sql = "UPDATE "
+ dialect.qualifyTable(source.getDatabaseName(), resolvePhysicalTableName(table))
+ " SET " + String.join(",", setClauses)
+ " WHERE " + dialect.quoteIdentifier("id") + " = ?";
values.add(id);
return new SqlMutation(sql, values);
}
/**
* 执行同构 SQL 批次。
*
* @param connection 数据库连接
* @param sql 参数化 SQL
* @param mutations 待执行参数
* @throws Exception JDBC 批处理失败时抛出
*/
private void executeBatch(Connection connection, String sql, List<SqlMutation> mutations) throws Exception {
try (PreparedStatement statement = connection.prepareStatement(sql)) {
for (SqlMutation mutation : mutations) {
for (int index = 0; index < mutation.parameters.size(); index++) {
statement.setObject(index + 1, mutation.parameters.get(index));
}
statement.addBatch();
}
statement.executeBatch();
}
}
/**
* 参数化写入描述。
*/
private static final class SqlMutation {
private final String sql;
private final List<Object> parameters;
/**
* 创建参数化写入。
*
* @param sql SQL 文本
* @param parameters SQL 参数
*/
private SqlMutation(String sql, List<Object> parameters) {
this.sql = sql;
this.parameters = parameters;
}
}
@Override
public void deleteRow(DatacenterSource source, DatacenterTable table, BigInteger id, LoginAccount account) {
String sql = "DELETE FROM " + dialect.qualifyTable(source.getDatabaseName(), resolvePhysicalTableName(table))

429
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/support/AbstractInternalTableConnector.java
View File

@@ -8,6 +8,7 @@ import com.mybatisflex.core.row.Db;
import com.mybatisflex.core.row.Row;
import com.mybatisflex.core.row.RowKey;
import org.springframework.util.CollectionUtils;
import org.springframework.dao.DuplicateKeyException;
import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.datacenter.connector.DatacenterConnector;
@@ -21,18 +22,30 @@ import tech.easyflow.datacenter.meta.enums.DatacenterSourceType;
import tech.easyflow.datacenter.meta.model.DatacenterCatalogMeta;
import tech.easyflow.datacenter.meta.model.DatacenterTableDetailMeta;
import javax.sql.DataSource;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.*;
import java.util.function.Consumer;
public abstract class AbstractInternalTableConnector implements DatacenterConnector {
private static final String WRITE_RECEIPT_TABLE = "tb_datacenter_write_receipt";
private final DatacenterSourceType sourceType;
private final Set<DatacenterCapability> capabilities;
private final DataSource dataSource;
protected AbstractInternalTableConnector(DatacenterSourceType sourceType, Set<DatacenterCapability> capabilities) {
protected AbstractInternalTableConnector(
DatacenterSourceType sourceType,
Set<DatacenterCapability> capabilities,
DataSource dataSource) {
this.sourceType = sourceType;
this.capabilities = capabilities;
this.dataSource = dataSource;
}
@Override
@@ -93,6 +106,58 @@ public abstract class AbstractInternalTableConnector implements DatacenterConnec
return rows;
}
/**
* {@inheritDoc}
*/
@Override
public void consumeBySql(
DatacenterSource source,
String sql,
int fetchSize,
Consumer<Row> consumer) {
if (fetchSize <= 0 || consumer == null) {
throw new IllegalArgumentException(
"fetchSize and consumer must be valid");
}
try (Connection connection =
dataSource.getConnection();
PreparedStatement statement =
connection.prepareStatement(
sql,
ResultSet.TYPE_FORWARD_ONLY,
ResultSet.CONCUR_READ_ONLY)) {
// 内部数据源使用项目 MySQL，启用驱动前向流式结果。
statement.setFetchSize(Integer.MIN_VALUE);
int timeoutSeconds = Integer.getInteger(
"easyflow.datacenter.query.timeout-seconds",
300);
if (timeoutSeconds > 0) {
statement.setQueryTimeout(timeoutSeconds);
}
try (ResultSet resultSet =
statement.executeQuery()) {
ResultSetMetaData metaData =
resultSet.getMetaData();
while (resultSet.next()) {
Row row = new Row();
for (int index = 1;
index <= metaData.getColumnCount();
index++) {
row.put(
metaData.getColumnLabel(index),
normalizeValue(
resultSet.getObject(index)));
}
consumer.accept(row);
}
}
} catch (Exception error) {
throw DatacenterConnectorExceptionSupport
.wrapAccessException(
"SQL 流式查询失败", error);
}
}
@Override
public void saveRow(DatacenterSource source, DatacenterTable table, JSONObject data, LoginAccount account) {
List<DatacenterTableField> fields = table.getFields();
@@ -100,21 +165,314 @@ public abstract class AbstractInternalTableConnector implements DatacenterConnec
throw new BusinessException("数据集字段为空，无法写入");
}
String actualTable = resolveTableName(table);
RowMutation mutation = buildRowMutation(fields, data, account);
if (mutation.insert) {
Db.insert(actualTable, mutation.row);
} else {
Db.updateById(actualTable, mutation.row);
}
}
/**
* 使用 MyBatis-Flex 动态表批处理保存数据行。
*
* @param source 数据源
* @param table 数据表
* @param rows 待保存数据行
* @param account 当前操作账号
* @param batchSize 单批最大行数
*/
@Override
public void saveRows(DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize) {
List<DatacenterTableField> fields = table.getFields();
if (CollectionUtils.isEmpty(fields)) {
throw new BusinessException("数据集字段为空，无法写入");
}
if (rows == null || rows.isEmpty()) {
return;
}
String actualTable = resolveTableName(table);
int effectiveBatchSize = Math.max(1, batchSize);
List<Row> batch = new ArrayList<>(Math.min(rows.size(), effectiveBatchSize));
Boolean insertBatch = null;
for (JSONObject data : rows) {
RowMutation mutation = buildRowMutation(fields, data, account);
if (insertBatch != null
&& (insertBatch != mutation.insert || batch.size() >= effectiveBatchSize)) {
executeRowBatch(actualTable, batch, insertBatch);
batch.clear();
}
insertBatch = mutation.insert;
batch.add(mutation.row);
}
if (!batch.isEmpty()) {
executeRowBatch(actualTable, batch, Boolean.TRUE.equals(insertBatch));
}
}
/**
* {@inheritDoc}
*/
@Override
public boolean saveRowsIdempotently(
DatacenterSource source,
DatacenterTable table,
List<JSONObject> rows,
LoginAccount account,
int batchSize,
String receiptKey,
String payloadHash) {
if (StrUtil.isBlank(receiptKey)) {
saveRows(source, table, rows, account, batchSize);
return true;
}
Row completedReceipt = Db.selectOneByMap(
WRITE_RECEIPT_TABLE,
Collections.singletonMap("idempotency_key", receiptKey));
if (completedReceipt != null) {
Object existingHash = completedReceipt.get("payload_hash");
if (!Objects.equals(payloadHash, existingHash)) {
throw new BusinessException("相同幂等键对应的数据内容不一致");
}
return false;
}
List<DatacenterTableField> fields = table.getFields();
if (CollectionUtils.isEmpty(fields)) {
throw new BusinessException("数据集字段为空，无法写入");
}
String actualTable = resolveTableName(table);
int effectiveBatchSize = Math.max(1, batchSize);
for (int offset = 0; offset < rows.size(); offset += effectiveBatchSize) {
int end = Math.min(rows.size(), offset + effectiveBatchSize);
List<PendingInternalRow> batch = new ArrayList<>(end - offset);
for (int index = offset; index < end; index++) {
batch.add(new PendingInternalRow(
WriteReceiptSupport.childKey(receiptKey, index),
rows.get(index),
buildRowMutation(
fields,
rows.get(index),
account)));
}
try {
Db.txWithResult(() -> {
insertReceiptsBatch(batch, payloadHash);
executePendingMutations(actualTable, batch);
return true;
});
} catch (RuntimeException batchError) {
/*
* 正常路径每批一个事务；批失败后才逐行回放，继续保持旧实现的
* 部分成功顺序边界，并利用子回执跳过已提交行。
*/
replayInternalRows(
actualTable,
fields,
batch,
account,
payloadHash);
}
}
return Db.txWithResult(() -> {
Row receipt = new Row();
receipt.put("idempotency_key", receiptKey);
receipt.put("payload_hash", payloadHash);
receipt.put("created", new Date());
try {
Db.insert(WRITE_RECEIPT_TABLE, receipt);
} catch (DuplicateKeyException duplicate) {
Row existing = Db.selectOneByMap(
WRITE_RECEIPT_TABLE,
Collections.singletonMap("idempotency_key", receiptKey));
Object existingHash = existing == null ? null : existing.get("payload_hash");
if (!Objects.equals(payloadHash, existingHash)) {
throw new BusinessException("相同幂等键对应的数据内容不一致");
}
return false;
}
return true;
});
}
/**
* 在当前 MyBatis-Flex 事务中写入单行动态表数据。
*
* @param actualTable 实际表名
* @param fields 可写字段
* @param data 原始行
* @param account 当前账号
*/
private void saveRowInCurrentTransaction(
String actualTable,
List<DatacenterTableField> fields,
JSONObject data,
LoginAccount account) {
RowMutation mutation = buildRowMutation(fields, data, account);
if (mutation.insert) {
Db.insert(actualTable, mutation.row);
} else {
Db.updateById(actualTable, mutation.row);
}
}
/**
* 创建写入回执。
*
* @param receiptKey 回执键
* @param payloadHash 负载摘要
*/
private void insertReceipt(String receiptKey, String payloadHash) {
Row receipt = new Row();
receipt.put("idempotency_key", receiptKey);
receipt.put("payload_hash", payloadHash);
receipt.put("created", new Date());
Db.insert(WRITE_RECEIPT_TABLE, receipt);
}
/**
* 在当前事务中批量创建子回执。
*
* @param rows 本批待写行
* @param payloadHash 负载摘要
*/
private void insertReceiptsBatch(
List<PendingInternalRow> rows,
String payloadHash) {
Date created = new Date();
List<Row> receipts = new ArrayList<>(rows.size());
for (PendingInternalRow pendingRow : rows) {
Row receipt = new Row();
receipt.put(
"idempotency_key",
pendingRow.receiptKey());
receipt.put("payload_hash", payloadHash);
receipt.put("created", created);
receipts.add(receipt);
}
Db.insertBatch(
WRITE_RECEIPT_TABLE,
receipts,
receipts.size());
}
/**
* 按原始顺序合并相邻同类型写入，减少正常路径 SQL 往返。
*
* @param actualTable 实际表名
* @param rows 本批待写行
*/
private void executePendingMutations(
String actualTable,
List<PendingInternalRow> rows) {
List<Row> batch = new ArrayList<>(rows.size());
Boolean insertBatch = null;
for (PendingInternalRow pendingRow : rows) {
RowMutation mutation = pendingRow.mutation();
if (insertBatch != null
&& insertBatch != mutation.insert) {
executeRowBatch(
actualTable,
batch,
insertBatch);
batch.clear();
}
insertBatch = mutation.insert;
batch.add(mutation.row);
}
if (!batch.isEmpty()) {
executeRowBatch(
actualTable,
batch,
Boolean.TRUE.equals(insertBatch));
}
}
/**
* 批失败后逐行回放，并校验重复回执的负载摘要。
*
* @param actualTable 实际表名
* @param fields 可写字段
* @param rows 本批数据
* @param account 当前账号
* @param payloadHash 负载摘要
*/
private void replayInternalRows(
String actualTable,
List<DatacenterTableField> fields,
List<PendingInternalRow> rows,
LoginAccount account,
String payloadHash) {
for (PendingInternalRow pendingRow : rows) {
Db.txWithResult(() -> {
try {
insertReceipt(pendingRow.receiptKey(), payloadHash);
} catch (DuplicateKeyException duplicate) {
Row existing = Db.selectOneByMap(
WRITE_RECEIPT_TABLE,
Collections.singletonMap(
"idempotency_key",
pendingRow.receiptKey()));
Object existingHash = existing == null
? null
: existing.get("payload_hash");
if (!Objects.equals(payloadHash, existingHash)) {
throw new BusinessException(
"相同幂等键对应的数据内容不一致");
}
return false;
}
saveRowInCurrentTransaction(
actualTable,
fields,
pendingRow.data(),
account);
return true;
});
}
}
/**
* 内部动态表待写行。
*
* @param receiptKey 子回执键
* @param data 原始行
* @param mutation 已构建的写入对象
*/
private record PendingInternalRow(
String receiptKey,
JSONObject data,
RowMutation mutation) {
}
/**
* 构建动态表单行写入对象。
*
* @param fields 数据表字段
* @param data 输入数据
* @param account 当前操作账号
* @return 行数据与写入类型
*/
private RowMutation buildRowMutation(
List<DatacenterTableField> fields, JSONObject data, LoginAccount account) {
Object id = data.get("id");
if (id == null) {
Date now = new Date();
Row row = Row.ofKey(RowKey.SNOW_FLAKE_ID);
row.put("dept_id", account.getDeptId());
row.put("tenant_id", account.getTenantId());
row.put("created", new Date());
row.put("created", now);
row.put("created_by", account.getId());
row.put("modified", new Date());
row.put("modified", now);
row.put("modified_by", account.getId());
row.put("remark", data.get("remark"));
for (DatacenterTableField field : fields) {
row.put(field.getFieldName(), data.get(field.getFieldName()));
}
Db.insert(actualTable, row);
return;
return new RowMutation(true, row);
}
Row row = Row.ofKey("id", id);
row.put("modified", new Date());
@@ -122,7 +480,42 @@ public abstract class AbstractInternalTableConnector implements DatacenterConnec
for (DatacenterTableField field : fields) {
row.put(field.getFieldName(), data.get(field.getFieldName()));
}
Db.updateById(actualTable, row);
return new RowMutation(false, row);
}
/**
* 执行动态表同类型批次。
*
* @param actualTable 实际表名
* @param rows 行数据
* @param insert 是否为新增批次
*/
private void executeRowBatch(String actualTable, List<Row> rows, boolean insert) {
if (insert) {
Db.insertBatch(actualTable, rows, rows.size());
} else {
Db.updateBatchById(actualTable, rows);
}
}
/**
* 动态表单行写入描述。
*/
private static final class RowMutation {
private final boolean insert;
private final Row row;
/**
* 创建动态表行写入描述。
*
* @param insert 是否新增
* @param row 行数据
*/
private RowMutation(boolean insert, Row row) {
this.insert = insert;
this.row = row;
}
}
@Override
@@ -138,15 +531,27 @@ public abstract class AbstractInternalTableConnector implements DatacenterConnec
for (Row record : records) {
Map<String, Object> converted = new LinkedHashMap<>();
for (Map.Entry<String, Object> entry : record.entrySet()) {
Object value = entry.getValue();
if (value instanceof BigInteger || value instanceof BigDecimal || value instanceof Long) {
converted.put(entry.getKey(), value.toString());
} else {
converted.put(entry.getKey(), value);
}
converted.put(
entry.getKey(),
normalizeValue(entry.getValue()));
}
record.clear();
record.putAll(converted);
}
}
/**
* 统一内部查询的数值 JSON 表现。
*
* @param value JDBC 原始值
* @return 兼容既有查询接口的值
*/
private Object normalizeValue(Object value) {
if (value instanceof BigInteger
|| value instanceof BigDecimal
|| value instanceof Long) {
return value.toString();
}
return value;
}
}

246
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/support/AbstractJdbcConnector.java
View File

@@ -28,6 +28,7 @@ import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.*;
import java.util.*;
import java.util.function.Consumer;
import java.util.stream.Collectors;
public abstract class AbstractJdbcConnector implements DatacenterConnector {
@@ -265,6 +266,43 @@ public abstract class AbstractJdbcConnector implements DatacenterConnector {
}
}
/**
* {@inheritDoc}
*/
@Override
public void consumeBySql(
DatacenterSource source,
String sql,
int fetchSize,
Consumer<Row> consumer) {
if (!capabilities.contains(
DatacenterCapability.READ_QUERY)) {
throw new BusinessException(
"当前数据源暂不支持查询");
}
if (fetchSize <= 0 || consumer == null) {
throw new IllegalArgumentException(
"fetchSize and consumer must be valid");
}
try {
withConnection(
source,
true,
connection -> {
consumeBySql(
connection,
sql,
fetchSize,
consumer);
return null;
});
} catch (Exception ex) {
throw DatacenterConnectorExceptionSupport
.wrapAccessException(
"SQL 流式查询失败", ex);
}
}
@Override
public void saveRow(DatacenterSource source, DatacenterTable table, JSONObject data, LoginAccount account) {
throw new BusinessException("当前数据源不支持写入");
@@ -330,9 +368,163 @@ public abstract class AbstractJdbcConnector implements DatacenterConnector {
}
protected List<Row> doQueryBySql(Connection connection, String sql) throws SQLException {
try (PreparedStatement statement = connection.prepareStatement(sql);
ResultSet resultSet = statement.executeQuery()) {
return readRows(resultSet);
try (PreparedStatement statement = connection.prepareStatement(sql)) {
configureStreamingQuery(statement);
try (ResultSet resultSet = statement.executeQuery()) {
return readRows(resultSet);
}
}
}
/**
* 使用单连接、单 ResultSet 顺序消费查询结果。
*
* @param connection JDBC 连接
* @param sql 已校验 SQL
* @param fetchSize JDBC 建议拉取行数
* @param consumer 单行消费者
* @throws SQLException 查询失败
*/
protected void consumeBySql(
Connection connection,
String sql,
int fetchSize,
Consumer<Row> consumer) throws SQLException {
boolean localCursorTransaction =
usesPostgresqlCursor()
&& connection.getAutoCommit();
if (localCursorTransaction) {
// PostgreSQL 协议仅在事务内按 fetchSize 使用服务端游标。
connection.setAutoCommit(false);
}
Throwable queryFailure = null;
try {
try (PreparedStatement statement =
connection.prepareStatement(
sql,
ResultSet.TYPE_FORWARD_ONLY,
ResultSet.CONCUR_READ_ONLY)) {
configureCursorQuery(statement, fetchSize);
try (ResultSet resultSet =
statement.executeQuery()) {
ResultSetMetaData metaData =
resultSet.getMetaData();
while (resultSet.next()) {
Row row = new Row();
for (int index = 1;
index <= metaData.getColumnCount();
index++) {
row.put(
metaData.getColumnLabel(index),
normalizeValue(
resultSet.getObject(index)));
}
consumer.accept(row);
}
}
}
} catch (SQLException | RuntimeException | Error ex) {
queryFailure = ex;
throw ex;
} finally {
if (localCursorTransaction) {
restoreCursorConnection(
connection, queryFailure);
}
}
}
/**
* 回滚只读游标事务并恢复连接池连接状态。
*
* @param connection JDBC 连接
* @param queryFailure 查询阶段异常；为空表示查询成功
* @throws SQLException 清理失败且查询本身成功
*/
private void restoreCursorConnection(
Connection connection,
Throwable queryFailure) throws SQLException {
SQLException cleanupFailure = null;
try {
connection.rollback();
} catch (SQLException ex) {
cleanupFailure = ex;
}
try {
connection.setAutoCommit(true);
} catch (SQLException ex) {
if (cleanupFailure == null) {
cleanupFailure = ex;
} else {
cleanupFailure.addSuppressed(ex);
}
}
if (cleanupFailure == null) {
return;
}
if (queryFailure != null) {
queryFailure.addSuppressed(cleanupFailure);
return;
}
throw cleanupFailure;
}
/**
* 判断当前连接器是否使用 PostgreSQL 游标协议。
*
* @return PostgreSQL 或 GaussDB 原生连接器返回 true
*/
private boolean usesPostgresqlCursor() {
return sourceType == DatacenterSourceType.POSTGRESQL
|| sourceType
== DatacenterSourceType.GAUSSDB_NATIVE;
}
/**
* 配置长结果游标的拉取策略和超时。
*
* @param statement JDBC 语句
* @param fetchSize 建议拉取行数
* @throws SQLException 配置失败
*/
private void configureCursorQuery(
PreparedStatement statement,
int fetchSize) throws SQLException {
if (sourceType == DatacenterSourceType.MYSQL
|| sourceType
== DatacenterSourceType.PROJECT_MYSQL
|| sourceType == DatacenterSourceType.GBASE_8A
|| sourceType == DatacenterSourceType.GBASE_8S) {
// MySQL 协议以该值启用前向只读流，避免驱动先缓存完整结果。
statement.setFetchSize(Integer.MIN_VALUE);
} else {
statement.setFetchSize(fetchSize);
}
int timeoutSeconds = Integer.getInteger(
"easyflow.datacenter.query.timeout-seconds",
300);
if (timeoutSeconds > 0) {
statement.setQueryTimeout(timeoutSeconds);
}
}
/**
* 为原生查询配置宽松但有限的流式拉取和超时。
*
* @param statement JDBC 语句
* @throws SQLException 配置失败
*/
private void configureStreamingQuery(
PreparedStatement statement) throws SQLException {
int fetchSize = Integer.getInteger(
"easyflow.datacenter.query.fetch-size", 1_000);
int timeoutSeconds = Integer.getInteger(
"easyflow.datacenter.query.timeout-seconds", 300);
if (fetchSize > 0) {
statement.setFetchSize(fetchSize);
}
if (timeoutSeconds > 0) {
statement.setQueryTimeout(timeoutSeconds);
}
}
@@ -349,17 +541,61 @@ public abstract class AbstractJdbcConnector implements DatacenterConnector {
protected List<Row> readRows(ResultSet resultSet) throws SQLException {
List<Row> records = new ArrayList<>();
ResultSetMetaData metaData = resultSet.getMetaData();
int maxRows = Integer.getInteger(
"easyflow.datacenter.query.max-rows", 1_000_000);
long maxBytes = Long.getLong(
"easyflow.datacenter.query.max-bytes",
512L * 1024L * 1024L);
long estimatedBytes = 0L;
while (resultSet.next()) {
if (maxRows > 0 && records.size() >= maxRows) {
throw new SQLException(
"数据集查询结果超过行数上限: " + maxRows);
}
Row row = new Row();
for (int i = 1; i <= metaData.getColumnCount(); i++) {
String columnLabel = metaData.getColumnLabel(i);
row.put(columnLabel, normalizeValue(resultSet.getObject(i)));
Object value = normalizeValue(resultSet.getObject(i));
row.put(columnLabel, value);
estimatedBytes += estimateQueryValueBytes(
columnLabel, value);
if (maxBytes > 0L && estimatedBytes > maxBytes) {
throw new SQLException(
"数据集查询结果超过字节上限: "
+ maxBytes);
}
}
records.add(row);
}
return records;
}
/**
* 估算查询结果在 JVM 中的最低占用，作为失控保护。
*
* @param columnLabel 列名
* @param value 列值
* @return 估算字节数
*/
private long estimateQueryValueBytes(
String columnLabel, Object value) {
long bytes = columnLabel == null
? 0L
: (long) columnLabel.length() * Character.BYTES;
if (value == null) {
return bytes + 8L;
}
if (value instanceof byte[]) {
return bytes + ((byte[]) value).length;
}
if (value instanceof CharSequence) {
return bytes
+ (long) value.toString().length()
* Character.BYTES;
}
return bytes + 64L;
}
protected String resolveCatalogArgument(DatacenterSource source, String catalogName) {
return usesCatalogNamespace() ? resolveCatalogName(source, catalogName) : source.getDatabaseName();
}
@@ -488,7 +724,7 @@ public abstract class AbstractJdbcConnector implements DatacenterConnector {
}
}
private Object normalizeValue(Object value) {
protected Object normalizeValue(Object value) {
if (value instanceof BigDecimal || value instanceof BigInteger || value instanceof Long) {
return value.toString();
}

33
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/connector/support/WriteReceiptSupport.java Normal file
View File

@@ -0,0 +1,33 @@
package tech.easyflow.datacenter.connector.support;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HexFormat;
/**
* 数据集写入回执键工具。
*/
public final class WriteReceiptSupport {
private WriteReceiptSupport() {
}
/**
* 为一行数据派生固定长度的稳定回执键。
*
* @param operationReceiptKey 整次写入的回执键
* @param rowIndex 行序号
* @return SHA-256 行回执键
*/
public static String childKey(String operationReceiptKey, int rowIndex) {
String value = operationReceiptKey + ':' + rowIndex;
try {
return HexFormat.of().formatHex(
MessageDigest.getInstance("SHA-256").digest(
value.getBytes(StandardCharsets.UTF_8)));
} catch (NoSuchAlgorithmException error) {
throw new IllegalStateException("SHA-256 is unavailable", error);
}
}
}

4
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/execution/model/DatasetRef.java
View File

@@ -2,7 +2,9 @@ package tech.easyflow.datacenter.execution.model;
import java.math.BigInteger;
public class DatasetRef {
public class DatasetRef implements java.io.Serializable {
private static final long serialVersionUID = 1L;
private BigInteger sourceId;
private BigInteger catalogId;
private String catalogName;

40
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/execution/service/DatacenterDatasetQueryService.java
View File

@@ -8,11 +8,51 @@ import tech.easyflow.datacenter.execution.model.DatacenterSqlQueryRequest;
import tech.easyflow.datacenter.execution.model.DatasetRef;
import java.util.List;
import java.util.function.Consumer;
public interface DatacenterDatasetQueryService {
/**
* 分页查询结构化数据集。
*
* @param request 查询请求
* @return 分页结果
*/
Page<Row> queryPage(DatacenterQueryRequest request);
/**
* 执行原生 SQL 并返回完整结果。
*
* @param request SQL 查询请求
* @return 完整结果
*/
List<Row> queryBySql(DatacenterSqlQueryRequest request);
/**
* 使用单次数据库查询流式消费原生 SQL 结果。
*
* @param request SQL 查询请求
* @param fetchSize JDBC 建议拉取行数
* @param consumer 单行消费者
*/
void consumeBySql(
DatacenterSqlQueryRequest request,
int fetchSize,
Consumer<Row> consumer);
/**
* 获取数据集结构。
*
* @param datasetRef 数据集引用
* @return 数据集结构
*/
DatacenterSchemaResponse getSchema(DatasetRef datasetRef);
/**
* 仅解析数据集定位信息，不加载版本和血缘。
*
* @param datasetRef 数据集引用
* @return 包含 source、catalog、table 的轻量响应
*/
DatacenterSchemaResponse getLocation(DatasetRef datasetRef);
}

45
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/execution/service/DatacenterDatasetWriteService.java
View File

@@ -5,9 +5,54 @@ import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.datacenter.execution.model.DatasetRef;
import java.math.BigInteger;
import java.util.List;
public interface DatacenterDatasetWriteService {
/**
* 保存单行数据。
*
* @param datasetRef 数据集引用
* @param data 待保存数据
* @param account 当前操作账号
*/
void saveRow(DatasetRef datasetRef, JSONObject data, LoginAccount account);
/**
* 批量保存数据集行。
*
* @param datasetRef 数据集引用
* @param rows 待保存数据行
* @param account 当前操作账号
* @param batchSize 单批最大行数
*/
void saveRows(DatasetRef datasetRef, List<JSONObject> rows, LoginAccount account, int batchSize);
/**
* 使用稳定幂等键批量保存数据集行。
*
* @param datasetRef 数据集引用
* @param rows 待保存数据行
* @param account 当前操作账号
* @param batchSize 单批最大行数
* @param idempotencyKey 稳定业务幂等键；为空时保持普通写入语义
* @return 本次实际执行写入时为 {@code true}，已有成功记录时为 {@code false}
*/
default boolean saveRowsIdempotently(DatasetRef datasetRef,
List<JSONObject> rows,
LoginAccount account,
int batchSize,
String idempotencyKey) {
saveRows(datasetRef, rows, account, batchSize);
return true;
}
/**
* 删除单行数据。
*
* @param datasetRef 数据集引用
* @param id 数据主键
* @param account 当前操作账号
*/
void deleteRow(DatasetRef datasetRef, BigInteger id, LoginAccount account);
}

174
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/execution/service/impl/DatacenterDatasetQueryServiceImpl.java
View File

@@ -30,8 +30,13 @@ import tech.easyflow.datacenter.utils.SqlSupportUtils;
import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@Service
public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQueryService {
@@ -73,6 +78,98 @@ public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQuery
@Override
public List<Row> queryBySql(DatacenterSqlQueryRequest request) {
ResolvedSqlQuery query = resolveSqlQuery(request);
return query.connector.queryBySql(
query.source, query.sql);
}
/**
* {@inheritDoc}
*/
@Override
public void consumeBySql(
DatacenterSqlQueryRequest request,
int fetchSize,
Consumer<Row> consumer) {
if (fetchSize <= 0 || consumer == null) {
throw new IllegalArgumentException(
"fetchSize and consumer must be valid");
}
ResolvedSqlQuery query = resolveSqlQuery(request);
int maxRows = Integer.getInteger(
"easyflow.datacenter.query.max-rows",
1_000_000);
long maxBytes = Long.getLong(
"easyflow.datacenter.query.max-bytes",
512L * 1024L * 1024L);
long[] accumulatedRows = {0L};
long[] accumulatedBytes = {0L};
query.connector.consumeBySql(
query.source,
query.sql,
fetchSize,
row -> {
accumulatedRows[0]++;
if (maxRows > 0
&& accumulatedRows[0] > maxRows) {
throw new BusinessException(
"数据集查询结果超过行数上限: "
+ maxRows);
}
for (Map.Entry<String, Object> entry
: row.entrySet()) {
accumulatedBytes[0] +=
estimateQueryValueBytes(
entry.getKey(),
entry.getValue());
if (maxBytes > 0L
&& accumulatedBytes[0] > maxBytes) {
throw new BusinessException(
"数据集查询结果超过字节上限: "
+ maxBytes);
}
}
consumer.accept(row);
}
);
}
/**
* 估算查询值在 JVM 中的最低占用，用于跨页累计保护。
*
* @param columnLabel 列名
* @param value 列值
* @return 估算字节数
*/
private long estimateQueryValueBytes(
String columnLabel,
Object value) {
long bytes = columnLabel == null
? 0L
: (long) columnLabel.length()
* Character.BYTES;
if (value == null) {
return bytes + 8L;
}
if (value instanceof byte[] binary) {
return bytes + binary.length;
}
if (value instanceof CharSequence text) {
return bytes
+ (long) text.length()
* Character.BYTES;
}
return bytes + 64L;
}
/**
* 校验请求并解析实际连接器与可执行 SQL。
*
* @param request SQL 查询请求
* @return 已解析查询
*/
private ResolvedSqlQuery resolveSqlQuery(
DatacenterSqlQueryRequest request) {
if (request == null || request.getDatasetRef() == null) {
throw new BusinessException("datasetRef 不能为空");
}
@@ -90,12 +187,33 @@ public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQuery
if (CollectionUtils.isEmpty(managedTables)) {
throw new BusinessException("当前连接下没有已接入表");
}
Map<BigInteger, DatacenterCatalog> catalogsById =
loadCatalogsById(managedTables);
SqlSupportUtils.ResolvedSql resolvedSql = SqlSupportUtils.resolve(
sql,
managedTables.stream().map(this::toManagedSqlTable).toList()
managedTables.stream()
.map(table -> toManagedSqlTable(
table, catalogsById))
.toList()
);
DatacenterConnector connector = connectorRegistry.getConnector(source.getSourceType());
return connector.queryBySql(source, resolvedSql.getExecutableSql());
return new ResolvedSqlQuery(
source,
connector,
resolvedSql.getExecutableSql());
}
/**
* 一次已校验的 SQL 查询上下文。
*
* @param source 数据源
* @param connector 数据连接器
* @param sql 可执行 SQL
*/
private record ResolvedSqlQuery(
DatacenterSource source,
DatacenterConnector connector,
String sql) {
}
@Override
@@ -115,6 +233,23 @@ public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQuery
return response;
}
/**
* {@inheritDoc}
*/
@Override
public DatacenterSchemaResponse getLocation(DatasetRef datasetRef) {
DatacenterTable table = resolveTable(datasetRef);
DatacenterSchemaResponse response =
new DatacenterSchemaResponse();
response.setDatasetRef(datasetRef);
response.setSource(
registryService.getSourceRequired(table.getSourceId()));
response.setCatalog(
registryService.getCatalogById(table.getCatalogId()));
response.setTable(table);
return response;
}
private DatacenterTable resolveTable(DatasetRef datasetRef) {
if (datasetRef.getTableId() != null) {
return registryService.getTableWithFields(datasetRef.getTableId());
@@ -174,8 +309,13 @@ public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQuery
return null;
}
private SqlSupportUtils.ManagedTable toManagedSqlTable(DatacenterTable table) {
DatacenterCatalog catalog = registryService.getCatalogById(table.getCatalogId());
private SqlSupportUtils.ManagedTable toManagedSqlTable(
DatacenterTable table,
Map<BigInteger, DatacenterCatalog> catalogsById) {
BigInteger catalogId = table.getCatalogId();
DatacenterCatalog catalog = catalogId == null
? null
: catalogsById.get(catalogId);
return new SqlSupportUtils.ManagedTable(
catalog == null ? null : catalog.getCatalogName(),
table.getTableName(),
@@ -183,6 +323,32 @@ public class DatacenterDatasetQueryServiceImpl implements DatacenterDatasetQuery
);
}
/**
* 一次批量加载 SQL 白名单表关联的目录，避免逐表查询。
*
* @param managedTables 已接入表
* @return 目录 ID 到目录实体
*/
private Map<BigInteger, DatacenterCatalog> loadCatalogsById(
List<DatacenterTable> managedTables) {
Set<BigInteger> catalogIds = managedTables.stream()
.map(DatacenterTable::getCatalogId)
.filter(java.util.Objects::nonNull)
.collect(Collectors.toCollection(
LinkedHashSet::new));
if (catalogIds.isEmpty()) {
return Map.of();
}
QueryWrapper wrapper = QueryWrapper.create();
wrapper.in(DatacenterCatalog::getId, catalogIds);
return catalogMapper.selectListByQuery(wrapper).stream()
.collect(Collectors.toMap(
DatacenterCatalog::getId,
Function.identity(),
(first, ignored) -> first,
LinkedHashMap::new));
}
private String resolvePhysicalTableName(DatacenterTable table) {
if (table == null) {
return null;

104
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/execution/service/impl/DatacenterDatasetWriteServiceImpl.java
View File

@@ -1,7 +1,10 @@
package tech.easyflow.datacenter.execution.service.impl;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import org.springframework.stereotype.Service;
import tech.easyflow.common.cache.RedisIdempotencyExecutor;
import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.datacenter.connector.DatacenterConnector;
@@ -14,6 +17,11 @@ import tech.easyflow.datacenter.execution.service.DatacenterDatasetWriteService;
import javax.annotation.Resource;
import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HexFormat;
import java.util.List;
@Service
public class DatacenterDatasetWriteServiceImpl implements DatacenterDatasetWriteService {
@@ -22,7 +30,12 @@ public class DatacenterDatasetWriteServiceImpl implements DatacenterDatasetWrite
private DatacenterDatasetRegistryService registryService;
@Resource
private DatacenterConnectorRegistry connectorRegistry;
@Resource
private RedisIdempotencyExecutor idempotencyExecutor;
/**
* {@inheritDoc}
*/
@Override
public void saveRow(DatasetRef datasetRef, JSONObject data, LoginAccount account) {
DatacenterTable table = resolveTable(datasetRef);
@@ -31,6 +44,55 @@ public class DatacenterDatasetWriteServiceImpl implements DatacenterDatasetWrite
connector.saveRow(source, table, data, account);
}
/**
* {@inheritDoc}
*/
@Override
public void saveRows(DatasetRef datasetRef, List<JSONObject> rows, LoginAccount account, int batchSize) {
if (rows == null || rows.isEmpty()) {
return;
}
DatacenterTable table = resolveTable(datasetRef);
DatacenterSource source = registryService.getSourceRequired(table.getSourceId());
DatacenterConnector connector = connectorRegistry.getConnector(source.getSourceType());
connector.saveRows(source, table, rows, account, Math.max(1, batchSize));
}
/**
* {@inheritDoc}
*/
@Override
public boolean saveRowsIdempotently(DatasetRef datasetRef,
List<JSONObject> rows,
LoginAccount account,
int batchSize,
String idempotencyKey) {
if (rows == null || rows.isEmpty()) {
return true;
}
DatacenterTable table = resolveTable(datasetRef);
DatacenterSource source = registryService.getSourceRequired(table.getSourceId());
DatacenterConnector connector = connectorRegistry.getConnector(source.getSourceType());
String payloadHash = sha256Rows(rows);
if (idempotencyKey == null || idempotencyKey.isBlank()) {
connector.saveRows(source, table, rows, account, Math.max(1, batchSize));
return true;
}
String receiptKey = sha256(idempotencyKey);
return idempotencyExecutor.executeOnce(idempotencyKey, payloadHash, () ->
connector.saveRowsIdempotently(
source,
table,
rows,
account,
Math.max(1, batchSize),
receiptKey,
payloadHash));
}
/**
* {@inheritDoc}
*/
@Override
public void deleteRow(DatasetRef datasetRef, BigInteger id, LoginAccount account) {
DatacenterTable table = resolveTable(datasetRef);
@@ -45,4 +107,46 @@ public class DatacenterDatasetWriteServiceImpl implements DatacenterDatasetWrite
}
return registryService.getTableWithFields(datasetRef.getTableId());
}
/**
* 计算稳定的 SHA-256 摘要。
*
* @param value 原始文本
* @return 十六进制摘要
*/
private String sha256(String value) {
try {
return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256")
.digest(value.getBytes(StandardCharsets.UTF_8)));
} catch (NoSuchAlgorithmException error) {
throw new IllegalStateException("SHA-256 is unavailable", error);
}
}
/**
* 逐行计算与排序字段 JSON 数组等价的 SHA-256，避免构造整批字符串副本。
*
* @param rows 待写入行
* @return 十六进制摘要
*/
private String sha256Rows(List<JSONObject> rows) {
try {
MessageDigest digest =
MessageDigest.getInstance("SHA-256");
digest.update((byte) '[');
for (int index = 0; index < rows.size(); index++) {
if (index > 0) {
digest.update((byte) ',');
}
digest.update(JSON.toJSONBytes(
rows.get(index),
JSONWriter.Feature.MapSortField));
}
digest.update((byte) ']');
return HexFormat.of().formatHex(digest.digest());
} catch (NoSuchAlgorithmException error) {
throw new IllegalStateException(
"SHA-256 is unavailable", error);
}
}
}

88
easyflow-modules/easyflow-module-datacenter/src/main/java/tech/easyflow/datacenter/schedule/DatacenterWriteReceiptCleanupJob.java Normal file
View File

@@ -0,0 +1,88 @@
package tech.easyflow.datacenter.schedule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.easyflow.common.cache.DistributedScheduledLock;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
/**
* 定期分批清理过期的数据集写入幂等回执。
*/
@Component
public class DatacenterWriteReceiptCleanupJob {
private static final Logger log =
LoggerFactory.getLogger(DatacenterWriteReceiptCleanupJob.class);
private static final String DELETE_SQL =
"DELETE FROM tb_datacenter_write_receipt "
+ "WHERE created < ? ORDER BY created LIMIT ?";
private static final long MIN_RETENTION_DAYS = 7L;
private final JdbcTemplate jdbcTemplate;
private final long retentionDays;
private final int batchSize;
private final int maxBatches;
/**
* 创建回执清理任务。
*
* @param jdbcTemplate JDBC 操作模板
* @param retentionDays 回执保留天数，最低七天
* @param batchSize 单批删除行数
* @param maxBatches 单次调度最多删除批次
*/
public DatacenterWriteReceiptCleanupJob(
JdbcTemplate jdbcTemplate,
@Value("${easyflow.workflow.data-write-receipt-retention-days:14}")
long retentionDays,
@Value("${easyflow.workflow.data-write-receipt-cleanup-batch-size:1000}")
int batchSize,
@Value("${easyflow.workflow.data-write-receipt-cleanup-max-batches:20}")
int maxBatches) {
this.jdbcTemplate = jdbcTemplate;
this.retentionDays = Math.max(MIN_RETENTION_DAYS, retentionDays);
this.batchSize = Math.max(1, batchSize);
this.maxBatches = Math.max(1, maxBatches);
}
/**
* 在单个集群节点上删除一批超过保留期的回执。
*/
@Scheduled(
fixedDelayString =
"${easyflow.workflow.data-write-receipt-cleanup-interval:1h}",
initialDelayString =
"${easyflow.workflow.data-write-receipt-cleanup-initial-delay:10m}")
@DistributedScheduledLock(
key = "easyflow:schedule:datacenter-write-receipt-cleanup",
leaseSeconds = 300L)
public void cleanup() {
Timestamp cutoff = Timestamp.from(
Instant.now().minus(retentionDays, ChronoUnit.DAYS));
int totalDeleted = 0;
try {
for (int batch = 0; batch < maxBatches; batch++) {
int deleted = jdbcTemplate.update(
DELETE_SQL, cutoff, batchSize);
totalDeleted += deleted;
if (deleted < batchSize) {
break;
}
}
if (totalDeleted > 0) {
log.info(
"Cleaned {} expired datacenter write receipts",
totalDeleted);
}
} catch (RuntimeException error) {
log.error("Datacenter write receipt cleanup failed", error);
}
}
}

133
easyflow-modules/easyflow-module-datacenter/src/test/java/tech/easyflow/datacenter/execution/service/impl/DatacenterDatasetQueryServiceImplTest.java Normal file
View File

@@ -0,0 +1,133 @@
package tech.easyflow.datacenter.execution.service.impl;
import com.mybatisflex.core.row.Row;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import tech.easyflow.datacenter.connector.DatacenterConnector;
import tech.easyflow.datacenter.connector.DatacenterConnectorRegistry;
import tech.easyflow.datacenter.entity.DatacenterTable;
import tech.easyflow.datacenter.execution.model.DatacenterSqlQueryRequest;
import tech.easyflow.datacenter.execution.model.DatasetRef;
import tech.easyflow.datacenter.meta.entity.DatacenterSource;
import tech.easyflow.datacenter.meta.service.DatacenterDatasetRegistryService;
import java.lang.reflect.Field;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
/**
* {@link DatacenterDatasetQueryServiceImpl} 分页 SQL 读取回归测试。
*/
public class DatacenterDatasetQueryServiceImplTest {
/**
* 验证惰性迭代器逐页读取并保持原始行顺序。
*
* @throws Exception 测试依赖注入失败时抛出
*/
@Test
public void consumeBySqlShouldStreamSingleQueryInOrder()
throws Exception {
BigInteger sourceId = BigInteger.valueOf(1001L);
DatacenterSource source = new DatacenterSource();
source.setId(sourceId);
source.setSourceType("MYSQL");
DatacenterTable table = new DatacenterTable();
table.setId(BigInteger.valueOf(2001L));
table.setSourceId(sourceId);
table.setTableName("orders");
table.setActualTable("orders_actual");
DatacenterDatasetRegistryService registry =
Mockito.mock(
DatacenterDatasetRegistryService.class);
Mockito.when(registry.getSourceRequired(sourceId))
.thenReturn(source);
Mockito.when(registry.listManagedTables(
sourceId, null))
.thenReturn(List.of(table));
DatacenterConnector connector =
Mockito.mock(DatacenterConnector.class);
Mockito.doAnswer(invocation -> {
@SuppressWarnings("unchecked")
Consumer<Row> consumer =
invocation.getArgument(3);
consumer.accept(row(1));
consumer.accept(row(2));
consumer.accept(row(3));
return null;
})
.when(connector)
.consumeBySql(
ArgumentMatchers.eq(source),
ArgumentMatchers.anyString(),
ArgumentMatchers.eq(2),
ArgumentMatchers.any());
DatacenterConnectorRegistry connectors =
Mockito.mock(
DatacenterConnectorRegistry.class);
Mockito.when(connectors.getConnector("MYSQL"))
.thenReturn(connector);
DatacenterDatasetQueryServiceImpl service =
new DatacenterDatasetQueryServiceImpl();
setField(service, "registryService", registry);
setField(service, "connectorRegistry", connectors);
DatacenterSqlQueryRequest request =
new DatacenterSqlQueryRequest();
DatasetRef datasetRef = new DatasetRef();
datasetRef.setSourceId(sourceId);
request.setDatasetRef(datasetRef);
request.setSql("SELECT * FROM orders ORDER BY id");
List<Integer> ids = new ArrayList<>();
service.consumeBySql(
request,
2,
current -> ids.add(
current.getInt("id")));
Assert.assertEquals(
List.of(1, 2, 3), ids);
Mockito.verify(connector, Mockito.times(1))
.consumeBySql(
ArgumentMatchers.eq(source),
ArgumentMatchers.anyString(),
ArgumentMatchers.eq(2),
ArgumentMatchers.any());
}
/**
* 创建测试数据行。
*
* @param id 行 ID
* @return 数据行
*/
private Row row(int id) {
Row row = new Row();
row.put("id", id);
return row;
}
/**
* 注入被测服务依赖。
*
* @param target 被测对象
* @param name 字段名
* @param value 字段值
* @throws Exception 反射失败时抛出
*/
private void setField(
Object target,
String name,
Object value) throws Exception {
Field field = target.getClass()
.getDeclaredField(name);
field.setAccessible(true);
field.set(target, value);
}
}