feat: 完善数据空间查询资源边界

- 将联邦源数与分片上限改为配置驱动

- 固化工作台一万行上限及结果字节保护
This commit is contained in:
2026-09-01 17:02:37 +08:00
parent 3e34b65bcb
commit c00369f6b9
10 changed files with 367 additions and 20 deletions

View File

@@ -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();
}
}

View File

@@ -2,6 +2,7 @@ package tech.easyflow.dataspace.config;
import org.mybatis.spring.annotation.MapperScan; import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.ComponentScan;
/** /**
@@ -10,5 +11,6 @@ import org.springframework.context.annotation.ComponentScan;
@MapperScan("tech.easyflow.dataspace.mapper") @MapperScan("tech.easyflow.dataspace.mapper")
@ComponentScan("tech.easyflow.dataspace") @ComponentScan("tech.easyflow.dataspace")
@AutoConfiguration @AutoConfiguration
@EnableConfigurationProperties({DataspaceFederationProperties.class, DataspaceQueryProperties.class})
public class DataspaceModuleConfig { public class DataspaceModuleConfig {
} }

View File

@@ -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");
}
}
}

View File

@@ -1,10 +1,11 @@
package tech.easyflow.dataspace.federation; package tech.easyflow.dataspace.federation;
import com.easyagents.federation.sql.adapter.jdbc.JdbcFederationSqlAdapterProvider; 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.FederationSqlEngine;
import com.easyagents.federation.sql.api.FederationSqlEngines; 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.api.FederationSqlException;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.source.FederationSourceDefinition; import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.FederationSourceView; import com.easyagents.federation.sql.source.FederationSourceView;
import com.easyagents.federation.sql.source.SourceApplyOptions; import com.easyagents.federation.sql.source.SourceApplyOptions;
@@ -15,6 +16,7 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
import tech.easyflow.dataspace.entity.DataspaceConnection; import tech.easyflow.dataspace.entity.DataspaceConnection;
/** /**
@@ -25,6 +27,7 @@ public class DataspaceFederationRuntime {
private final DataspaceDefinitionFactory definitionFactory; private final DataspaceDefinitionFactory definitionFactory;
private final FederationSqlEngine engine; private final FederationSqlEngine engine;
private final FederationExecutionPolicy executionPolicy;
private final ConcurrentMap<SourceId, FederationSourceDefinition> definitions = private final ConcurrentMap<SourceId, FederationSourceDefinition> definitions =
new ConcurrentHashMap<>(); new ConcurrentHashMap<>();
@@ -34,16 +37,20 @@ public class DataspaceFederationRuntime {
* @param resolver DataSource Resolver * @param resolver DataSource Resolver
* @param definitionFactory Definition 工厂 * @param definitionFactory Definition 工厂
* @param sqlPolicy 数据空间表白名单策略 * @param sqlPolicy 数据空间表白名单策略
* @param federationProperties 联邦资源上限
*/ */
public DataspaceFederationRuntime( public DataspaceFederationRuntime(
DataspaceDataSourceResolver resolver, DataspaceDataSourceResolver resolver,
DataspaceDefinitionFactory definitionFactory, DataspaceDefinitionFactory definitionFactory,
DataspaceSqlPolicy sqlPolicy) { DataspaceSqlPolicy sqlPolicy,
DataspaceFederationProperties federationProperties) {
this.definitionFactory = definitionFactory; this.definitionFactory = definitionFactory;
this.executionPolicy = federationProperties.toExecutionPolicy();
this.engine = FederationSqlEngines.builder() this.engine = FederationSqlEngines.builder()
.dataSourceResolver(resolver) .dataSourceResolver(resolver)
.adapter(new JdbcFederationSqlAdapterProvider()) .adapter(new JdbcFederationSqlAdapterProvider())
.policy(sqlPolicy) .policy(sqlPolicy)
.federationExecutionPolicy(executionPolicy)
.maximumPlanCacheEntries(1024) .maximumPlanCacheEntries(1024)
.crossSourceEnabled(true) .crossSourceEnabled(true)
.build(); .build();
@@ -125,6 +132,15 @@ public class DataspaceFederationRuntime {
return engine; return engine;
} }
/**
* 返回 Engine 与 Query Scope 共用的资源上限。
*
* @return 联邦执行策略
*/
public FederationExecutionPolicy executionPolicy() {
return executionPolicy;
}
/** /**
* 应用停止时关闭计划缓存与连接池。 * 应用停止时关闭计划缓存与连接池。
*/ */

View File

@@ -15,7 +15,6 @@ import com.easyagents.federation.sql.execute.FederationQueryMetricsSnapshot;
import com.easyagents.federation.sql.execute.FederationResultCursor; import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.QueryId; import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.SqlExecutionOptions; 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.FederationLogicalTableDefinition;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition; import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition; import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
@@ -48,6 +47,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import tech.easyflow.common.entity.LoginAccount; import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.web.exceptions.BusinessException; import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.dataspace.config.DataspaceQueryProperties;
import tech.easyflow.dataspace.entity.DataspaceConnection; import tech.easyflow.dataspace.entity.DataspaceConnection;
import tech.easyflow.dataspace.entity.DataspaceObject; import tech.easyflow.dataspace.entity.DataspaceObject;
import tech.easyflow.dataspace.entity.DataspaceQueryAudit; 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 Logger LOG = LoggerFactory.getLogger(DataspaceQueryService.class);
private static final int DEFAULT_MAX_ROWS = 500; 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 DEFAULT_TIMEOUT_SECONDS = 30;
private static final int MAX_TIMEOUT_SECONDS = 60; 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 DataspaceService dataspaceService;
private final DataspaceFederationRuntime runtime; private final DataspaceFederationRuntime runtime;
private final DataspaceDefinitionFactory definitionFactory; private final DataspaceDefinitionFactory definitionFactory;
private final DataspaceQueryAuditMapper auditMapper; private final DataspaceQueryAuditMapper auditMapper;
private final DataspaceQueryProperties queryProperties;
private final ConcurrentMap<String, ActiveQueryOwner> activeQueries = new ConcurrentHashMap<>(); private final ConcurrentMap<String, ActiveQueryOwner> activeQueries = new ConcurrentHashMap<>();
/** /**
@@ -89,16 +88,19 @@ public class DataspaceQueryService {
* @param runtime Federation Runtime * @param runtime Federation Runtime
* @param definitionFactory Definition 工厂 * @param definitionFactory Definition 工厂
* @param auditMapper 查询审计 Mapper * @param auditMapper 查询审计 Mapper
* @param queryProperties 同步查询结果保护配置
*/ */
public DataspaceQueryService( public DataspaceQueryService(
DataspaceService dataspaceService, DataspaceService dataspaceService,
DataspaceFederationRuntime runtime, DataspaceFederationRuntime runtime,
DataspaceDefinitionFactory definitionFactory, DataspaceDefinitionFactory definitionFactory,
DataspaceQueryAuditMapper auditMapper) { DataspaceQueryAuditMapper auditMapper,
DataspaceQueryProperties queryProperties) {
this.dataspaceService = dataspaceService; this.dataspaceService = dataspaceService;
this.runtime = runtime; this.runtime = runtime;
this.definitionFactory = definitionFactory; this.definitionFactory = definitionFactory;
this.auditMapper = auditMapper; this.auditMapper = auditMapper;
this.queryProperties = queryProperties;
} }
/** /**
@@ -112,7 +114,7 @@ public class DataspaceQueryService {
DataspaceService.RuntimeSnapshot snapshot = dataspaceService.loadSnapshot(request.dataspaceId()); DataspaceService.RuntimeSnapshot snapshot = dataspaceService.loadSnapshot(request.dataspaceId());
String executableSql = normalizeSql(request.sql()); String executableSql = normalizeSql(request.sql());
FederationQueryScopeDefinition scope = scopeSafely(snapshot, false); 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); int timeout = normalize(request.timeoutSeconds(), DEFAULT_TIMEOUT_SECONDS, MAX_TIMEOUT_SECONDS);
QueryId queryId = request.queryId() == null || request.queryId().isBlank() QueryId queryId = request.queryId() == null || request.queryId().isBlank()
? QueryId.create() : new QueryId(request.queryId().trim()); ? QueryId.create() : new QueryId(request.queryId().trim());
@@ -133,7 +135,8 @@ public class DataspaceQueryService {
.map(this::columnView).toList(); .map(this::columnView).toList();
List<List<Object>> rows = new ArrayList<>(); List<List<Object>> rows = new ArrayList<>();
DataspaceQueryResultBudget resultBudget = new DataspaceQueryResultBudget( DataspaceQueryResultBudget resultBudget = new DataspaceQueryResultBudget(
MAX_RESULT_BYTES, MAX_CELL_BYTES); queryProperties.getMaximumResultBytes().toBytes(),
queryProperties.getMaximumCellBytes().toBytes());
while (cursor.next()) { while (cursor.next()) {
resultBudget.beginRow(columns.size()); resultBudget.beginRow(columns.size());
List<Object> row = new ArrayList<>(columns.size()); List<Object> row = new ArrayList<>(columns.size());
@@ -329,7 +332,7 @@ public class DataspaceQueryService {
String defaultBinding = definitions.keySet().iterator().next(); String defaultBinding = definitions.keySet().iterator().next();
return FederationQueryScopeDefinition.virtual( return FederationQueryScopeDefinition.virtual(
scopeId(snapshot), snapshot.revision().getRevisionNo(), definitions, 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 { private String readText(Reader reader) throws Exception {
try (reader) { try (reader) {
long maximumCellBytes = queryProperties.getMaximumCellBytes().toBytes();
char[] buffer = new char[8_192]; char[] buffer = new char[8_192];
StringBuilder value = new StringBuilder(); StringBuilder value = new StringBuilder();
int count; int count;
while ((count = reader.read(buffer)) >= 0) { while ((count = reader.read(buffer)) >= 0) {
value.append(buffer, 0, count); value.append(buffer, 0, count);
if (value.length() > MAX_CELL_BYTES) { if (value.length() > maximumCellBytes) {
throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception( throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
"单个文本字段超过 1 MiB 限制"); "单个文本字段超过配置的字节限制");
} }
} }
return value.toString(); return value.toString();
@@ -526,10 +530,12 @@ public class DataspaceQueryService {
*/ */
private byte[] readBytes(InputStream stream) throws Exception { private byte[] readBytes(InputStream stream) throws Exception {
try (stream) { try (stream) {
byte[] bytes = stream.readNBytes((int) MAX_CELL_BYTES + 1); int maximumCellBytes = Math.toIntExact(
if (bytes.length > MAX_CELL_BYTES) { queryProperties.getMaximumCellBytes().toBytes());
byte[] bytes = stream.readNBytes(maximumCellBytes + 1);
if (bytes.length > maximumCellBytes) {
throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception( throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
"单个二进制字段超过 1 MiB 限制"); "单个二进制字段超过配置的字节限制");
} }
return bytes; return bytes;
} }
@@ -651,6 +657,22 @@ public class DataspaceQueryService {
return Math.min(normalized, maximum); return Math.min(normalized, maximum);
} }
/**
* 规范化工作台最大返回行数,并拒绝绕过产品上限的直接 API 请求。
*
* @param value 调用方值
* @return 允许执行的最大返回行数
* @throws BusinessException 值不在 110,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(
"最大返回行数必须在 110,000 之间");
}
return normalized;
}
/** /**
* 将纳秒转换为毫秒并保留未知值。 * 将纳秒转换为毫秒并保留未知值。
* *

View File

@@ -24,6 +24,7 @@ import tech.easyflow.common.cache.RedisLockExecutor;
import tech.easyflow.common.entity.LoginAccount; import tech.easyflow.common.entity.LoginAccount;
import tech.easyflow.common.util.SearchKeywordUtil; import tech.easyflow.common.util.SearchKeywordUtil;
import tech.easyflow.common.web.exceptions.BusinessException; import tech.easyflow.common.web.exceptions.BusinessException;
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
import tech.easyflow.dataspace.entity.Dataspace; import tech.easyflow.dataspace.entity.Dataspace;
import tech.easyflow.dataspace.entity.DataspaceConnection; import tech.easyflow.dataspace.entity.DataspaceConnection;
import tech.easyflow.dataspace.entity.DataspaceObject; 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_WAIT = Duration.ofSeconds(3);
private static final Duration LOCK_LEASE = Duration.ofSeconds(30); private static final Duration LOCK_LEASE = Duration.ofSeconds(30);
private static final int MAX_TABLES = 100; private static final int MAX_TABLES = 100;
private static final int MAX_SOURCES = 2;
private final DataspaceMapper dataspaceMapper; private final DataspaceMapper dataspaceMapper;
private final DataspaceRevisionMapper revisionMapper; private final DataspaceRevisionMapper revisionMapper;
private final DataspaceTableBindingMapper bindingMapper; private final DataspaceTableBindingMapper bindingMapper;
@@ -64,6 +64,7 @@ public class DataspaceService {
private final DataspaceMetadataResolver metadataResolver; private final DataspaceMetadataResolver metadataResolver;
private final RedisLockExecutor lockExecutor; private final RedisLockExecutor lockExecutor;
private final TransactionTemplate transactionTemplate; private final TransactionTemplate transactionTemplate;
private final DataspaceFederationProperties federationProperties;
/** /**
* 创建数据空间服务。 * 创建数据空间服务。
@@ -78,6 +79,7 @@ public class DataspaceService {
* @param metadataResolver 元数据兼容解析器 * @param metadataResolver 元数据兼容解析器
* @param lockExecutor 分布式锁执行器 * @param lockExecutor 分布式锁执行器
* @param transactionTemplate 事务模板 * @param transactionTemplate 事务模板
* @param federationProperties 联邦执行与建模上限
*/ */
public DataspaceService( public DataspaceService(
DataspaceMapper dataspaceMapper, DataspaceMapper dataspaceMapper,
@@ -89,7 +91,8 @@ public class DataspaceService {
DataspaceDatabaseProviderRegistry providerRegistry, DataspaceDatabaseProviderRegistry providerRegistry,
DataspaceMetadataResolver metadataResolver, DataspaceMetadataResolver metadataResolver,
RedisLockExecutor lockExecutor, RedisLockExecutor lockExecutor,
TransactionTemplate transactionTemplate) { TransactionTemplate transactionTemplate,
DataspaceFederationProperties federationProperties) {
this.dataspaceMapper = dataspaceMapper; this.dataspaceMapper = dataspaceMapper;
this.revisionMapper = revisionMapper; this.revisionMapper = revisionMapper;
this.bindingMapper = bindingMapper; this.bindingMapper = bindingMapper;
@@ -100,6 +103,7 @@ public class DataspaceService {
this.metadataResolver = metadataResolver; this.metadataResolver = metadataResolver;
this.lockExecutor = lockExecutor; this.lockExecutor = lockExecutor;
this.transactionTemplate = transactionTemplate; this.transactionTemplate = transactionTemplate;
this.federationProperties = federationProperties;
} }
/** /**
@@ -448,9 +452,10 @@ public class DataspaceService {
} }
Set<BigInteger> connectionIds = objects.values().stream() Set<BigInteger> connectionIds = objects.values().stream()
.map(DataspaceObject::getConnectionId).collect(java.util.stream.Collectors.toSet()); .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( throw DataspaceErrorCode.DEFINITION_INVALID.exception(
"当前阶段单个数据空间最多支持个物理数据源"); "单个数据空间最多支持 " + maximumSources + " 个物理数据源");
} }
Map<BigInteger, DataspaceConnection> connections = new HashMap<>(); Map<BigInteger, DataspaceConnection> connections = new HashMap<>();
connectionMapper.selectListByQuery(QueryWrapper.create().in(DataspaceConnection::getId, connectionIds)) connectionMapper.selectListByQuery(QueryWrapper.create().in(DataspaceConnection::getId, connectionIds))

View File

@@ -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();
}
}

View File

@@ -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();
}
}

View File

@@ -16,6 +16,7 @@ import java.util.List;
import java.util.Map; import java.util.Map;
import org.junit.Test; import org.junit.Test;
import tech.easyflow.common.cache.RedisLockExecutor; import tech.easyflow.common.cache.RedisLockExecutor;
import tech.easyflow.dataspace.config.DataspaceFederationProperties;
import tech.easyflow.dataspace.entity.Dataspace; import tech.easyflow.dataspace.entity.Dataspace;
import tech.easyflow.dataspace.entity.DataspaceConnection; import tech.easyflow.dataspace.entity.DataspaceConnection;
import tech.easyflow.dataspace.entity.DataspaceObject; import tech.easyflow.dataspace.entity.DataspaceObject;
@@ -183,7 +184,8 @@ public class DataspaceMetadataCompatibilityTest {
mock(DataspaceDatabaseProviderRegistry.class), mock(DataspaceDatabaseProviderRegistry.class),
resolver, resolver,
mock(RedisLockExecutor.class), mock(RedisLockExecutor.class),
mock(org.springframework.transaction.support.TransactionTemplate.class)); mock(org.springframework.transaction.support.TransactionTemplate.class),
new DataspaceFederationProperties());
return new Fixture(service); return new Fixture(service);
} }

View File

@@ -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());
}
}
}