Compare commits
2 Commits
3e34b65bcb
...
1f37b0a8ae
| Author | SHA1 | Date | |
|---|---|---|---|
| 1f37b0a8ae | |||
| c00369f6b9 |
@@ -0,0 +1,94 @@
|
||||
package tech.easyflow.dataspace.config;
|
||||
|
||||
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
|
||||
import java.time.Duration;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.util.unit.DataSize;
|
||||
|
||||
/**
|
||||
* 数据空间联邦查询的节点硬上限和 Query Scope 上限。
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "easyflow.dataspace.federation")
|
||||
public class DataspaceFederationProperties implements InitializingBean {
|
||||
|
||||
private int maximumReferencedSources = 2;
|
||||
private int maximumFragments = 8;
|
||||
private int maximumConcurrentFragments = 2;
|
||||
private long maximumIntermediateRows = 100_000L;
|
||||
private DataSize maximumIntermediateBytes = DataSize.ofMegabytes(64);
|
||||
private Duration maximumExecutionTime = Duration.ofSeconds(60);
|
||||
|
||||
/** @return 单条 SQL 最多实际引用的数据源数 */
|
||||
public int getMaximumReferencedSources() { return maximumReferencedSources; }
|
||||
/** @param maximumReferencedSources 单条 SQL 最多实际引用的数据源数 */
|
||||
public void setMaximumReferencedSources(int maximumReferencedSources) {
|
||||
this.maximumReferencedSources = maximumReferencedSources;
|
||||
}
|
||||
/** @return 最多物理查询分片数 */
|
||||
public int getMaximumFragments() { return maximumFragments; }
|
||||
/** @param maximumFragments 最多物理查询分片数 */
|
||||
public void setMaximumFragments(int maximumFragments) {
|
||||
this.maximumFragments = maximumFragments;
|
||||
}
|
||||
/** @return 最大并发分片数 */
|
||||
public int getMaximumConcurrentFragments() { return maximumConcurrentFragments; }
|
||||
/** @param maximumConcurrentFragments 最大并发分片数 */
|
||||
public void setMaximumConcurrentFragments(int maximumConcurrentFragments) {
|
||||
this.maximumConcurrentFragments = maximumConcurrentFragments;
|
||||
}
|
||||
/** @return 最大中间结果行数 */
|
||||
public long getMaximumIntermediateRows() { return maximumIntermediateRows; }
|
||||
/** @param maximumIntermediateRows 最大中间结果行数 */
|
||||
public void setMaximumIntermediateRows(long maximumIntermediateRows) {
|
||||
this.maximumIntermediateRows = maximumIntermediateRows;
|
||||
}
|
||||
/** @return 最大中间结果估算字节数 */
|
||||
public DataSize getMaximumIntermediateBytes() { return maximumIntermediateBytes; }
|
||||
/** @param maximumIntermediateBytes 最大中间结果估算字节数 */
|
||||
public void setMaximumIntermediateBytes(DataSize maximumIntermediateBytes) {
|
||||
this.maximumIntermediateBytes = maximumIntermediateBytes;
|
||||
}
|
||||
/** @return 联邦执行总时限 */
|
||||
public Duration getMaximumExecutionTime() { return maximumExecutionTime; }
|
||||
/** @param maximumExecutionTime 联邦执行总时限 */
|
||||
public void setMaximumExecutionTime(Duration maximumExecutionTime) {
|
||||
this.maximumExecutionTime = maximumExecutionTime;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构造 Engine 和 Query Scope 共用的不可变执行策略。
|
||||
*
|
||||
* @return 联邦执行策略
|
||||
*/
|
||||
public FederationExecutionPolicy toExecutionPolicy() {
|
||||
return new FederationExecutionPolicy(
|
||||
maximumReferencedSources,
|
||||
maximumFragments,
|
||||
maximumConcurrentFragments,
|
||||
maximumIntermediateRows,
|
||||
maximumIntermediateBytes.toBytes(),
|
||||
maximumExecutionTime.toMillis());
|
||||
}
|
||||
|
||||
/**
|
||||
* 在 Engine 创建前拒绝无界或互相矛盾的配置。
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
if (maximumReferencedSources <= 0 || maximumFragments <= 0
|
||||
|| maximumConcurrentFragments <= 0 || maximumIntermediateRows <= 0
|
||||
|| maximumIntermediateBytes == null || maximumIntermediateBytes.toBytes() <= 0
|
||||
|| maximumExecutionTime == null || maximumExecutionTime.isZero()
|
||||
|| maximumExecutionTime.isNegative()) {
|
||||
throw new IllegalStateException("数据空间联邦查询资源上限必须为正值");
|
||||
}
|
||||
if (maximumReferencedSources > maximumFragments) {
|
||||
throw new IllegalStateException("最大引用数据源数不能超过最大分片数");
|
||||
}
|
||||
if (maximumConcurrentFragments > maximumFragments) {
|
||||
throw new IllegalStateException("最大并发分片数不能超过最大分片数");
|
||||
}
|
||||
toExecutionPolicy();
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package tech.easyflow.dataspace.config;
|
||||
|
||||
import org.mybatis.spring.annotation.MapperScan;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.ComponentScan;
|
||||
|
||||
/**
|
||||
@@ -10,5 +11,6 @@ import org.springframework.context.annotation.ComponentScan;
|
||||
@MapperScan("tech.easyflow.dataspace.mapper")
|
||||
@ComponentScan("tech.easyflow.dataspace")
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties({DataspaceFederationProperties.class, DataspaceQueryProperties.class})
|
||||
public class DataspaceModuleConfig {
|
||||
}
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package tech.easyflow.dataspace.config;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.util.unit.DataSize;
|
||||
|
||||
/**
|
||||
* 数据空间同步查询结果保护配置。
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "easyflow.dataspace.query")
|
||||
public class DataspaceQueryProperties implements InitializingBean {
|
||||
|
||||
private DataSize maximumResultBytes = DataSize.ofMegabytes(16);
|
||||
private DataSize maximumCellBytes = DataSize.ofMegabytes(1);
|
||||
|
||||
/** @return 单次同步查询最大结果估算字节数 */
|
||||
public DataSize getMaximumResultBytes() { return maximumResultBytes; }
|
||||
/** @param maximumResultBytes 单次同步查询最大结果估算字节数 */
|
||||
public void setMaximumResultBytes(DataSize maximumResultBytes) {
|
||||
this.maximumResultBytes = maximumResultBytes;
|
||||
}
|
||||
/** @return 单个字段最大字节数 */
|
||||
public DataSize getMaximumCellBytes() { return maximumCellBytes; }
|
||||
/** @param maximumCellBytes 单个字段最大字节数 */
|
||||
public void setMaximumCellBytes(DataSize maximumCellBytes) {
|
||||
this.maximumCellBytes = maximumCellBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* 在查询服务启动前拒绝无效或互相矛盾的结果保护配置。
|
||||
*/
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
if (maximumResultBytes == null || maximumResultBytes.toBytes() <= 0
|
||||
|| maximumCellBytes == null || maximumCellBytes.toBytes() <= 0) {
|
||||
throw new IllegalStateException("数据空间查询结果字节上限必须为正值");
|
||||
}
|
||||
if (maximumCellBytes.toBytes() > maximumResultBytes.toBytes()) {
|
||||
throw new IllegalStateException("单字段字节上限不能超过查询结果字节上限");
|
||||
}
|
||||
if (maximumCellBytes.toBytes() >= Integer.MAX_VALUE) {
|
||||
throw new IllegalStateException("单字段字节上限必须小于 2 GiB");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,11 @@
|
||||
package tech.easyflow.dataspace.federation;
|
||||
|
||||
import com.easyagents.federation.sql.adapter.jdbc.JdbcFederationSqlAdapterProvider;
|
||||
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
|
||||
import com.easyagents.federation.sql.api.FederationSqlEngine;
|
||||
import com.easyagents.federation.sql.api.FederationSqlEngines;
|
||||
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
|
||||
import com.easyagents.federation.sql.api.FederationSqlException;
|
||||
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
|
||||
import com.easyagents.federation.sql.source.FederationSourceDefinition;
|
||||
import com.easyagents.federation.sql.source.FederationSourceView;
|
||||
import com.easyagents.federation.sql.source.SourceApplyOptions;
|
||||
@@ -15,6 +16,7 @@ import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
|
||||
import tech.easyflow.dataspace.entity.DataspaceConnection;
|
||||
|
||||
/**
|
||||
@@ -25,6 +27,7 @@ public class DataspaceFederationRuntime {
|
||||
|
||||
private final DataspaceDefinitionFactory definitionFactory;
|
||||
private final FederationSqlEngine engine;
|
||||
private final FederationExecutionPolicy executionPolicy;
|
||||
private final ConcurrentMap<SourceId, FederationSourceDefinition> definitions =
|
||||
new ConcurrentHashMap<>();
|
||||
|
||||
@@ -34,16 +37,20 @@ public class DataspaceFederationRuntime {
|
||||
* @param resolver DataSource Resolver
|
||||
* @param definitionFactory Definition 工厂
|
||||
* @param sqlPolicy 数据空间表白名单策略
|
||||
* @param federationProperties 联邦资源上限
|
||||
*/
|
||||
public DataspaceFederationRuntime(
|
||||
DataspaceDataSourceResolver resolver,
|
||||
DataspaceDefinitionFactory definitionFactory,
|
||||
DataspaceSqlPolicy sqlPolicy) {
|
||||
DataspaceSqlPolicy sqlPolicy,
|
||||
DataspaceFederationProperties federationProperties) {
|
||||
this.definitionFactory = definitionFactory;
|
||||
this.executionPolicy = federationProperties.toExecutionPolicy();
|
||||
this.engine = FederationSqlEngines.builder()
|
||||
.dataSourceResolver(resolver)
|
||||
.adapter(new JdbcFederationSqlAdapterProvider())
|
||||
.policy(sqlPolicy)
|
||||
.federationExecutionPolicy(executionPolicy)
|
||||
.maximumPlanCacheEntries(1024)
|
||||
.crossSourceEnabled(true)
|
||||
.build();
|
||||
@@ -125,6 +132,15 @@ public class DataspaceFederationRuntime {
|
||||
return engine;
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回 Engine 与 Query Scope 共用的资源上限。
|
||||
*
|
||||
* @return 联邦执行策略
|
||||
*/
|
||||
public FederationExecutionPolicy executionPolicy() {
|
||||
return executionPolicy;
|
||||
}
|
||||
|
||||
/**
|
||||
* 应用停止时关闭计划缓存与连接池。
|
||||
*/
|
||||
|
||||
@@ -15,7 +15,6 @@ import com.easyagents.federation.sql.execute.FederationQueryMetricsSnapshot;
|
||||
import com.easyagents.federation.sql.execute.FederationResultCursor;
|
||||
import com.easyagents.federation.sql.execute.QueryId;
|
||||
import com.easyagents.federation.sql.execute.SqlExecutionOptions;
|
||||
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
|
||||
import com.easyagents.federation.sql.federation.FederationLogicalTableDefinition;
|
||||
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
|
||||
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
|
||||
@@ -48,6 +47,7 @@ import org.slf4j.LoggerFactory;
|
||||
import org.springframework.stereotype.Service;
|
||||
import tech.easyflow.common.entity.LoginAccount;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.dataspace.config.DataspaceQueryProperties;
|
||||
import tech.easyflow.dataspace.entity.DataspaceConnection;
|
||||
import tech.easyflow.dataspace.entity.DataspaceObject;
|
||||
import tech.easyflow.dataspace.entity.DataspaceQueryAudit;
|
||||
@@ -71,15 +71,14 @@ public class DataspaceQueryService {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(DataspaceQueryService.class);
|
||||
private static final int DEFAULT_MAX_ROWS = 500;
|
||||
private static final int MAX_ROWS = 2_000;
|
||||
private static final int MAX_RESULT_ROWS = 10_000;
|
||||
private static final int DEFAULT_TIMEOUT_SECONDS = 30;
|
||||
private static final int MAX_TIMEOUT_SECONDS = 60;
|
||||
private static final long MAX_RESULT_BYTES = 16L * 1024L * 1024L;
|
||||
private static final long MAX_CELL_BYTES = 1024L * 1024L;
|
||||
private final DataspaceService dataspaceService;
|
||||
private final DataspaceFederationRuntime runtime;
|
||||
private final DataspaceDefinitionFactory definitionFactory;
|
||||
private final DataspaceQueryAuditMapper auditMapper;
|
||||
private final DataspaceQueryProperties queryProperties;
|
||||
private final ConcurrentMap<String, ActiveQueryOwner> activeQueries = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
@@ -89,16 +88,19 @@ public class DataspaceQueryService {
|
||||
* @param runtime Federation Runtime
|
||||
* @param definitionFactory Definition 工厂
|
||||
* @param auditMapper 查询审计 Mapper
|
||||
* @param queryProperties 同步查询结果保护配置
|
||||
*/
|
||||
public DataspaceQueryService(
|
||||
DataspaceService dataspaceService,
|
||||
DataspaceFederationRuntime runtime,
|
||||
DataspaceDefinitionFactory definitionFactory,
|
||||
DataspaceQueryAuditMapper auditMapper) {
|
||||
DataspaceQueryAuditMapper auditMapper,
|
||||
DataspaceQueryProperties queryProperties) {
|
||||
this.dataspaceService = dataspaceService;
|
||||
this.runtime = runtime;
|
||||
this.definitionFactory = definitionFactory;
|
||||
this.auditMapper = auditMapper;
|
||||
this.queryProperties = queryProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -112,7 +114,7 @@ public class DataspaceQueryService {
|
||||
DataspaceService.RuntimeSnapshot snapshot = dataspaceService.loadSnapshot(request.dataspaceId());
|
||||
String executableSql = normalizeSql(request.sql());
|
||||
FederationQueryScopeDefinition scope = scopeSafely(snapshot, false);
|
||||
int maxRows = normalize(request.maxRows(), DEFAULT_MAX_ROWS, MAX_ROWS);
|
||||
int maxRows = normalizeMaximumRows(request.maxRows());
|
||||
int timeout = normalize(request.timeoutSeconds(), DEFAULT_TIMEOUT_SECONDS, MAX_TIMEOUT_SECONDS);
|
||||
QueryId queryId = request.queryId() == null || request.queryId().isBlank()
|
||||
? QueryId.create() : new QueryId(request.queryId().trim());
|
||||
@@ -133,7 +135,8 @@ public class DataspaceQueryService {
|
||||
.map(this::columnView).toList();
|
||||
List<List<Object>> rows = new ArrayList<>();
|
||||
DataspaceQueryResultBudget resultBudget = new DataspaceQueryResultBudget(
|
||||
MAX_RESULT_BYTES, MAX_CELL_BYTES);
|
||||
queryProperties.getMaximumResultBytes().toBytes(),
|
||||
queryProperties.getMaximumCellBytes().toBytes());
|
||||
while (cursor.next()) {
|
||||
resultBudget.beginRow(columns.size());
|
||||
List<Object> row = new ArrayList<>(columns.size());
|
||||
@@ -329,7 +332,7 @@ public class DataspaceQueryService {
|
||||
String defaultBinding = definitions.keySet().iterator().next();
|
||||
return FederationQueryScopeDefinition.virtual(
|
||||
scopeId(snapshot), snapshot.revision().getRevisionNo(), definitions,
|
||||
defaultBinding, logicalTables, FederationExecutionPolicy.basic());
|
||||
defaultBinding, logicalTables, runtime.executionPolicy());
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -503,14 +506,15 @@ public class DataspaceQueryService {
|
||||
*/
|
||||
private String readText(Reader reader) throws Exception {
|
||||
try (reader) {
|
||||
long maximumCellBytes = queryProperties.getMaximumCellBytes().toBytes();
|
||||
char[] buffer = new char[8_192];
|
||||
StringBuilder value = new StringBuilder();
|
||||
int count;
|
||||
while ((count = reader.read(buffer)) >= 0) {
|
||||
value.append(buffer, 0, count);
|
||||
if (value.length() > MAX_CELL_BYTES) {
|
||||
if (value.length() > maximumCellBytes) {
|
||||
throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
|
||||
"单个文本字段超过 1 MiB 限制");
|
||||
"单个文本字段超过配置的字节限制");
|
||||
}
|
||||
}
|
||||
return value.toString();
|
||||
@@ -526,10 +530,12 @@ public class DataspaceQueryService {
|
||||
*/
|
||||
private byte[] readBytes(InputStream stream) throws Exception {
|
||||
try (stream) {
|
||||
byte[] bytes = stream.readNBytes((int) MAX_CELL_BYTES + 1);
|
||||
if (bytes.length > MAX_CELL_BYTES) {
|
||||
int maximumCellBytes = Math.toIntExact(
|
||||
queryProperties.getMaximumCellBytes().toBytes());
|
||||
byte[] bytes = stream.readNBytes(maximumCellBytes + 1);
|
||||
if (bytes.length > maximumCellBytes) {
|
||||
throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
|
||||
"单个二进制字段超过 1 MiB 限制");
|
||||
"单个二进制字段超过配置的字节限制");
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
@@ -651,6 +657,22 @@ public class DataspaceQueryService {
|
||||
return Math.min(normalized, maximum);
|
||||
}
|
||||
|
||||
/**
|
||||
* 规范化工作台最大返回行数,并拒绝绕过产品上限的直接 API 请求。
|
||||
*
|
||||
* @param value 调用方值
|
||||
* @return 允许执行的最大返回行数
|
||||
* @throws BusinessException 值不在 1~10,000 范围内时抛出
|
||||
*/
|
||||
static int normalizeMaximumRows(Integer value) {
|
||||
int normalized = value == null ? DEFAULT_MAX_ROWS : value;
|
||||
if (normalized <= 0 || normalized > MAX_RESULT_ROWS) {
|
||||
throw DataspaceErrorCode.QUERY_REQUEST_INVALID.exception(
|
||||
"最大返回行数必须在 1~10,000 之间");
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/**
|
||||
* 将纳秒转换为毫秒并保留未知值。
|
||||
*
|
||||
|
||||
@@ -24,6 +24,7 @@ import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
import tech.easyflow.common.entity.LoginAccount;
|
||||
import tech.easyflow.common.util.SearchKeywordUtil;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
|
||||
import tech.easyflow.dataspace.entity.Dataspace;
|
||||
import tech.easyflow.dataspace.entity.DataspaceConnection;
|
||||
import tech.easyflow.dataspace.entity.DataspaceObject;
|
||||
@@ -53,7 +54,6 @@ public class DataspaceService {
|
||||
private static final Duration LOCK_WAIT = Duration.ofSeconds(3);
|
||||
private static final Duration LOCK_LEASE = Duration.ofSeconds(30);
|
||||
private static final int MAX_TABLES = 100;
|
||||
private static final int MAX_SOURCES = 2;
|
||||
private final DataspaceMapper dataspaceMapper;
|
||||
private final DataspaceRevisionMapper revisionMapper;
|
||||
private final DataspaceTableBindingMapper bindingMapper;
|
||||
@@ -64,6 +64,7 @@ public class DataspaceService {
|
||||
private final DataspaceMetadataResolver metadataResolver;
|
||||
private final RedisLockExecutor lockExecutor;
|
||||
private final TransactionTemplate transactionTemplate;
|
||||
private final DataspaceFederationProperties federationProperties;
|
||||
|
||||
/**
|
||||
* 创建数据空间服务。
|
||||
@@ -78,6 +79,7 @@ public class DataspaceService {
|
||||
* @param metadataResolver 元数据兼容解析器
|
||||
* @param lockExecutor 分布式锁执行器
|
||||
* @param transactionTemplate 事务模板
|
||||
* @param federationProperties 联邦执行与建模上限
|
||||
*/
|
||||
public DataspaceService(
|
||||
DataspaceMapper dataspaceMapper,
|
||||
@@ -89,7 +91,8 @@ public class DataspaceService {
|
||||
DataspaceDatabaseProviderRegistry providerRegistry,
|
||||
DataspaceMetadataResolver metadataResolver,
|
||||
RedisLockExecutor lockExecutor,
|
||||
TransactionTemplate transactionTemplate) {
|
||||
TransactionTemplate transactionTemplate,
|
||||
DataspaceFederationProperties federationProperties) {
|
||||
this.dataspaceMapper = dataspaceMapper;
|
||||
this.revisionMapper = revisionMapper;
|
||||
this.bindingMapper = bindingMapper;
|
||||
@@ -100,6 +103,7 @@ public class DataspaceService {
|
||||
this.metadataResolver = metadataResolver;
|
||||
this.lockExecutor = lockExecutor;
|
||||
this.transactionTemplate = transactionTemplate;
|
||||
this.federationProperties = federationProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -448,9 +452,10 @@ public class DataspaceService {
|
||||
}
|
||||
Set<BigInteger> connectionIds = objects.values().stream()
|
||||
.map(DataspaceObject::getConnectionId).collect(java.util.stream.Collectors.toSet());
|
||||
if (connectionIds.size() > MAX_SOURCES) {
|
||||
int maximumSources = federationProperties.getMaximumReferencedSources();
|
||||
if (connectionIds.size() > maximumSources) {
|
||||
throw DataspaceErrorCode.DEFINITION_INVALID.exception(
|
||||
"当前阶段单个数据空间最多支持两个物理数据源");
|
||||
"单个数据空间最多支持 " + maximumSources + " 个物理数据源");
|
||||
}
|
||||
Map<BigInteger, DataspaceConnection> connections = new HashMap<>();
|
||||
connectionMapper.selectListByQuery(QueryWrapper.create().in(DataspaceConnection::getId, connectionIds))
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package tech.easyflow.dataspace.config;
|
||||
|
||||
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
|
||||
import java.time.Duration;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.springframework.util.unit.DataSize;
|
||||
|
||||
/**
|
||||
* 数据空间联邦执行策略配置测试。
|
||||
*/
|
||||
public class DataspaceFederationPropertiesTest {
|
||||
|
||||
/**
|
||||
* 验证默认配置保持原有两源安全边界。
|
||||
*/
|
||||
@Test
|
||||
public void shouldKeepBasicPolicyByDefault() throws Exception {
|
||||
DataspaceFederationProperties properties = new DataspaceFederationProperties();
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
FederationExecutionPolicy policy = properties.toExecutionPolicy();
|
||||
|
||||
Assert.assertEquals(FederationExecutionPolicy.basic(), policy);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 benchmark 可只扩大引用源和分片数,不放宽行、字节和时限。
|
||||
*/
|
||||
@Test
|
||||
public void shouldBuildSixSourceBenchmarkPolicy() throws Exception {
|
||||
DataspaceFederationProperties properties = new DataspaceFederationProperties();
|
||||
properties.setMaximumReferencedSources(6);
|
||||
properties.setMaximumFragments(12);
|
||||
properties.setMaximumConcurrentFragments(2);
|
||||
properties.setMaximumIntermediateRows(100_000L);
|
||||
properties.setMaximumIntermediateBytes(DataSize.ofMegabytes(64));
|
||||
properties.setMaximumExecutionTime(Duration.ofSeconds(60));
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
FederationExecutionPolicy policy = properties.toExecutionPolicy();
|
||||
|
||||
Assert.assertEquals(6, policy.maximumReferencedSources());
|
||||
Assert.assertEquals(12, policy.maximumFragments());
|
||||
Assert.assertEquals(2, policy.maximumConcurrentFragments());
|
||||
Assert.assertEquals(100_000L, policy.maximumIntermediateRows());
|
||||
Assert.assertEquals(64L * 1024L * 1024L, policy.maximumIntermediateBytes());
|
||||
Assert.assertEquals(60_000L, policy.maximumExecutionTimeMillis());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证引用源数不能超过分片上限。
|
||||
*/
|
||||
@Test(expected = IllegalStateException.class)
|
||||
public void shouldRejectSourceLimitGreaterThanFragmentLimit() throws Exception {
|
||||
DataspaceFederationProperties properties = new DataspaceFederationProperties();
|
||||
properties.setMaximumReferencedSources(6);
|
||||
properties.setMaximumFragments(4);
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package tech.easyflow.dataspace.config;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.springframework.util.unit.DataSize;
|
||||
|
||||
/**
|
||||
* 数据空间同步查询结果保护配置测试。
|
||||
*/
|
||||
public class DataspaceQueryPropertiesTest {
|
||||
|
||||
/**
|
||||
* 验证生产默认值保持 16 MiB 总结果和 1 MiB 单字段保护。
|
||||
*/
|
||||
@Test
|
||||
public void shouldKeepSafeDefaults() throws Exception {
|
||||
DataspaceQueryProperties properties = new DataspaceQueryProperties();
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
|
||||
Assert.assertEquals(16L * 1024L * 1024L,
|
||||
properties.getMaximumResultBytes().toBytes());
|
||||
Assert.assertEquals(1024L * 1024L,
|
||||
properties.getMaximumCellBytes().toBytes());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 benchmark 可显式提高同步结果字节上限。
|
||||
*/
|
||||
@Test
|
||||
public void shouldAcceptLargeResultBenchmarkLimits() throws Exception {
|
||||
DataspaceQueryProperties properties = new DataspaceQueryProperties();
|
||||
properties.setMaximumResultBytes(DataSize.ofMegabytes(64));
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
|
||||
Assert.assertEquals(64L * 1024L * 1024L,
|
||||
properties.getMaximumResultBytes().toBytes());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证单字段上限不能超过总结果上限。
|
||||
*/
|
||||
@Test(expected = IllegalStateException.class)
|
||||
public void shouldRejectCellLimitGreaterThanResultLimit() throws Exception {
|
||||
DataspaceQueryProperties properties = new DataspaceQueryProperties();
|
||||
properties.setMaximumResultBytes(DataSize.ofKilobytes(512));
|
||||
properties.setMaximumCellBytes(DataSize.ofMegabytes(1));
|
||||
|
||||
properties.afterPropertiesSet();
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
|
||||
import tech.easyflow.dataspace.entity.Dataspace;
|
||||
import tech.easyflow.dataspace.entity.DataspaceConnection;
|
||||
import tech.easyflow.dataspace.entity.DataspaceObject;
|
||||
@@ -183,7 +184,8 @@ public class DataspaceMetadataCompatibilityTest {
|
||||
mock(DataspaceDatabaseProviderRegistry.class),
|
||||
resolver,
|
||||
mock(RedisLockExecutor.class),
|
||||
mock(org.springframework.transaction.support.TransactionTemplate.class));
|
||||
mock(org.springframework.transaction.support.TransactionTemplate.class),
|
||||
new DataspaceFederationProperties());
|
||||
return new Fixture(service);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
package tech.easyflow.dataspace.service;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
|
||||
/**
|
||||
* SQL 工作台最大返回行数测试。
|
||||
*/
|
||||
public class DataspaceQueryRowLimitTest {
|
||||
|
||||
/**
|
||||
* 验证调用方未指定时使用 500 行默认值。
|
||||
*/
|
||||
@Test
|
||||
public void shouldUseDefaultRowsWhenMissing() {
|
||||
assertEquals(500, DataspaceQueryService.normalizeMaximumRows(null));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 10,000 行产品上限可直接使用。
|
||||
*/
|
||||
@Test
|
||||
public void shouldAcceptMaximumRows() {
|
||||
assertEquals(10_000, DataspaceQueryService.normalizeMaximumRows(10_000));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证非正数和超过产品上限的请求被明确拒绝。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectRowsOutsideProductRange() {
|
||||
assertInvalid(0);
|
||||
assertInvalid(10_001);
|
||||
}
|
||||
|
||||
private void assertInvalid(int value) {
|
||||
try {
|
||||
DataspaceQueryService.normalizeMaximumRows(value);
|
||||
fail("超出范围的最大返回行数应被拒绝");
|
||||
} catch (BusinessException exception) {
|
||||
assertEquals(40066, exception.getErrorCode());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -117,7 +117,6 @@ onMounted(() => {
|
||||
width: 100%;
|
||||
height: 100%;
|
||||
min-height: 0;
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
@media (max-width: 900px) {
|
||||
|
||||
@@ -22,11 +22,32 @@ vi.mock('#/api/dataspace', () => ({
|
||||
queryDataspace: apiMocks.query,
|
||||
}));
|
||||
|
||||
vi.mock('element-plus/es/components/table-v2/style/css.mjs', () => ({}));
|
||||
|
||||
const mountWorkbench = () =>
|
||||
shallowMount(SqlWorkbenchPanel, {
|
||||
global: {
|
||||
stubs: {
|
||||
ElOption: {
|
||||
name: 'ElOption',
|
||||
props: ['label', 'value'],
|
||||
template: '<div></div>',
|
||||
},
|
||||
ElSelect: {
|
||||
name: 'ElSelect',
|
||||
props: ['modelValue'],
|
||||
template: '<div><slot /></div>',
|
||||
},
|
||||
ElTooltip: { template: '<div><slot /></div>' },
|
||||
ElTableV2: {
|
||||
name: 'ElTableV2',
|
||||
props: ['columns', 'data', 'rowKey'],
|
||||
template: '<div class="table-v2-test-stub"></div>',
|
||||
},
|
||||
ElAutoResizer: {
|
||||
name: 'ElAutoResizer',
|
||||
template: '<div><slot :height="320" :width="960" /></div>',
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -190,6 +211,112 @@ describe('sql workbench panel', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('offers row limits through 10000 and sends the selected value', async () => {
|
||||
const rows = Array.from({ length: 10_000 }, (_, index) => [index + 1]);
|
||||
apiMocks.query.mockResolvedValueOnce({
|
||||
columns: [
|
||||
{ jdbcType: -5, name: 'id', nullable: false, typeName: 'BIGINT' },
|
||||
],
|
||||
metrics: {
|
||||
databaseMillis: 20,
|
||||
firstRowMillis: 21,
|
||||
intermediateRows: 10_000,
|
||||
localMillis: 0,
|
||||
planCacheHit: true,
|
||||
planningMillis: 2,
|
||||
queryMode: 'SINGLE_SOURCE',
|
||||
returnedRows: 10_000,
|
||||
totalMillis: 30,
|
||||
truncated: false,
|
||||
},
|
||||
queryId: 'query-10000',
|
||||
rows,
|
||||
});
|
||||
const wrapper = mountWorkbench();
|
||||
await flushPromises();
|
||||
|
||||
const rowLimit = wrapper
|
||||
.findAllComponents({ name: 'ElSelect' })
|
||||
.find((component) => component.classes().includes('row-limit'));
|
||||
expect(rowLimit?.props('modelValue')).toBe(500);
|
||||
const labels = wrapper
|
||||
.findAllComponents({ name: 'ElOption' })
|
||||
.map((option) => option.props('label'))
|
||||
.filter((label) => String(label).startsWith('最多'));
|
||||
expect(labels).toEqual([
|
||||
'最多 100 行',
|
||||
'最多 500 行',
|
||||
'最多 1000 行',
|
||||
'最多 2000 行',
|
||||
'最多 5000 行',
|
||||
'最多 10000 行',
|
||||
]);
|
||||
|
||||
rowLimit?.vm.$emit('update:modelValue', 10_000);
|
||||
wrapper
|
||||
.getComponent({ name: 'CodeEditor' })
|
||||
.vm.$emit('update:modelValue', 'SELECT id FROM outlet');
|
||||
await flushPromises();
|
||||
await wrapper.get('[aria-label="执行 SQL"]').trigger('click');
|
||||
await flushPromises();
|
||||
|
||||
expect(apiMocks.query).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ maxRows: 10_000 }),
|
||||
);
|
||||
const virtualTable = wrapper.getComponent({ name: 'ElTableV2' });
|
||||
expect(virtualTable.props('data')).toHaveLength(10_000);
|
||||
expect(virtualTable.props('columns')).toHaveLength(2);
|
||||
expect(virtualTable.props('rowKey')).toBe('rowKey');
|
||||
});
|
||||
|
||||
it('exports every returned row independently of virtual rendering', async () => {
|
||||
const rows = Array.from({ length: 10_000 }, (_, index) => [index + 1]);
|
||||
apiMocks.query.mockResolvedValueOnce({
|
||||
columns: [
|
||||
{ jdbcType: -5, name: 'id', nullable: false, typeName: 'BIGINT' },
|
||||
],
|
||||
metrics: {
|
||||
databaseMillis: 20,
|
||||
firstRowMillis: 21,
|
||||
intermediateRows: 10_000,
|
||||
localMillis: 0,
|
||||
planCacheHit: true,
|
||||
planningMillis: 2,
|
||||
queryMode: 'SINGLE_SOURCE',
|
||||
returnedRows: 10_000,
|
||||
totalMillis: 30,
|
||||
truncated: false,
|
||||
},
|
||||
queryId: 'query-export',
|
||||
rows,
|
||||
});
|
||||
const createObjectUrl = vi
|
||||
.spyOn(URL, 'createObjectURL')
|
||||
.mockReturnValue('blob:query-export');
|
||||
const revokeObjectUrl = vi
|
||||
.spyOn(URL, 'revokeObjectURL')
|
||||
.mockImplementation(() => undefined);
|
||||
const anchorClick = vi
|
||||
.spyOn(HTMLAnchorElement.prototype, 'click')
|
||||
.mockImplementation(() => undefined);
|
||||
const wrapper = mountWorkbench();
|
||||
await flushPromises();
|
||||
|
||||
wrapper
|
||||
.getComponent({ name: 'CodeEditor' })
|
||||
.vm.$emit('update:modelValue', 'SELECT id FROM outlet');
|
||||
await flushPromises();
|
||||
await wrapper.get('[aria-label="执行 SQL"]').trigger('click');
|
||||
await flushPromises();
|
||||
await wrapper.get('[aria-label="导出 CSV"]').trigger('click');
|
||||
|
||||
const blob = createObjectUrl.mock.calls[0]?.[0];
|
||||
expect(blob).toBeInstanceOf(Blob);
|
||||
expect((blob as Blob).size).toBeGreaterThan(48_000);
|
||||
expect(anchorClick).toHaveBeenCalledOnce();
|
||||
expect(revokeObjectUrl).toHaveBeenCalledWith('blob:query-export');
|
||||
});
|
||||
|
||||
it('keeps result metrics and an icon-only export action beside the tabs', async () => {
|
||||
const wrapper = mountWorkbench();
|
||||
await flushPromises();
|
||||
|
||||
@@ -13,6 +13,7 @@ import type {
|
||||
|
||||
import {
|
||||
computed,
|
||||
h,
|
||||
nextTick,
|
||||
onBeforeUnmount,
|
||||
onMounted,
|
||||
@@ -48,10 +49,12 @@ import {
|
||||
ElMessage,
|
||||
ElOption,
|
||||
ElSelect,
|
||||
ElTable,
|
||||
ElTableColumn,
|
||||
ElTooltip,
|
||||
} from 'element-plus';
|
||||
import {
|
||||
ElAutoResizer as TableV2AutoResizer,
|
||||
ElTableV2,
|
||||
} from 'element-plus/es/components/table-v2/index.mjs';
|
||||
|
||||
import {
|
||||
cancelDataspaceQuery,
|
||||
@@ -65,6 +68,13 @@ import {
|
||||
import { dataspaceErrorMessage } from '../dataspace-errors';
|
||||
import { formatCalciteSql } from '../dataspace-sql';
|
||||
|
||||
import 'element-plus/es/components/table-v2/style/css.mjs';
|
||||
|
||||
interface VirtualResultRow {
|
||||
cells: unknown[];
|
||||
rowKey: number;
|
||||
}
|
||||
|
||||
const props = defineProps<{
|
||||
initialDataspaceId?: DataspaceId;
|
||||
}>();
|
||||
@@ -96,11 +106,51 @@ let resizeStartHeight = 0;
|
||||
const SPLITTER_SIZE = 8;
|
||||
const MIN_EDITOR_HEIGHT = 184;
|
||||
const MIN_OUTPUT_HEIGHT = 176;
|
||||
const MAX_ROW_OPTIONS = [100, 500, 1000, 2000, 5000, 10_000];
|
||||
const RESULT_HEADER_HEIGHT = 40;
|
||||
const RESULT_ROW_HEIGHT = 40;
|
||||
|
||||
const hasOutput = computed(() =>
|
||||
Boolean(result.value || explainResult.value || errorMessage.value),
|
||||
);
|
||||
|
||||
const virtualResultRows = computed<VirtualResultRow[]>(() =>
|
||||
(result.value?.rows || []).map((cells, rowKey) => ({ cells, rowKey })),
|
||||
);
|
||||
|
||||
const virtualResultColumns = computed(() => [
|
||||
{
|
||||
align: 'right' as const,
|
||||
cellRenderer: ({ rowIndex }: { rowIndex: number }) =>
|
||||
h('span', { class: 'result-index-cell' }, String(rowIndex + 1)),
|
||||
dataKey: 'rowKey',
|
||||
key: 'rowKey',
|
||||
title: '#',
|
||||
width: 56,
|
||||
},
|
||||
...(result.value?.columns || []).map((column, index) => ({
|
||||
cellRenderer: ({ rowData }: { rowData: VirtualResultRow }) => {
|
||||
const value = rowData.cells[index];
|
||||
const text =
|
||||
value === null || value === undefined ? 'NULL' : String(value);
|
||||
return h(
|
||||
'span',
|
||||
{
|
||||
class: ['result-cell-value', { 'is-null': value == null }],
|
||||
title: text,
|
||||
},
|
||||
text,
|
||||
);
|
||||
},
|
||||
dataKey: `column-${index}`,
|
||||
flexGrow: 1,
|
||||
key: `${column.name}-${index}`,
|
||||
minWidth: 160,
|
||||
title: column.name,
|
||||
width: 160,
|
||||
})),
|
||||
]);
|
||||
|
||||
const filteredTables = computed(() => {
|
||||
const normalized = keyword.value.trim().toLowerCase();
|
||||
if (!normalized) return detail.value?.tables || [];
|
||||
@@ -462,6 +512,11 @@ function queryRequest(queryId: string) {
|
||||
};
|
||||
}
|
||||
|
||||
/** 为虚拟结果表提供稳定的隔行底色。 */
|
||||
function resultRowClass({ rowIndex }: { rowIndex: number }) {
|
||||
return rowIndex % 2 === 1 ? 'is-striped' : '';
|
||||
}
|
||||
|
||||
/** 执行 Calcite SQL,并展示数据与常用消耗指标。 */
|
||||
async function execute() {
|
||||
const epoch = ++requestEpoch;
|
||||
@@ -711,9 +766,9 @@ defineExpose({ loadSpaces });
|
||||
/>
|
||||
</ElTooltip>
|
||||
</div>
|
||||
<ElSelect v-model="maxRows" class="row-limit">
|
||||
<ElSelect v-model="maxRows" class="row-limit" aria-label="最大返回行数">
|
||||
<ElOption
|
||||
v-for="size in [100, 500, 1000, 5000]"
|
||||
v-for="size in MAX_ROW_OPTIONS"
|
||||
:key="size"
|
||||
:label="`最多 ${size} 行`"
|
||||
:value="size"
|
||||
@@ -876,19 +931,32 @@ defineExpose({ loadSpaces });
|
||||
|
||||
<template v-else-if="result">
|
||||
<div v-if="resultTab === 'result'" class="result-panel">
|
||||
<ElTable :data="result.rows" height="100%" stripe>
|
||||
<ElTableColumn type="index" width="56" />
|
||||
<ElTableColumn
|
||||
v-for="(column, index) in result.columns"
|
||||
:key="`${column.name}-${index}`"
|
||||
:label="column.name"
|
||||
min-width="160"
|
||||
>
|
||||
<template #default="{ row }">
|
||||
{{ row[index] ?? 'NULL' }}
|
||||
<ElEmpty
|
||||
v-if="virtualResultRows.length === 0"
|
||||
description="查询结果为空"
|
||||
/>
|
||||
<div
|
||||
v-else
|
||||
class="virtual-result-table"
|
||||
role="region"
|
||||
aria-label="SQL 查询结果"
|
||||
>
|
||||
<TableV2AutoResizer>
|
||||
<template #default="{ height, width }">
|
||||
<ElTableV2
|
||||
:columns="virtualResultColumns"
|
||||
:data="virtualResultRows"
|
||||
:header-height="RESULT_HEADER_HEIGHT"
|
||||
:height="height"
|
||||
:row-class="resultRowClass"
|
||||
:row-height="RESULT_ROW_HEIGHT"
|
||||
:width="width"
|
||||
fixed
|
||||
row-key="rowKey"
|
||||
/>
|
||||
</template>
|
||||
</ElTableColumn>
|
||||
</ElTable>
|
||||
</TableV2AutoResizer>
|
||||
</div>
|
||||
</div>
|
||||
<div v-else class="metrics-panel">
|
||||
<div
|
||||
@@ -1577,6 +1645,7 @@ defineExpose({ loadSpaces });
|
||||
display: grid;
|
||||
grid-template-rows: minmax(0, 1fr);
|
||||
padding: 0 var(--space-3) var(--space-3);
|
||||
overflow: hidden;
|
||||
background: hsl(var(--surface-subtle));
|
||||
}
|
||||
|
||||
@@ -1616,12 +1685,52 @@ defineExpose({ loadSpaces });
|
||||
background: hsl(var(--nav-item-hover));
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table) {
|
||||
font-size: 12px;
|
||||
.virtual-result-table {
|
||||
width: 100%;
|
||||
min-width: 0;
|
||||
min-height: 0;
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table__cell) {
|
||||
padding: var(--space-2) 0;
|
||||
.result-panel :deep(.el-table-v2) {
|
||||
font-size: 12px;
|
||||
color: hsl(var(--foreground));
|
||||
background: hsl(var(--surface-panel));
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table-v2__header-cell),
|
||||
.result-panel :deep(.el-table-v2__row-cell) {
|
||||
padding: 0 var(--space-2);
|
||||
border-right: 1px solid hsl(var(--divider-faint));
|
||||
border-bottom: 1px solid hsl(var(--divider-faint));
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table-v2__header-cell) {
|
||||
color: hsl(var(--text-strong));
|
||||
font-weight: 600;
|
||||
background: hsl(var(--surface-subtle));
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table-v2__row.is-striped) {
|
||||
background: hsl(var(--surface-contrast-soft) / 48%);
|
||||
}
|
||||
|
||||
.result-panel :deep(.el-table-v2__row:hover) {
|
||||
background: hsl(var(--table-row-hover));
|
||||
}
|
||||
|
||||
.result-panel :deep(.result-cell-value),
|
||||
.result-panel :deep(.result-index-cell) {
|
||||
display: block;
|
||||
min-width: 0;
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.result-panel :deep(.result-index-cell),
|
||||
.result-panel :deep(.result-cell-value.is-null) {
|
||||
color: hsl(var(--text-muted));
|
||||
}
|
||||
|
||||
.metrics-panel {
|
||||
|
||||
Reference in New Issue
Block a user