feat: 新增统一数据空间与联邦查询能力
- 提供数据连接、元数据、逻辑表与关联编排能力 - 接入联邦查询、执行分析、统计估算和 SQL 补全接口 - 增加查询预算、凭据保护、租户隔离和 V61 初始化迁移
This commit is contained in:
@@ -0,0 +1,50 @@
|
||||
package tech.easyflow.dataspace.model;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* 数据空间保存请求反序列化测试。
|
||||
*/
|
||||
public class DataspaceDefinitionTest {
|
||||
|
||||
/**
|
||||
* 验证 Fastjson 反序列化 record 时遗留的嵌套 JSON 对象会恢复为强类型定义。
|
||||
*/
|
||||
@Test
|
||||
public void shouldNormalizeNestedDefinitionsAfterFastjsonDeserialization() {
|
||||
String json = """
|
||||
{
|
||||
"id": "448492737717747712",
|
||||
"expectedRevision": 1,
|
||||
"name": "网点经营分析",
|
||||
"description": "",
|
||||
"tables": [{
|
||||
"clientKey": "binding:1",
|
||||
"objectId": "448679746354769926",
|
||||
"sourceAlias": "MYSQL_1",
|
||||
"schemaAlias": "MAIN",
|
||||
"tableAlias": "outlet_main",
|
||||
"positionX": 80,
|
||||
"positionY": 80
|
||||
}],
|
||||
"relations": [{
|
||||
"leftClientKey": "binding:1",
|
||||
"rightClientKey": "binding:2",
|
||||
"joinType": "INNER",
|
||||
"leftColumn": "source_row",
|
||||
"rightColumn": "source_row"
|
||||
}]
|
||||
}
|
||||
""";
|
||||
|
||||
DataspaceDefinition definition = JSON.parseObject(json, DataspaceDefinition.class);
|
||||
|
||||
assertEquals("outlet_main", definition.tables().get(0).tableAlias());
|
||||
assertEquals("source_row", definition.relations().get(0).leftColumn());
|
||||
assertTrue(definition.tables().get(0) instanceof DataspaceDefinition.TableDefinition);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
package tech.easyflow.dataspace.model;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertSame;
|
||||
|
||||
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
|
||||
import com.easyagents.federation.sql.api.FederationSqlException;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.dataspace.security.DataspaceCredentialUnavailableException;
|
||||
|
||||
/**
|
||||
* 数据空间错误码映射契约测试。
|
||||
*/
|
||||
public class DataspaceErrorCodeTest {
|
||||
|
||||
/**
|
||||
* 验证 SQL 语法错误保留 HTTP 400 和稳定业务码。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapSqlParseFailure() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SQL_PARSE_FAILED,
|
||||
"Encountered token at line 1"),
|
||||
false);
|
||||
|
||||
assertEquals(400, exception.getHttpStatus());
|
||||
assertEquals(40061, exception.getErrorCode());
|
||||
assertEquals(true, exception.getMessage().contains("SQL 语法解析失败"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证未知逻辑表返回可直接处理的中文提示。
|
||||
*/
|
||||
@Test
|
||||
public void shouldLocalizeUnknownLogicalTable() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SQL_VALIDATION_FAILED,
|
||||
"logical table is not declared in the query scope: missing_table"),
|
||||
false);
|
||||
|
||||
assertEquals(400, exception.getHttpStatus());
|
||||
assertEquals(40062, exception.getErrorCode());
|
||||
assertEquals(true, exception.getMessage().contains("逻辑表 “missing_table” 不在当前数据空间中"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证数据源初始化失败映射为可恢复的 503 错误。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapSourceInitializationFailure() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED,
|
||||
"driver failure"),
|
||||
false);
|
||||
|
||||
assertEquals(503, exception.getHttpStatus());
|
||||
assertEquals(50361, exception.getErrorCode());
|
||||
assertEquals(DataspaceErrorCode.SOURCE_UNAVAILABLE.defaultMessage(), exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证数据源初始化包装后仍保留凭据失效的精确恢复提示。
|
||||
*/
|
||||
@Test
|
||||
public void shouldPreserveCredentialUnavailableFailure() {
|
||||
DataspaceCredentialUnavailableException credentialException =
|
||||
new DataspaceCredentialUnavailableException(
|
||||
new IllegalArgumentException("credential key mismatch"));
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED,
|
||||
"failed to initialize source",
|
||||
credentialException),
|
||||
false);
|
||||
|
||||
assertSame(credentialException, exception);
|
||||
assertEquals(409, exception.getHttpStatus());
|
||||
assertEquals(40905, exception.getErrorCode());
|
||||
assertEquals(DataspaceCredentialUnavailableException.USER_MESSAGE,
|
||||
exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证联邦资源限制映射为 422，便于页面提示缩小查询范围。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapFederationResourceLimit() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
|
||||
"too many rows"),
|
||||
false);
|
||||
|
||||
assertEquals(422, exception.getHttpStatus());
|
||||
assertEquals(42261, exception.getErrorCode());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证补全内核异常使用独立错误码，避免伪装成数据源不可用。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapSqlCompletionFailure() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SQL_COMPLETION_FAILED,
|
||||
"advisor failed"),
|
||||
false);
|
||||
|
||||
assertEquals(503, exception.getHttpStatus());
|
||||
assertEquals(50364, exception.getErrorCode());
|
||||
assertEquals(DataspaceErrorCode.SQL_COMPLETION_FAILED.defaultMessage(),
|
||||
exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 SQL 编译超时使用独立错误码和可恢复提示。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapSqlCompileTimeout() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.SQL_COMPILE_TIMEOUT,
|
||||
"plan cache wait expired"),
|
||||
false);
|
||||
|
||||
assertEquals(408, exception.getHttpStatus());
|
||||
assertEquals(40863, exception.getErrorCode());
|
||||
assertEquals(DataspaceErrorCode.SQL_COMPILE_TIMEOUT.defaultMessage(),
|
||||
exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证连接池等待超时不会被泛化为数据源不可用。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapConnectionAcquisitionTimeout() {
|
||||
BusinessException exception = DataspaceErrorCode.fromFederation(
|
||||
new FederationSqlException(
|
||||
FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT,
|
||||
"pool acquisition timeout"),
|
||||
false);
|
||||
|
||||
assertEquals(503, exception.getHttpStatus());
|
||||
assertEquals(50365, exception.getErrorCode());
|
||||
assertEquals(DataspaceErrorCode.CONNECTION_ACQUISITION_TIMEOUT.defaultMessage(),
|
||||
exception.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package tech.easyflow.dataspace.provider;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import org.junit.Assume;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* 本机 Docker MySQL 与 PostgreSQL 元数据接入验证。
|
||||
*/
|
||||
public class DataspaceDatabaseProviderIntegrationTest {
|
||||
|
||||
/**
|
||||
* 验证 MySQL data-sheet 元数据可被读取。
|
||||
*
|
||||
* @throws Exception JDBC 连接或元数据读取失败
|
||||
*/
|
||||
@Test
|
||||
public void shouldInspectMysqlDataSheet() throws Exception {
|
||||
Assume.assumeTrue(Boolean.getBoolean("dataspace.integration.enabled"));
|
||||
MySqlProvider provider = new MySqlProvider();
|
||||
DataspaceConnectionConfig config = new DataspaceConnectionConfig(
|
||||
DataspaceDatabaseType.MYSQL,
|
||||
System.getProperty("dataspace.mysql.host", "127.0.0.1"),
|
||||
Integer.getInteger("dataspace.mysql.port", 33306),
|
||||
System.getProperty("dataspace.mysql.database", "data-sheet"),
|
||||
System.getProperty("dataspace.mysql.username", "root"),
|
||||
System.getProperty("dataspace.mysql.password", "root"),
|
||||
false,
|
||||
Map.of());
|
||||
|
||||
try (Connection connection = DriverManager.getConnection(
|
||||
provider.jdbcUrl(config), config.username(), config.password())) {
|
||||
List<DataspaceObjectMetadata> objects = provider.inspect(connection, config);
|
||||
assertFalse(objects.isEmpty());
|
||||
assertTrue(objects.stream().anyMatch(object -> "outlet".equalsIgnoreCase(object.name())));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 PostgreSQL public Schema 元数据可被读取。
|
||||
*
|
||||
* @throws Exception JDBC 连接或元数据读取失败
|
||||
*/
|
||||
@Test
|
||||
public void shouldInspectPostgresql() throws Exception {
|
||||
Assume.assumeTrue(Boolean.getBoolean("dataspace.integration.enabled"));
|
||||
PostgresqlProvider provider = new PostgresqlProvider();
|
||||
DataspaceConnectionConfig config = new DataspaceConnectionConfig(
|
||||
DataspaceDatabaseType.POSTGRESQL,
|
||||
System.getProperty("dataspace.pg.host", "127.0.0.1"),
|
||||
Integer.getInteger("dataspace.pg.port", 54329),
|
||||
System.getProperty("dataspace.pg.database", "harmony_adapter"),
|
||||
System.getProperty("dataspace.pg.username", "harmony"),
|
||||
System.getProperty("dataspace.pg.password", "harmony"),
|
||||
false,
|
||||
Map.of());
|
||||
|
||||
try (Connection connection = DriverManager.getConnection(
|
||||
provider.jdbcUrl(config), config.username(), config.password())) {
|
||||
List<DataspaceObjectMetadata> objects = provider.inspect(connection, config);
|
||||
assertFalse(objects.isEmpty());
|
||||
assertTrue(objects.stream().allMatch(object -> !object.schema().startsWith("pg_")));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package tech.easyflow.dataspace.provider;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertThrows;
|
||||
|
||||
import java.util.Map;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* MySQL 与 PostgreSQL Provider 契约测试。
|
||||
*/
|
||||
public class DataspaceDatabaseProviderTest {
|
||||
|
||||
/**
|
||||
* 验证 MySQL URL 与固定逻辑 Schema。
|
||||
*/
|
||||
@Test
|
||||
public void shouldBuildMysqlDefinition() {
|
||||
MySqlProvider provider = new MySqlProvider();
|
||||
DataspaceConnectionConfig config = new DataspaceConnectionConfig(
|
||||
DataspaceDatabaseType.MYSQL,
|
||||
"127.0.0.1",
|
||||
33306,
|
||||
"data-sheet",
|
||||
"root",
|
||||
"root",
|
||||
false,
|
||||
Map.of());
|
||||
|
||||
assertEquals("MAIN", provider.logicalSchema(null));
|
||||
assertEquals(
|
||||
"jdbc:mysql://127.0.0.1:33306/data-sheet"
|
||||
+ "?useUnicode=true&characterEncoding=utf8&useSSL=false"
|
||||
+ "&serverTimezone=UTC&allowMultiQueries=false",
|
||||
provider.jdbcUrl(config));
|
||||
assertFalse(provider.jdbcUrl(config).contains("allowMultiQueries=true"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 PostgreSQL URL、SSL 策略与 Schema 规范化。
|
||||
*/
|
||||
@Test
|
||||
public void shouldBuildPostgresqlDefinition() {
|
||||
PostgresqlProvider provider = new PostgresqlProvider();
|
||||
DataspaceConnectionConfig config = new DataspaceConnectionConfig(
|
||||
DataspaceDatabaseType.POSTGRESQL,
|
||||
"127.0.0.1",
|
||||
54329,
|
||||
"harmony_adapter",
|
||||
"harmony",
|
||||
"harmony",
|
||||
false,
|
||||
Map.of());
|
||||
|
||||
assertEquals("PUBLIC", provider.logicalSchema(null));
|
||||
assertEquals("SALES", provider.logicalSchema("sales"));
|
||||
assertEquals(
|
||||
"jdbc:postgresql://127.0.0.1:54329/harmony_adapter"
|
||||
+ "?sslmode=disable&ApplicationName=easyflow-dataspace",
|
||||
provider.jdbcUrl(config));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证未知数据库类型被显式拒绝。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectUnknownDatabaseType() {
|
||||
assertThrows(
|
||||
IllegalArgumentException.class,
|
||||
() -> DataspaceDatabaseType.parse("oracle"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package tech.easyflow.dataspace.security;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotEquals;
|
||||
import static org.junit.Assert.assertThrows;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* 数据空间凭据认证加密测试。
|
||||
*/
|
||||
public class DataspaceCredentialCipherTest {
|
||||
|
||||
private static final String MASTER_KEY = "dataspace-test-master-key-32-characters";
|
||||
|
||||
/**
|
||||
* 验证同一明文使用随机 IV，并能正确解密。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRoundTripWithRandomIv() {
|
||||
DataspaceCredentialCipher cipher = new DataspaceCredentialCipher(MASTER_KEY);
|
||||
|
||||
String first = cipher.encrypt("root");
|
||||
String second = cipher.encrypt("root");
|
||||
|
||||
assertNotEquals(first, second);
|
||||
assertEquals("root", cipher.decrypt(first));
|
||||
assertEquals("root", cipher.decrypt(second));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证认证标签能够拒绝被篡改的密文。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectTamperedCipherText() {
|
||||
DataspaceCredentialCipher cipher = new DataspaceCredentialCipher(MASTER_KEY);
|
||||
String encrypted = cipher.encrypt("root");
|
||||
char replacement = encrypted.endsWith("A") ? 'B' : 'A';
|
||||
String tampered = encrypted.substring(0, encrypted.length() - 1) + replacement;
|
||||
|
||||
DataspaceCredentialUnavailableException exception = assertThrows(
|
||||
DataspaceCredentialUnavailableException.class,
|
||||
() -> cipher.decrypt(tampered));
|
||||
|
||||
assertEquals(409, exception.getHttpStatus());
|
||||
assertEquals(DataspaceCredentialUnavailableException.ERROR_CODE, exception.getErrorCode());
|
||||
assertEquals(DataspaceCredentialUnavailableException.USER_MESSAGE, exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证部署密钥变更后返回可恢复的稳定业务错误。
|
||||
*/
|
||||
@Test
|
||||
public void shouldDescribeCredentialRecoveryWhenMasterKeyChanges() {
|
||||
DataspaceCredentialCipher original = new DataspaceCredentialCipher(MASTER_KEY);
|
||||
DataspaceCredentialCipher changed = new DataspaceCredentialCipher(
|
||||
"changed-dataspace-master-key-32-characters");
|
||||
|
||||
DataspaceCredentialUnavailableException exception = assertThrows(
|
||||
DataspaceCredentialUnavailableException.class,
|
||||
() -> changed.decrypt(original.encrypt("root")));
|
||||
|
||||
assertEquals(409, exception.getHttpStatus());
|
||||
assertEquals(DataspaceCredentialUnavailableException.ERROR_CODE, exception.getErrorCode());
|
||||
assertEquals(DataspaceCredentialUnavailableException.USER_MESSAGE, exception.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证部署主密钥长度下限。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectShortMasterKey() {
|
||||
assertThrows(
|
||||
IllegalStateException.class,
|
||||
() -> new DataspaceCredentialCipher("too-short"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,326 @@
|
||||
package tech.easyflow.dataspace.service;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import com.mybatisflex.core.query.QueryWrapper;
|
||||
import java.math.BigInteger;
|
||||
import java.sql.Types;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.common.cache.RedisLockExecutor;
|
||||
import tech.easyflow.dataspace.entity.Dataspace;
|
||||
import tech.easyflow.dataspace.entity.DataspaceConnection;
|
||||
import tech.easyflow.dataspace.entity.DataspaceObject;
|
||||
import tech.easyflow.dataspace.entity.DataspaceRelation;
|
||||
import tech.easyflow.dataspace.entity.DataspaceRevision;
|
||||
import tech.easyflow.dataspace.entity.DataspaceTableBinding;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceConnectionMapper;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceMapper;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceObjectMapper;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceRelationMapper;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceRevisionMapper;
|
||||
import tech.easyflow.dataspace.mapper.DataspaceTableBindingMapper;
|
||||
import tech.easyflow.dataspace.model.DataspaceErrorCode;
|
||||
import tech.easyflow.dataspace.model.DataspaceView;
|
||||
import tech.easyflow.dataspace.provider.DataspaceDatabaseProviderRegistry;
|
||||
|
||||
/**
|
||||
* 数据空间历史绑定与最新元数据兼容解析测试。
|
||||
*/
|
||||
public class DataspaceMetadataCompatibilityTest {
|
||||
|
||||
private static final BigInteger SPACE_ID = BigInteger.valueOf(1L);
|
||||
private static final BigInteger REVISION_ID = BigInteger.valueOf(2L);
|
||||
private static final BigInteger BINDING_ID = BigInteger.valueOf(3L);
|
||||
private static final BigInteger CONNECTION_ID = BigInteger.valueOf(4L);
|
||||
private static final BigInteger HISTORICAL_OBJECT_ID = BigInteger.valueOf(5L);
|
||||
private static final BigInteger CURRENT_OBJECT_ID = BigInteger.valueOf(6L);
|
||||
private static final BigInteger SECOND_BINDING_ID = BigInteger.valueOf(7L);
|
||||
private static final BigInteger SECOND_HISTORICAL_OBJECT_ID = BigInteger.valueOf(8L);
|
||||
private static final BigInteger SECOND_CURRENT_OBJECT_ID = BigInteger.valueOf(9L);
|
||||
private static final BigInteger RELATION_ID = BigInteger.valueOf(10L);
|
||||
|
||||
/**
|
||||
* 验证逻辑表名兼容产品默认使用的小写名称和手动下划线命名。
|
||||
*/
|
||||
@Test
|
||||
public void shouldAcceptLowercaseLogicalTableNameWithUnderscore() {
|
||||
assertTrue(DataspaceService.isIdentifier("outlet_main"));
|
||||
assertTrue(DataspaceService.isIdentifier("OUTLET_2"));
|
||||
assertFalse(DataspaceService.isIdentifier("2_outlet"));
|
||||
assertFalse(DataspaceService.isIdentifier("outlet-name"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证兼容元数据刷新后详情自动使用最新对象和字段。
|
||||
*/
|
||||
@Test
|
||||
public void shouldResolveHistoricalBindingToCurrentMetadata() {
|
||||
Fixture fixture = fixture(List.of(currentObject()));
|
||||
|
||||
DataspaceView detail = fixture.service().detail(SPACE_ID);
|
||||
|
||||
assertEquals(1, detail.tables().size());
|
||||
assertEquals(CURRENT_OBJECT_ID, detail.tables().get(0).objectId());
|
||||
assertEquals("MYSQL", detail.tables().get(0).databaseType());
|
||||
assertEquals("ACTIVE", detail.tables().get(0).status());
|
||||
assertEquals(2, detail.tables().get(0).columns().size());
|
||||
assertNull(detail.tables().get(0).issueCode());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证表从最新快照消失时仅返回局部失效状态，不阻断详情加载。
|
||||
*/
|
||||
@Test
|
||||
public void shouldExposeMissingBindingWithoutRejectingWholeDataspace() {
|
||||
Fixture fixture = fixture(List.of());
|
||||
|
||||
DataspaceView detail = fixture.service().detail(SPACE_ID);
|
||||
|
||||
assertEquals(1, detail.tables().size());
|
||||
assertEquals(HISTORICAL_OBJECT_ID, detail.tables().get(0).objectId());
|
||||
assertEquals("MISSING", detail.tables().get(0).status());
|
||||
assertEquals(
|
||||
Integer.valueOf(DataspaceErrorCode.OBJECT_MISSING.code()),
|
||||
detail.tables().get(0).issueCode());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证关联字段被删除时仅标记该关系失效，其他表仍可继续查询。
|
||||
*/
|
||||
@Test
|
||||
public void shouldInvalidateOnlyRelationWhenJoinColumnDisappears() {
|
||||
DataspaceObject historicalRegion = object(
|
||||
SECOND_HISTORICAL_OBJECT_ID,
|
||||
1L,
|
||||
"outlet_region",
|
||||
List.of(column("outlet_id", Types.BIGINT)));
|
||||
DataspaceObject currentRegion = object(
|
||||
SECOND_CURRENT_OBJECT_ID,
|
||||
2L,
|
||||
"outlet_region",
|
||||
List.of(column("region_name", Types.VARCHAR)));
|
||||
Fixture fixture = fixture(
|
||||
List.of(historicalObject(), historicalRegion),
|
||||
List.of(binding(), secondBinding()),
|
||||
List.of(relation()),
|
||||
List.of(currentObject(), currentRegion));
|
||||
|
||||
DataspaceView detail = fixture.service().detail(SPACE_ID);
|
||||
|
||||
assertEquals("ACTIVE", detail.tables().get(0).status());
|
||||
assertEquals("ACTIVE", detail.tables().get(1).status());
|
||||
assertEquals("INVALID", detail.relations().get(0).status());
|
||||
assertEquals(
|
||||
Integer.valueOf(DataspaceErrorCode.RELATION_FIELD_MISSING.code()),
|
||||
detail.relations().get(0).issueCode());
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建只覆盖详情读取路径的测试夹具。
|
||||
*
|
||||
* @param currentObjects 最新元数据对象
|
||||
* @return 测试夹具
|
||||
*/
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
private Fixture fixture(List<DataspaceObject> currentObjects) {
|
||||
return fixture(
|
||||
List.of(historicalObject()),
|
||||
List.of(binding()),
|
||||
List.of(),
|
||||
currentObjects);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建指定历史与当前元数据的测试夹具。
|
||||
*
|
||||
* @param historicalObjects revision 保存的历史对象
|
||||
* @param bindings revision 表绑定
|
||||
* @param relations revision 表关系
|
||||
* @param currentObjects 最新元数据对象
|
||||
* @return 测试夹具
|
||||
*/
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
private Fixture fixture(
|
||||
List<DataspaceObject> historicalObjects,
|
||||
List<DataspaceTableBinding> bindings,
|
||||
List<DataspaceRelation> relations,
|
||||
List<DataspaceObject> currentObjects) {
|
||||
DataspaceMapper dataspaceMapper = mock(DataspaceMapper.class);
|
||||
DataspaceRevisionMapper revisionMapper = mock(DataspaceRevisionMapper.class);
|
||||
DataspaceTableBindingMapper bindingMapper = mock(DataspaceTableBindingMapper.class);
|
||||
DataspaceRelationMapper relationMapper = mock(DataspaceRelationMapper.class);
|
||||
DataspaceObjectMapper objectMapper = mock(DataspaceObjectMapper.class);
|
||||
DataspaceConnectionMapper connectionMapper = mock(DataspaceConnectionMapper.class);
|
||||
|
||||
when(dataspaceMapper.selectOneById(SPACE_ID)).thenReturn(dataspace());
|
||||
when(revisionMapper.selectOneByQuery(any(QueryWrapper.class))).thenReturn(revision());
|
||||
when(bindingMapper.selectListByQuery(any(QueryWrapper.class)))
|
||||
.thenReturn(bindings);
|
||||
when(relationMapper.selectListByQuery(any(QueryWrapper.class))).thenReturn(relations);
|
||||
// 首次读取不可变历史对象，随后由解析器读取连接当前快照。
|
||||
when(objectMapper.selectListByQuery(any(QueryWrapper.class)))
|
||||
.thenReturn(historicalObjects, currentObjects);
|
||||
when(connectionMapper.selectListByQuery(any(QueryWrapper.class)))
|
||||
.thenReturn(List.of(connection()));
|
||||
|
||||
DataspaceMetadataResolver resolver = new DataspaceMetadataResolver(objectMapper);
|
||||
DataspaceService service = new DataspaceService(
|
||||
dataspaceMapper,
|
||||
revisionMapper,
|
||||
bindingMapper,
|
||||
relationMapper,
|
||||
objectMapper,
|
||||
connectionMapper,
|
||||
mock(DataspaceDatabaseProviderRegistry.class),
|
||||
resolver,
|
||||
mock(RedisLockExecutor.class),
|
||||
mock(org.springframework.transaction.support.TransactionTemplate.class));
|
||||
return new Fixture(service);
|
||||
}
|
||||
|
||||
/** @return 数据空间记录 */
|
||||
private Dataspace dataspace() {
|
||||
Dataspace value = new Dataspace();
|
||||
value.setId(SPACE_ID);
|
||||
value.setName("网点经营分析");
|
||||
value.setDescription("");
|
||||
value.setCurrentRevision(1L);
|
||||
value.setStatus("ENABLED");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 当前 revision */
|
||||
private DataspaceRevision revision() {
|
||||
DataspaceRevision value = new DataspaceRevision();
|
||||
value.setId(REVISION_ID);
|
||||
value.setDataspaceId(SPACE_ID);
|
||||
value.setRevisionNo(1L);
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 历史表绑定 */
|
||||
private DataspaceTableBinding binding() {
|
||||
DataspaceTableBinding value = new DataspaceTableBinding();
|
||||
value.setId(BINDING_ID);
|
||||
value.setDataspaceId(SPACE_ID);
|
||||
value.setRevisionId(REVISION_ID);
|
||||
value.setConnectionId(CONNECTION_ID);
|
||||
value.setSourceRevision(1L);
|
||||
value.setObjectId(HISTORICAL_OBJECT_ID);
|
||||
value.setSourceAlias("MYSQL_1");
|
||||
value.setSchemaAlias("MAIN");
|
||||
value.setTableAlias("outlet");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 第二张历史表绑定 */
|
||||
private DataspaceTableBinding secondBinding() {
|
||||
DataspaceTableBinding value = binding();
|
||||
value.setId(SECOND_BINDING_ID);
|
||||
value.setObjectId(SECOND_HISTORICAL_OBJECT_ID);
|
||||
value.setTableAlias("outlet_region");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 依赖已删除字段的表关系 */
|
||||
private DataspaceRelation relation() {
|
||||
DataspaceRelation value = new DataspaceRelation();
|
||||
value.setId(RELATION_ID);
|
||||
value.setDataspaceId(SPACE_ID);
|
||||
value.setRevisionId(REVISION_ID);
|
||||
value.setLeftBindingId(BINDING_ID);
|
||||
value.setRightBindingId(SECOND_BINDING_ID);
|
||||
value.setJoinType("INNER");
|
||||
value.setLeftColumn("id");
|
||||
value.setRightColumn("outlet_id");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 当前物理连接 */
|
||||
private DataspaceConnection connection() {
|
||||
DataspaceConnection value = new DataspaceConnection();
|
||||
value.setId(CONNECTION_ID);
|
||||
value.setName("销售主库");
|
||||
value.setDatabaseType("MYSQL");
|
||||
value.setDefinitionRevision(2L);
|
||||
value.setStatus("ENABLED");
|
||||
return value;
|
||||
}
|
||||
|
||||
/** @return 历史元数据对象 */
|
||||
private DataspaceObject historicalObject() {
|
||||
return object(
|
||||
HISTORICAL_OBJECT_ID,
|
||||
1L,
|
||||
"outlet",
|
||||
List.of(column("id", Types.BIGINT)));
|
||||
}
|
||||
|
||||
/** @return 最新元数据对象 */
|
||||
private DataspaceObject currentObject() {
|
||||
return object(
|
||||
CURRENT_OBJECT_ID,
|
||||
2L,
|
||||
"outlet",
|
||||
List.of(column("id", Types.BIGINT), column("name", Types.VARCHAR)));
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建元数据对象。
|
||||
*
|
||||
* @param id 对象 ID
|
||||
* @param revision 元数据 revision
|
||||
* @param objectName 对象名
|
||||
* @param columns 字段
|
||||
* @return 元数据对象
|
||||
*/
|
||||
private DataspaceObject object(
|
||||
BigInteger id,
|
||||
long revision,
|
||||
String objectName,
|
||||
List<Map<String, Object>> columns) {
|
||||
DataspaceObject value = new DataspaceObject();
|
||||
value.setId(id);
|
||||
value.setConnectionId(CONNECTION_ID);
|
||||
value.setCatalogName("data-sheet");
|
||||
value.setSchemaName(null);
|
||||
value.setObjectName(objectName);
|
||||
value.setObjectType("TABLE");
|
||||
value.setColumnsJson(columns);
|
||||
value.setMetadataRevision(revision);
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建字段元数据，并保留空备注验证兼容复制。
|
||||
*
|
||||
* @param name 字段名
|
||||
* @param jdbcType JDBC 类型
|
||||
* @return 字段元数据
|
||||
*/
|
||||
private Map<String, Object> column(String name, int jdbcType) {
|
||||
Map<String, Object> value = new LinkedHashMap<>();
|
||||
value.put("name", name);
|
||||
value.put("jdbcType", jdbcType);
|
||||
value.put("typeName", jdbcType == Types.BIGINT ? "BIGINT" : "VARCHAR");
|
||||
value.put("remarks", null);
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* 测试夹具。
|
||||
*
|
||||
* @param service 数据空间服务
|
||||
*/
|
||||
private record Fixture(DataspaceService service) {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package tech.easyflow.dataspace.service;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.dataspace.model.DataspaceErrorCode;
|
||||
|
||||
/**
|
||||
* 数据空间查询结果字节预算测试。
|
||||
*/
|
||||
public class DataspaceQueryResultBudgetTest {
|
||||
|
||||
/**
|
||||
* 验证多个合法单元格累计超过结果上限时返回稳定业务错误。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectAccumulatedResultBytes() {
|
||||
DataspaceQueryResultBudget budget = new DataspaceQueryResultBudget(24L, 16L);
|
||||
budget.beginRow(2);
|
||||
budget.reserveCell(budget.validateText("12345678"));
|
||||
|
||||
BusinessException exception = expectLimitExceeded(() ->
|
||||
budget.reserveCell(budget.validateText("abcdefgh")));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("查询结果数据"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证文本按 UTF-8 字节数校验，避免多字节字符绕过单字段上限。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectUtf8TextCellOverLimit() {
|
||||
DataspaceQueryResultBudget budget = new DataspaceQueryResultBudget(128L, 4L);
|
||||
|
||||
BusinessException exception = expectLimitExceeded(() ->
|
||||
budget.validateText("中文"));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("文本字段"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证二进制字段在 Base64 编码前按原始大小拦截。
|
||||
*/
|
||||
@Test
|
||||
public void shouldRejectBinaryCellOverLimit() {
|
||||
DataspaceQueryResultBudget budget = new DataspaceQueryResultBudget(128L, 4L);
|
||||
|
||||
BusinessException exception = expectLimitExceeded(() ->
|
||||
budget.validateBinary(new byte[5]));
|
||||
|
||||
Assert.assertTrue(exception.getMessage().contains("二进制字段"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证 JSON 转义字符计入总结果字节预算。
|
||||
*/
|
||||
@Test
|
||||
public void shouldAccountForJsonEscaping() {
|
||||
DataspaceQueryResultBudget budget = new DataspaceQueryResultBudget(9L, 8L);
|
||||
budget.beginRow(1);
|
||||
|
||||
expectLimitExceeded(() -> budget.reserveCell(budget.validateText("\"\n")));
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行操作并断言返回查询上限业务错误。
|
||||
*
|
||||
* @param action 待执行操作
|
||||
* @return 捕获的业务异常
|
||||
*/
|
||||
private BusinessException expectLimitExceeded(Runnable action) {
|
||||
try {
|
||||
action.run();
|
||||
Assert.fail("expected query result limit exception");
|
||||
return null;
|
||||
} catch (BusinessException exception) {
|
||||
Assert.assertEquals(
|
||||
DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.code(),
|
||||
exception.getErrorCode());
|
||||
return exception;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package tech.easyflow.dataspace.service;
|
||||
|
||||
import com.easyagents.federation.sql.compile.FederationFragmentExplain;
|
||||
import com.easyagents.federation.sql.compile.SqlExplainLevel;
|
||||
import com.easyagents.federation.sql.compile.SqlExplainResult;
|
||||
import com.easyagents.federation.sql.execute.FederationPhysicalExplain;
|
||||
import com.easyagents.federation.sql.federation.FederationCostEstimate;
|
||||
import com.easyagents.federation.sql.federation.FederationJoinAlgorithm;
|
||||
import com.easyagents.federation.sql.federation.FederationJoinOptimization;
|
||||
import com.easyagents.federation.sql.federation.FederationJoinSelectionReason;
|
||||
import com.easyagents.federation.sql.federation.FederationQueryMode;
|
||||
import com.easyagents.federation.sql.federation.FederationStatisticsStatus;
|
||||
import com.easyagents.federation.sql.source.SourceId;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.dataspace.model.DataspaceExplainResult;
|
||||
|
||||
/**
|
||||
* 数据空间 Explain API 视图映射测试。
|
||||
*/
|
||||
public class DataspaceQueryServiceExplainTest {
|
||||
|
||||
/**
|
||||
* 验证聚合成本、Join 决策和分片统计完整映射到 API 视图。
|
||||
*/
|
||||
@Test
|
||||
public void shouldMapFederationExplainDetails() {
|
||||
Instant collectedAt = Instant.parse("2026-08-23T08:00:00Z");
|
||||
FederationCostEstimate cost = new FederationCostEstimate(
|
||||
120D, 64L, 7_680D, "catalog", "stats-7", collectedAt,
|
||||
false, FederationStatisticsStatus.COMPLETE);
|
||||
FederationPhysicalExplain physical = new FederationPhysicalExplain(
|
||||
true, "Index Scan", "INDEX_SCAN", "INDEX", List.of("idx_region"),
|
||||
"idx_region", 120L, null, "");
|
||||
FederationFragmentExplain fragment = new FederationFragmentExplain(
|
||||
"fragment-1", "REGION", new SourceId("region-source"), "jdbc",
|
||||
"SELECT * FROM region", List.of(), List.of(), cost,
|
||||
List.of("TableScan", "Filter"), physical);
|
||||
FederationJoinOptimization join = new FederationJoinOptimization(
|
||||
1, List.of("SALES"), List.of("REGION"), "REGION",
|
||||
FederationJoinAlgorithm.HASH_JOIN,
|
||||
FederationJoinSelectionReason.SMALLER_BUILD_SIDE, 7_680D);
|
||||
SqlExplainResult explain = new SqlExplainResult(
|
||||
SqlExplainLevel.PHYSICAL, FederationQueryMode.FEDERATED,
|
||||
FederationStatisticsStatus.COMPLETE, true, 12_400D, 3_100L,
|
||||
List.of(join), "SELECT * FROM sales JOIN region", "",
|
||||
"logical", "execution", List.of(fragment),
|
||||
Set.of(new SourceId("region-source")), null, true, false, "");
|
||||
|
||||
DataspaceExplainResult view = DataspaceQueryService.explainView(explain);
|
||||
|
||||
Assert.assertEquals("FEDERATED", view.queryMode());
|
||||
Assert.assertEquals("COMPLETE", view.statisticsStatus());
|
||||
Assert.assertTrue(view.estimateAvailable());
|
||||
Assert.assertEquals(12_400D, view.estimatedTransferBytes(), 0D);
|
||||
Assert.assertEquals(3_100L, view.estimatedLocalMemoryBytes());
|
||||
Assert.assertEquals(1, view.joins().get(0).stageIndex());
|
||||
Assert.assertEquals(List.of("SALES"), view.joins().get(0).leftSources());
|
||||
Assert.assertEquals(List.of("REGION"), view.joins().get(0).rightSources());
|
||||
Assert.assertEquals("REGION", view.joins().get(0).buildSource());
|
||||
Assert.assertEquals("INDEX", view.fragments().get(0).scanType());
|
||||
Assert.assertEquals("idx_region", view.fragments().get(0).chosenIndex());
|
||||
Assert.assertEquals(collectedAt.toString(),
|
||||
view.fragments().get(0).statisticsCollectedAt());
|
||||
Assert.assertEquals(List.of("TableScan", "Filter"),
|
||||
view.fragments().get(0).pushedDownOperators());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证未知估算保持显式不可用，避免页面把零值展示成精确结果。
|
||||
*/
|
||||
@Test
|
||||
public void shouldPreserveUnavailableEstimate() {
|
||||
FederationCostEstimate unavailable = new FederationCostEstimate(
|
||||
0D, 0L, 0D, "default", "none", Instant.EPOCH,
|
||||
true, FederationStatisticsStatus.MISSING, false);
|
||||
FederationFragmentExplain fragment = new FederationFragmentExplain(
|
||||
"fragment-1", "SALES", new SourceId("sales-source"), "jdbc",
|
||||
"SELECT 1", List.of(), List.of(), unavailable, List.of(), null);
|
||||
SqlExplainResult explain = new SqlExplainResult(
|
||||
SqlExplainLevel.LOGICAL, FederationQueryMode.SINGLE_SOURCE,
|
||||
FederationStatisticsStatus.MISSING, false, 0D, 0L, List.of(),
|
||||
"SELECT 1", "SELECT 1", "logical", "execution", List.of(fragment),
|
||||
Set.of(new SourceId("sales-source")), null, true, false, "");
|
||||
|
||||
DataspaceExplainResult view = DataspaceQueryService.explainView(explain);
|
||||
|
||||
Assert.assertFalse(view.estimateAvailable());
|
||||
Assert.assertFalse(view.fragments().get(0).estimateAvailable());
|
||||
Assert.assertNull(view.fragments().get(0).statisticsCollectedAt());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user