From c00369f6b92a4eb4907bde6b955d25aec1b185a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Tue, 1 Sep 2026 17:02:37 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E5=96=84=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=A9=BA=E9=97=B4=E6=9F=A5=E8=AF=A2=E8=B5=84=E6=BA=90=E8=BE=B9?= =?UTF-8?q?=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将联邦源数与分片上限改为配置驱动 - 固化工作台一万行上限及结果字节保护 --- .../config/DataspaceFederationProperties.java | 94 +++++++++++++++++++ .../config/DataspaceModuleConfig.java | 2 + .../config/DataspaceQueryProperties.java | 45 +++++++++ .../DataspaceFederationRuntime.java | 20 +++- .../service/DataspaceQueryService.java | 48 +++++++--- .../dataspace/service/DataspaceService.java | 13 ++- .../DataspaceFederationPropertiesTest.java | 62 ++++++++++++ .../config/DataspaceQueryPropertiesTest.java | 52 ++++++++++ .../DataspaceMetadataCompatibilityTest.java | 4 +- .../service/DataspaceQueryRowLimitTest.java | 47 ++++++++++ 10 files changed, 367 insertions(+), 20 deletions(-) create mode 100644 easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceFederationProperties.java create mode 100644 easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceQueryProperties.java create mode 100644 easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceFederationPropertiesTest.java create mode 100644 easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceQueryPropertiesTest.java create mode 100644 easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceQueryRowLimitTest.java diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceFederationProperties.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceFederationProperties.java new file mode 100644 index 00000000..474fa83c --- /dev/null +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceFederationProperties.java @@ -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(); + } +} diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceModuleConfig.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceModuleConfig.java index 7a348f77..16131d6f 100644 --- a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceModuleConfig.java +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceModuleConfig.java @@ -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 { } diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceQueryProperties.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceQueryProperties.java new file mode 100644 index 00000000..45329347 --- /dev/null +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/config/DataspaceQueryProperties.java @@ -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"); + } + } +} diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/federation/DataspaceFederationRuntime.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/federation/DataspaceFederationRuntime.java index 549bba60..a00a15fc 100644 --- a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/federation/DataspaceFederationRuntime.java +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/federation/DataspaceFederationRuntime.java @@ -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 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; + } + /** * 应用停止时关闭计划缓存与连接池。 */ diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryService.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryService.java index 129d5db8..08fb87ba 100644 --- a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryService.java +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryService.java @@ -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 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> 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 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; + } + /** * 将纳秒转换为毫秒并保留未知值。 * diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceService.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceService.java index 672bcf2f..f44bb1e2 100644 --- a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceService.java +++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceService.java @@ -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 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 connections = new HashMap<>(); connectionMapper.selectListByQuery(QueryWrapper.create().in(DataspaceConnection::getId, connectionIds)) diff --git a/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceFederationPropertiesTest.java b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceFederationPropertiesTest.java new file mode 100644 index 00000000..3b7aa3fc --- /dev/null +++ b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceFederationPropertiesTest.java @@ -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(); + } +} diff --git a/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceQueryPropertiesTest.java b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceQueryPropertiesTest.java new file mode 100644 index 00000000..3ad9e90f --- /dev/null +++ b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/config/DataspaceQueryPropertiesTest.java @@ -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(); + } +} diff --git a/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceMetadataCompatibilityTest.java b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceMetadataCompatibilityTest.java index bfb2fc0f..c4d281bc 100644 --- a/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceMetadataCompatibilityTest.java +++ b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceMetadataCompatibilityTest.java @@ -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); } diff --git a/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceQueryRowLimitTest.java b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceQueryRowLimitTest.java new file mode 100644 index 00000000..e6b7d476 --- /dev/null +++ b/easyflow-modules/easyflow-module-dataspace/src/test/java/tech/easyflow/dataspace/service/DataspaceQueryRowLimitTest.java @@ -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()); + } + } +}