entry : historicalObjects.entrySet()) {
+ DataspaceObject historical = entry.getValue();
+ DataspaceConnection connection = connections.get(historical.getConnectionId());
+ if (connection == null) {
+ result.put(entry.getKey(), new ObjectResolution(
+ historical, null, "MISSING",
+ DataspaceErrorCode.CONNECTION_NOT_FOUND.code(),
+ DataspaceErrorCode.CONNECTION_NOT_FOUND.defaultMessage()));
+ continue;
+ }
+ if (!"ENABLED".equals(connection.getStatus())) {
+ result.put(entry.getKey(), new ObjectResolution(
+ historical, null, "CONNECTION_DISABLED",
+ DataspaceErrorCode.CONNECTION_DISABLED.code(),
+ DataspaceErrorCode.CONNECTION_DISABLED.defaultMessage()));
+ continue;
+ }
+ DataspaceObject current = currentByConnection
+ .getOrDefault(connection.getId(), Map.of())
+ .get(identity(historical, connection));
+ result.put(entry.getKey(), current == null
+ ? new ObjectResolution(
+ historical, null, "MISSING",
+ DataspaceErrorCode.OBJECT_MISSING.code(),
+ "表或视图 “" + historical.getObjectName()
+ + "” 已不在最新元数据中，请刷新或移除该表")
+ : new ObjectResolution(historical, current, "ACTIVE", null, null));
+ }
+ return Map.copyOf(result);
+ }
+
+ /**
+ * 获取连接当前对应的元数据快照 revision。
+ *
+ * 首期元数据快照随 Federation Definition 一同推进；解析逻辑集中在此处，
+ * 后续拆分独立 metadata revision 时无需修改调用方。
+ *
+ * @param connection 物理连接
+ * @return 当前元数据 revision，缺失时返回 0
+ */
+ public long currentMetadataRevision(DataspaceConnection connection) {
+ return connection == null || connection.getDefinitionRevision() == null
+ ? 0L : connection.getDefinitionRevision();
+ }
+
+ /**
+ * 判断对象字段是否存在。
+ *
+ * @param object 元数据对象
+ * @param columnName 字段名
+ * @param databaseType 数据库类型
+ * @return 是否存在
+ */
+ public boolean hasColumn(
+ DataspaceObject object,
+ String columnName,
+ String databaseType) {
+ if (object == null || columnName == null || columnName.isBlank()) {
+ return false;
+ }
+ return columns(object).stream()
+ .map(column -> String.valueOf(column.get("name")))
+ .anyMatch(name -> sameIdentifier(name, columnName.trim(), databaseType));
+ }
+
+ /**
+ * 返回指定字段的元数据。
+ *
+ * @param object 元数据对象
+ * @param columnName 字段名
+ * @param databaseType 数据库类型
+ * @return 字段元数据，不存在时返回空映射
+ */
+ public Map column(
+ DataspaceObject object,
+ String columnName,
+ String databaseType) {
+ if (object == null || columnName == null || columnName.isBlank()) {
+ return Map.of();
+ }
+ return columns(object).stream()
+ .filter(column -> sameIdentifier(
+ String.valueOf(column.get("name")),
+ columnName.trim(),
+ databaseType))
+ .findFirst()
+ .map(column -> java.util.Collections.unmodifiableMap(
+ new LinkedHashMap<>(column)))
+ .orElseGet(Map::of);
+ }
+
+ /**
+ * 安全读取对象字段，兼容历史异常数据中的空字段集合。
+ *
+ * @param object 元数据对象
+ * @return 非空字段集合
+ */
+ private List> columns(DataspaceObject object) {
+ return object.getColumnsJson() == null ? List.of() : object.getColumnsJson();
+ }
+
+ /**
+ * 按数据库标识符规则比较字段名。
+ *
+ * @param left 元数据字段名
+ * @param right 保存的字段名
+ * @param databaseType 数据库类型
+ * @return 是否指向同一字段
+ */
+ private boolean sameIdentifier(String left, String right, String databaseType) {
+ return "POSTGRESQL".equalsIgnoreCase(databaseType)
+ ? left.equals(right)
+ : left.equalsIgnoreCase(right);
+ }
+
+ /**
+ * 判断两个字段的 JDBC 类型是否可用于等值关联。
+ *
+ * @param left 左字段元数据
+ * @param right 右字段元数据
+ * @return 是否可比较
+ */
+ public boolean comparable(Map left, Map right) {
+ if (left.isEmpty() || right.isEmpty()) {
+ return false;
+ }
+ String leftFamily = typeFamily(left);
+ String rightFamily = typeFamily(right);
+ return leftFamily.equals(rightFamily);
+ }
+
+ /**
+ * 构造稳定物理身份。
+ *
+ * @param object 元数据对象
+ * @param connection 物理连接
+ * @return 物理身份
+ */
+ private PhysicalIdentity identity(
+ DataspaceObject object,
+ DataspaceConnection connection) {
+ boolean caseSensitive = "POSTGRESQL".equalsIgnoreCase(connection.getDatabaseType());
+ return new PhysicalIdentity(
+ normalize(object.getCatalogName(), caseSensitive),
+ normalize(object.getSchemaName(), caseSensitive),
+ normalize(object.getObjectName(), caseSensitive),
+ normalize(object.getObjectType(), false));
+ }
+
+ /**
+ * 规范化物理身份片段。
+ *
+ * @param value 原始值
+ * @param caseSensitive 是否保留大小写
+ * @return 大小写无关的稳定值
+ */
+ private String normalize(String value, boolean caseSensitive) {
+ String normalized = value == null ? "" : value.trim();
+ return caseSensitive ? normalized : normalized.toUpperCase(Locale.ROOT);
+ }
+
+ /**
+ * 将 JDBC 类型归一化为关联比较族。
+ *
+ * @param column 字段元数据
+ * @return 类型族
+ */
+ private String typeFamily(Map column) {
+ int jdbcType;
+ try {
+ jdbcType = Integer.parseInt(String.valueOf(column.get("jdbcType")));
+ } catch (NumberFormatException exception) {
+ return "OTHER:" + String.valueOf(column.get("typeName")).toUpperCase(Locale.ROOT);
+ }
+ return switch (jdbcType) {
+ case java.sql.Types.TINYINT, java.sql.Types.SMALLINT, java.sql.Types.INTEGER,
+ java.sql.Types.BIGINT, java.sql.Types.FLOAT, java.sql.Types.REAL,
+ java.sql.Types.DOUBLE, java.sql.Types.NUMERIC, java.sql.Types.DECIMAL -> "NUMERIC";
+ case java.sql.Types.CHAR, java.sql.Types.VARCHAR, java.sql.Types.LONGVARCHAR,
+ java.sql.Types.NCHAR, java.sql.Types.NVARCHAR, java.sql.Types.LONGNVARCHAR -> "TEXT";
+ case java.sql.Types.DATE, java.sql.Types.TIME, java.sql.Types.TIME_WITH_TIMEZONE,
+ java.sql.Types.TIMESTAMP, java.sql.Types.TIMESTAMP_WITH_TIMEZONE -> "TEMPORAL";
+ case java.sql.Types.BOOLEAN, java.sql.Types.BIT -> "BOOLEAN";
+ case java.sql.Types.BINARY, java.sql.Types.VARBINARY, java.sql.Types.LONGVARBINARY -> "BINARY";
+ default -> "OTHER:" + String.valueOf(column.get("typeName")).toUpperCase(Locale.ROOT);
+ };
+ }
+
+ /**
+ * 历史对象到当前元数据对象的解析结果。
+ *
+ * @param historical 历史对象
+ * @param current 当前对象；不可用时为空
+ * @param status ACTIVE、MISSING 或 CONNECTION_DISABLED
+ * @param issueCode 稳定问题码
+ * @param issueMessage 用户可读提示
+ */
+ public record ObjectResolution(
+ DataspaceObject historical,
+ DataspaceObject current,
+ String status,
+ Integer issueCode,
+ String issueMessage) {
+
+ /**
+ * 返回当前对象；不可用时返回历史快照供页面定位与修复。
+ *
+ * @return 可展示对象
+ */
+ public DataspaceObject displayObject() {
+ return current == null ? historical : current;
+ }
+
+ /**
+ * 判断对象是否可用。
+ *
+ * @return 是否为 ACTIVE
+ */
+ public boolean active() {
+ return "ACTIVE".equals(status);
+ }
+ }
+
+ /**
+ * 数据库对象稳定物理身份。
+ *
+ * @param catalog Catalog
+ * @param schema Schema
+ * @param name 对象名
+ * @param type 对象类型
+ */
+ private record PhysicalIdentity(String catalog, String schema, String name, String type) {
+ }
+}
diff --git a/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryResultBudget.java b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryResultBudget.java
new file mode 100644
index 00000000..e0e53862
--- /dev/null
+++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryResultBudget.java
@@ -0,0 +1,161 @@
+package tech.easyflow.dataspace.service;
+
+import tech.easyflow.dataspace.model.DataspaceErrorCode;
+
+/**
+ * 数据空间查询结果的单字段与序列化数据量预算。
+ */
+final class DataspaceQueryResultBudget {
+
+ private final long maximumResultBytes;
+ private final long maximumCellBytes;
+ private long reservedBytes = 2L;
+ private int rows;
+
+ /**
+ * 创建结果预算。
+ *
+ * @param maximumResultBytes 结果行数据的最大 JSON 字节数
+ * @param maximumCellBytes 单个文本或二进制字段的最大原始字节数
+ * @throws IllegalArgumentException 上限不为正时抛出
+ */
+ DataspaceQueryResultBudget(long maximumResultBytes, long maximumCellBytes) {
+ if (maximumResultBytes <= 0L || maximumCellBytes <= 0L) {
+ throw new IllegalArgumentException("查询结果预算必须为正数");
+ }
+ this.maximumResultBytes = maximumResultBytes;
+ this.maximumCellBytes = maximumCellBytes;
+ }
+
+ /**
+ * 为一行结果预留数组和分隔符开销。
+ *
+ * @param columnCount 当前行列数
+ * @throws IllegalArgumentException 列数为负数时抛出
+ */
+ void beginRow(int columnCount) {
+ if (columnCount < 0) {
+ throw new IllegalArgumentException("列数不能为负数");
+ }
+ reserve(2L + Math.max(0, columnCount - 1) + (rows == 0 ? 0L : 1L));
+ rows++;
+ }
+
+ /**
+ * 校验单个文本字段的 UTF-8 字节数。
+ *
+ * @param value 文本值
+ * @return 原文本值
+ * @throws tech.easyflow.common.web.exceptions.BusinessException 超过单字段上限时抛出
+ */
+ String validateText(String value) {
+ if (utf8Bytes(value) > maximumCellBytes) {
+ throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
+ "单个文本字段超过 1 MiB 限制");
+ }
+ return value;
+ }
+
+ /**
+ * 校验单个二进制字段的原始字节数。
+ *
+ * @param value 二进制值
+ * @return 原二进制值
+ * @throws tech.easyflow.common.web.exceptions.BusinessException 超过单字段上限时抛出
+ */
+ byte[] validateBinary(byte[] value) {
+ if (value.length > maximumCellBytes) {
+ throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
+ "单个二进制字段超过 1 MiB 限制");
+ }
+ return value;
+ }
+
+ /**
+ * 将一个已规范化单元格计入结果序列化预算。
+ *
+ * @param value JSON 安全的单元格值
+ * @throws tech.easyflow.common.web.exceptions.BusinessException 超过结果上限时抛出
+ */
+ void reserveCell(Object value) {
+ long bytes;
+ if (value == null) {
+ bytes = 4L;
+ } else if (value instanceof String text) {
+ bytes = jsonStringBytes(text);
+ } else {
+ bytes = utf8Bytes(String.valueOf(value));
+ }
+ reserve(bytes);
+ }
+
+ /**
+ * 原子检查并增加已使用预算。
+ *
+ * @param bytes 新增字节数
+ * @throws tech.easyflow.common.web.exceptions.BusinessException 超过结果上限时抛出
+ */
+ private void reserve(long bytes) {
+ if (bytes < 0L || bytes > maximumResultBytes - reservedBytes) {
+ throw DataspaceErrorCode.QUERY_LIMIT_EXCEEDED.exception(
+ "查询结果数据超过 16 MiB 限制，请减少返回行数或字段");
+ }
+ reservedBytes += bytes;
+ }
+
+ /**
+ * 计算字符串的 UTF-8 字节数，避免为预算检查额外复制大字符串。
+ *
+ * @param value 字符串
+ * @return UTF-8 字节数
+ */
+ private static long utf8Bytes(String value) {
+ long bytes = 0L;
+ for (int index = 0; index < value.length(); index++) {
+ char current = value.charAt(index);
+ if (current <= 0x7F) {
+ bytes++;
+ } else if (current <= 0x7FF) {
+ bytes += 2L;
+ } else if (Character.isHighSurrogate(current)
+ && index + 1 < value.length()
+ && Character.isLowSurrogate(value.charAt(index + 1))) {
+ bytes += 4L;
+ index++;
+ } else {
+ bytes += 3L;
+ }
+ }
+ return bytes;
+ }
+
+ /**
+ * 计算 JSON 字符串编码后的字节数，包括引号与转义字符。
+ *
+ * @param value 字符串
+ * @return JSON 字符串字节数
+ */
+ private static long jsonStringBytes(String value) {
+ long bytes = 2L;
+ for (int index = 0; index < value.length(); index++) {
+ char current = value.charAt(index);
+ if (current == '"' || current == '\\') {
+ bytes += 2L;
+ } else if (current <= 0x1F) {
+ bytes += 6L;
+ } else if (current <= 0x7F) {
+ bytes++;
+ } else if (current <= 0x7FF) {
+ bytes += 2L;
+ } else if (Character.isHighSurrogate(current)
+ && index + 1 < value.length()
+ && Character.isLowSurrogate(value.charAt(index + 1))) {
+ bytes += 4L;
+ index++;
+ } else {
+ bytes += 3L;
+ }
+ }
+ return bytes;
+ }
+}
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
new file mode 100644
index 00000000..129d5db8
--- /dev/null
+++ b/easyflow-modules/easyflow-module-dataspace/src/main/java/tech/easyflow/dataspace/service/DataspaceQueryService.java
@@ -0,0 +1,701 @@
+package tech.easyflow.dataspace.service;
+
+import com.easyagents.federation.sql.api.FederationSqlException;
+import com.easyagents.federation.sql.api.SqlCompletionRequest;
+import com.easyagents.federation.sql.api.SqlCompletionResult;
+import com.easyagents.federation.sql.api.SqlQueryCommand;
+import com.easyagents.federation.sql.compile.FederationFragmentExplain;
+import com.easyagents.federation.sql.compile.SqlCompileRequest;
+import com.easyagents.federation.sql.compile.SqlExplainLevel;
+import com.easyagents.federation.sql.compile.SqlExplainRequest;
+import com.easyagents.federation.sql.compile.SqlExplainResult;
+import com.easyagents.federation.sql.execute.FederationColumn;
+import com.easyagents.federation.sql.execute.FederationPhysicalExplain;
+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;
+import com.easyagents.federation.sql.source.FederationSourceDefinition;
+import com.easyagents.federation.sql.source.SourceId;
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.SQLXML;
+import java.time.Duration;
+import java.time.temporal.TemporalAccessor;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Date;
+import java.util.HexFormat;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.slf4j.Logger;
+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.entity.DataspaceConnection;
+import tech.easyflow.dataspace.entity.DataspaceObject;
+import tech.easyflow.dataspace.entity.DataspaceQueryAudit;
+import tech.easyflow.dataspace.entity.DataspaceTableBinding;
+import tech.easyflow.dataspace.federation.DataspaceDefinitionFactory;
+import tech.easyflow.dataspace.federation.DataspaceFederationRuntime;
+import tech.easyflow.dataspace.mapper.DataspaceQueryAuditMapper;
+import tech.easyflow.dataspace.model.DataspaceExplainResult;
+import tech.easyflow.dataspace.model.DataspaceErrorCode;
+import tech.easyflow.dataspace.model.DataspaceQueryRequest;
+import tech.easyflow.dataspace.model.DataspaceQueryResult;
+import tech.easyflow.dataspace.model.DataspaceSqlCompletionRequest;
+import tech.easyflow.dataspace.model.DataspaceSqlCompletionResult;
+import tech.easyflow.dataspace.security.DataspaceActor;
+
+/**
+ * 数据空间单源与联邦只读 Query、Explain、Complete 和 Cancel 服务。
+ */
+@Service
+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 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 ConcurrentMap activeQueries = new ConcurrentHashMap<>();
+
+ /**
+ * 创建查询服务。
+ *
+ * @param dataspaceService 数据空间服务
+ * @param runtime Federation Runtime
+ * @param definitionFactory Definition 工厂
+ * @param auditMapper 查询审计 Mapper
+ */
+ public DataspaceQueryService(
+ DataspaceService dataspaceService,
+ DataspaceFederationRuntime runtime,
+ DataspaceDefinitionFactory definitionFactory,
+ DataspaceQueryAuditMapper auditMapper) {
+ this.dataspaceService = dataspaceService;
+ this.runtime = runtime;
+ this.definitionFactory = definitionFactory;
+ this.auditMapper = auditMapper;
+ }
+
+ /**
+ * 执行一条 Calcite 只读 SQL，并返回数据和常用消耗指标。
+ *
+ * @param request 查询请求
+ * @return 查询结果
+ */
+ public DataspaceQueryResult query(DataspaceQueryRequest request) {
+ validateRequest(request);
+ 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 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());
+ LoginAccount account = DataspaceActor.current();
+ ActiveQueryOwner owner = new ActiveQueryOwner(account.getTenantId(), account.getId());
+ if (activeQueries.putIfAbsent(queryId.value(), owner) != null) {
+ throw DataspaceErrorCode.QUERY_ID_CONFLICT.exception();
+ }
+ SqlQueryCommand command = new SqlQueryCommand(
+ queryId, executableSql, scope, List.of(),
+ new SqlExecutionOptions(500, maxRows, timeout, true),
+ Duration.ofSeconds(5).toMillis(), policyVersion(snapshot));
+ FederationQueryMetricsSnapshot metrics = null;
+ String failureCode = null;
+ String failureMessage = null;
+ try (FederationResultCursor cursor = runtime.engine().query(command)) {
+ List columns = cursor.columns().stream()
+ .map(this::columnView).toList();
+ List> rows = new ArrayList<>();
+ DataspaceQueryResultBudget resultBudget = new DataspaceQueryResultBudget(
+ MAX_RESULT_BYTES, MAX_CELL_BYTES);
+ while (cursor.next()) {
+ resultBudget.beginRow(columns.size());
+ List