feat: 新增 SQL 联邦查询与数据库适配底座

- 提供统一编译、逻辑表映射与单源/联邦自动路由

- 增加有界执行、查询生命周期、统计成本优化与执行分析

- 内置 MySQL 与 PostgreSQL JDBC 适配和统计采集
This commit is contained in:
2026-08-25 01:00:49 +08:00
parent 2f67d90144
commit 02b8fdd3ae
155 changed files with 27704 additions and 0 deletions

View File

@@ -0,0 +1,54 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.StatementLifecycle;
import java.sql.SQLTimeoutException;
/**
* 按查询注册表已经确定的终态分类 JDBC 执行与读取异常。
*/
final class JdbcFailureClassifier {
/**
* 工具类无需实例化。
*/
private JdbcFailureClassifier() {
}
/**
* 将 JDBC 异常转换为稳定的查询错误,优先保留先到达的取消或超时终态。
*
* @param lifecycle 查询 Statement 生命周期
* @param cause JDBC 或驱动异常
* @param timeoutMessage 超时提示
* @param cancellationMessage 取消提示
* @param failureMessage 普通执行失败提示
* @return 分类后的统一异常
*/
static FederationSqlException classify(
StatementLifecycle lifecycle,
Throwable cause,
String timeoutMessage,
String cancellationMessage,
String failureMessage
) {
FederationSqlErrorCode errorCode;
String message;
if (lifecycle.timeoutRequested()) {
errorCode = FederationSqlErrorCode.QUERY_TIMEOUT;
message = timeoutMessage;
} else if (lifecycle.cancellationRequested()) {
// Statement.cancel() 后部分驱动会抛 SQLTimeoutException已登记的取消终态必须优先。
errorCode = FederationSqlErrorCode.QUERY_CANCELLED;
message = cancellationMessage;
} else if (cause instanceof SQLTimeoutException) {
errorCode = FederationSqlErrorCode.QUERY_TIMEOUT;
message = timeoutMessage;
} else {
errorCode = FederationSqlErrorCode.EXECUTION_FAILED;
message = failureMessage;
}
return new FederationSqlException(errorCode, message, cause);
}
}

View File

@@ -0,0 +1,200 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationColumn;
import com.easyagents.federation.sql.execute.FederationFragmentExecutionContext;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.SqlParameter;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
/**
* 直接使用 PreparedStatement 执行目标数据库 SQL 的流式 JDBC 执行器。
*/
final class JdbcFederationFragmentExecutor implements FederationFragmentExecutor {
/**
* 获取连接、应用只读限制、绑定参数并返回持有全部资源的流式游标。
*
* @param context 执行上下文
* @return 流式游标
*/
@Override
public FederationResultCursor execute(FederationFragmentExecutionContext context) {
Connection connection = null;
PreparedStatement statement = null;
boolean registered = false;
boolean connectionAcquired = false;
try {
context.executionGuard().ensureAllowed();
long connectionStarted = System.nanoTime();
connection = context.dataSource().getConnection();
connectionAcquired = true;
context.observer().connectionAcquired(System.nanoTime() - connectionStarted);
context.executionGuard().ensureAllowed();
configureConnection(connection, context);
statement = connection.prepareStatement(
context.sql(),
ResultSet.TYPE_FORWARD_ONLY,
ResultSet.CONCUR_READ_ONLY
);
applyOptions(statement, context);
bindParameters(statement, context.parameters());
context.statementLifecycle().register(statement);
registered = true;
context.executionGuard().ensureAllowed();
long executionStarted = System.nanoTime();
ResultSet resultSet = statement.executeQuery();
context.executionGuard().ensureAllowed();
context.observer().databaseExecutionCompleted(System.nanoTime() - executionStarted);
List<FederationColumn> columns = readColumns(resultSet.getMetaData());
return new JdbcFederationResultCursor(
context.queryId(),
columns,
resultSet,
statement,
connection,
context.statementLifecycle(),
context.executionGuard(),
context.observer()
);
} catch (SQLException | RuntimeException exception) {
if (!connectionAcquired) {
// 连接池等待可能跨过统一截止时间;总超时或显式取消应保持为查询终态。
context.executionGuard().ensureAllowed();
}
if (registered) {
context.statementLifecycle().unregister(statement);
}
closeAfterFailure(statement, connection, exception);
if (exception instanceof FederationSqlException federationSqlException) {
throw federationSqlException;
}
boolean connectionTimedOut = !connectionAcquired
&& (exception instanceof SQLTransientConnectionException
|| exception instanceof SQLTimeoutException);
if (connectionTimedOut) {
throw new FederationSqlException(
FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT,
"timed out while acquiring a JDBC connection",
exception
);
}
if (!connectionAcquired) {
throw new FederationSqlException(
FederationSqlErrorCode.CONNECTION_ACQUISITION_FAILED,
"failed to acquire a JDBC connection",
exception
);
}
throw JdbcFailureClassifier.classify(
context.statementLifecycle(),
exception,
"JDBC query timed out",
"JDBC query was cancelled",
"JDBC query execution failed"
);
}
}
private static void applyOptions(
PreparedStatement statement,
FederationFragmentExecutionContext context
) throws SQLException {
int fetchSize = effectiveFetchSize(context);
if (fetchSize != 0) {
statement.setFetchSize(fetchSize);
}
if (context.options().maxRows() > 0) {
statement.setMaxRows(context.options().maxRows());
}
int queryTimeout = context.executionGuard().boundedQueryTimeoutSeconds(
context.options().queryTimeoutSeconds()
);
if (queryTimeout > 0) {
statement.setQueryTimeout(queryTimeout);
}
}
private static void configureConnection(
Connection connection,
FederationFragmentExecutionContext context
) throws SQLException {
if (!connection.isReadOnly()) {
connection.setReadOnly(true);
}
String product = context.compatibility().databaseProduct().toLowerCase(Locale.ROOT);
// PostgreSQL 只有在事务模式下才会按正 fetchSize 使用服务端游标。
if (product.contains("postgres") && context.options().fetchSize() > 0
&& connection.getAutoCommit()) {
connection.setAutoCommit(false);
}
}
private static int effectiveFetchSize(FederationFragmentExecutionContext context) {
String product = context.compatibility().databaseProduct().toLowerCase(Locale.ROOT);
if (product.contains("mysql")
&& "legacy".equalsIgnoreCase(context.adapterOptions().get("mysqlStreamingMode"))) {
// Connector/J 旧式逐行流需要显式 MIN_VALUE默认仍使用正 fetchSize + useCursorFetch。
return Integer.MIN_VALUE;
}
return context.options().fetchSize();
}
private static void bindParameters(PreparedStatement statement, List<SqlParameter> parameters)
throws SQLException {
for (int index = 0; index < parameters.size(); index++) {
SqlParameter parameter = parameters.get(index);
int jdbcIndex = index + 1;
if (parameter.value() == null) {
statement.setNull(jdbcIndex, parameter.jdbcType());
} else {
statement.setObject(jdbcIndex, parameter.value(), parameter.jdbcType());
}
}
}
private static List<FederationColumn> readColumns(ResultSetMetaData metadata) throws SQLException {
List<FederationColumn> columns = new ArrayList<>(metadata.getColumnCount());
for (int index = 1; index <= metadata.getColumnCount(); index++) {
columns.add(new FederationColumn(
index,
metadata.getColumnLabel(index),
metadata.getColumnType(index),
metadata.getColumnTypeName(index),
metadata.isNullable(index) != ResultSetMetaData.columnNoNulls
));
}
return List.copyOf(columns);
}
private static void closeAfterFailure(
PreparedStatement statement,
Connection connection,
Throwable original
) {
closeAndSuppress(statement, original);
closeAndSuppress(connection, original);
}
private static void closeAndSuppress(AutoCloseable closeable, Throwable original) {
if (closeable == null) {
return;
}
try {
closeable.close();
} catch (Exception closeException) {
original.addSuppressed(closeException);
}
}
}

View File

@@ -0,0 +1,307 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationFragmentExplainContext;
import com.easyagents.federation.sql.execute.FederationFragmentExplainer;
import com.easyagents.federation.sql.execute.FederationPhysicalExplain;
import com.easyagents.federation.sql.execute.SqlParameter;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
/**
* MySQL、PostgreSQL 和 H2 的非 ANALYZE 物理 Explain 实现。
*/
final class JdbcFederationFragmentExplainer implements FederationFragmentExplainer {
private static final ObjectMapper JSON = new ObjectMapper();
/** {@inheritDoc} */
@Override
public FederationPhysicalExplain explain(FederationFragmentExplainContext context) {
String product = normalize(context.compatibility().databaseProduct());
String explainSql = explainSql(product, context.sql());
if (explainSql == null) {
return FederationPhysicalExplain.unavailable(
"physical Explain is not implemented for "
+ context.compatibility().databaseProduct()
);
}
Connection acquired = acquireConnection(context);
try (Connection connection = acquired) {
context.executionGuard().ensureAllowed();
if (!connection.isReadOnly()) {
connection.setReadOnly(true);
}
try (PreparedStatement statement = connection.prepareStatement(explainSql)) {
int queryTimeout = context.executionGuard().boundedQueryTimeoutSeconds(
context.queryTimeoutSeconds()
);
if (queryTimeout > 0) {
statement.setQueryTimeout(queryTimeout);
}
bind(statement, context.parameters());
context.executionGuard().ensureAllowed();
try (ResultSet resultSet = statement.executeQuery()) {
context.executionGuard().ensureAllowed();
String nativePlan = readPlan(resultSet);
return normalizePlan(product, nativePlan);
}
}
} catch (SQLException | RuntimeException exception) {
if (exception instanceof FederationSqlException federationSqlException) {
throw federationSqlException;
}
throw new FederationSqlException(
FederationSqlErrorCode.EXPLAIN_FAILED,
"physical database Explain failed",
exception
);
}
}
/**
* 在统一截止时间约束下获取物理 Explain 连接。
*
* @param context 分片 Explain 上下文
* @return 已获取连接
* @throws FederationSqlException 获取超时、失败或查询已终止时抛出
*/
private static Connection acquireConnection(FederationFragmentExplainContext context) {
try {
context.executionGuard().ensureAllowed();
return context.dataSource().getConnection();
} catch (SQLException | RuntimeException exception) {
if (exception instanceof FederationSqlException federationSqlException) {
throw federationSqlException;
}
context.executionGuard().ensureAllowed();
FederationSqlErrorCode code = exception instanceof SQLTimeoutException
|| exception instanceof SQLTransientConnectionException
? FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT
: FederationSqlErrorCode.CONNECTION_ACQUISITION_FAILED;
String message = code == FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT
? "timed out while acquiring a JDBC connection for physical Explain"
: "failed to acquire a JDBC connection for physical Explain";
throw new FederationSqlException(code, message, exception);
}
}
private static String explainSql(String product, String sql) {
if (product.contains("mysql")) {
return "EXPLAIN FORMAT=JSON " + sql;
}
if (product.contains("postgres")) {
return "EXPLAIN (FORMAT JSON, ANALYZE FALSE, COSTS TRUE, VERBOSE FALSE, BUFFERS FALSE) "
+ sql;
}
if (product.equals("h2")) {
return "EXPLAIN " + sql;
}
return null;
}
private static void bind(PreparedStatement statement, List<SqlParameter> parameters)
throws SQLException {
for (int index = 0; index < parameters.size(); index++) {
SqlParameter parameter = parameters.get(index);
if (parameter.value() == null) {
statement.setNull(index + 1, parameter.jdbcType());
} else {
statement.setObject(index + 1, parameter.value(), parameter.jdbcType());
}
}
}
private static String readPlan(ResultSet resultSet) throws SQLException {
StringBuilder plan = new StringBuilder();
ResultSetMetaData metadata = resultSet.getMetaData();
while (resultSet.next()) {
if (!plan.isEmpty()) {
plan.append('\n');
}
for (int column = 1; column <= metadata.getColumnCount(); column++) {
if (column > 1) {
plan.append('\t');
}
Object value = resultSet.getObject(column);
if (value != null) {
plan.append(value);
}
}
}
return plan.toString();
}
private static FederationPhysicalExplain normalizePlan(String product, String nativePlan) {
if (!nativePlan.isBlank() && (product.contains("mysql") || product.contains("postgres"))) {
try {
JsonNode root = JSON.readTree(nativePlan);
return product.contains("mysql")
? normalizeMysql(root, nativePlan)
: normalizePostgresql(root, nativePlan);
} catch (Exception ignored) {
// 原生计划仍可用;归一化失败不会伪造索引结论。
}
}
return new FederationPhysicalExplain(
true,
nativePlan,
null,
null,
List.of(),
null,
null,
null,
"native plan is available; normalized index fields are unavailable"
);
}
private static FederationPhysicalExplain normalizeMysql(JsonNode root, String nativePlan) {
JsonNode table = findObjectWithField(root, "access_type");
if (table == null) {
return nativeOnly(nativePlan, "MySQL plan contains no normalized table access node");
}
List<String> candidates = stringValues(table.get("possible_keys"));
return new FederationPhysicalExplain(
true,
nativePlan,
"table",
text(table, "access_type"),
candidates,
text(table, "key"),
longValue(table, "rows_examined_per_scan", "rows"),
firstText(table, "attached_condition", "index_condition"),
"normalized from MySQL JSON Explain"
);
}
private static FederationPhysicalExplain normalizePostgresql(JsonNode root, String nativePlan) {
JsonNode plan = root.isArray() && !root.isEmpty() ? root.get(0).get("Plan") : root.get("Plan");
JsonNode scan = findObjectWithField(plan, "Index Name");
if (scan == null) {
scan = findObjectWithTextSuffix(plan, "Node Type", "Scan");
}
if (scan == null) {
return nativeOnly(nativePlan, "PostgreSQL plan contains no normalized plan node");
}
return new FederationPhysicalExplain(
true,
nativePlan,
text(scan, "Node Type"),
text(scan, "Node Type"),
List.of(),
text(scan, "Index Name"),
longValue(scan, "Plan Rows"),
firstText(scan, "Index Cond", "Filter", "Join Filter"),
"normalized from PostgreSQL JSON Explain"
);
}
private static FederationPhysicalExplain nativeOnly(String nativePlan, String diagnostic) {
return new FederationPhysicalExplain(
true,
nativePlan,
null,
null,
List.of(),
null,
null,
null,
diagnostic
);
}
private static JsonNode findObjectWithField(JsonNode node, String field) {
if (node == null) {
return null;
}
if (node.isObject() && node.has(field)) {
return node;
}
Iterator<JsonNode> children = node.elements();
while (children.hasNext()) {
JsonNode found = findObjectWithField(children.next(), field);
if (found != null) {
return found;
}
}
return null;
}
private static JsonNode findObjectWithTextSuffix(
JsonNode node,
String field,
String suffix
) {
if (node == null) {
return null;
}
if (node.isObject()) {
String value = text(node, field);
if (value != null && value.endsWith(suffix)) {
return node;
}
}
Iterator<JsonNode> children = node.elements();
while (children.hasNext()) {
JsonNode found = findObjectWithTextSuffix(children.next(), field, suffix);
if (found != null) {
return found;
}
}
return null;
}
private static List<String> stringValues(JsonNode node) {
if (node == null || node.isNull()) {
return List.of();
}
if (node.isArray()) {
List<String> values = new ArrayList<>();
node.forEach(value -> values.add(value.asText()));
return List.copyOf(values);
}
return List.of(node.asText());
}
private static String firstText(JsonNode node, String... fields) {
for (String field : fields) {
String value = text(node, field);
if (value != null) {
return value;
}
}
return null;
}
private static String text(JsonNode node, String field) {
JsonNode value = node == null ? null : node.get(field);
return value == null || value.isNull() ? null : value.asText();
}
private static Long longValue(JsonNode node, String... fields) {
for (String field : fields) {
JsonNode value = node == null ? null : node.get(field);
if (value != null && value.isNumber()) {
return value.longValue();
}
}
return null;
}
private static String normalize(String product) {
return product == null ? "" : product.trim().toLowerCase(Locale.ROOT);
}
}

View File

@@ -0,0 +1,446 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationColumn;
import com.easyagents.federation.sql.execute.FederationExecutionGuard;
import com.easyagents.federation.sql.execute.FederationExecutionObserver;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.StatementLifecycle;
import java.io.FilterInputStream;
import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 持有 ResultSet、Statement、Connection 和 Engine 资源的 JDBC 流式游标。
*/
final class JdbcFederationResultCursor implements FederationResultCursor {
private final QueryId queryId;
private final List<FederationColumn> columns;
private final ResultSet resultSet;
private final PreparedStatement statement;
private final Connection connection;
private final StatementLifecycle statementLifecycle;
private final FederationExecutionGuard executionGuard;
private final FederationExecutionObserver observer;
private final AtomicBoolean closed = new AtomicBoolean();
private final AtomicBoolean firstRowObserved = new AtomicBoolean();
private final long resultSetCreatedNanos = System.nanoTime();
/**
* 创建 JDBC 流式游标。
*
* @param queryId 查询标识
* @param columns 结果列
* @param resultSet JDBC ResultSet
* @param statement JDBC Statement
* @param connection JDBC Connection
* @param statementLifecycle Statement 生命周期回调
* @param executionGuard 查询取消与截止时间检查器
* @param observer Fragment 执行阶段观察器
*/
JdbcFederationResultCursor(
QueryId queryId,
List<FederationColumn> columns,
ResultSet resultSet,
PreparedStatement statement,
Connection connection,
StatementLifecycle statementLifecycle,
FederationExecutionGuard executionGuard,
FederationExecutionObserver observer
) {
this.queryId = queryId;
this.columns = List.copyOf(columns);
this.resultSet = resultSet;
this.statement = statement;
this.connection = connection;
this.statementLifecycle = statementLifecycle;
this.executionGuard = executionGuard;
this.observer = observer;
}
/**
* 创建不采集阶段指标的兼容 JDBC 游标。
*
* @param queryId 查询标识
* @param columns 结果列
* @param resultSet JDBC ResultSet
* @param statement JDBC Statement
* @param connection JDBC Connection
* @param statementLifecycle Statement 生命周期
*/
JdbcFederationResultCursor(
QueryId queryId,
List<FederationColumn> columns,
ResultSet resultSet,
PreparedStatement statement,
Connection connection,
StatementLifecycle statementLifecycle
) {
this(
queryId,
columns,
resultSet,
statement,
connection,
statementLifecycle,
FederationExecutionGuard.none(),
FederationExecutionObserver.none()
);
}
/**
* 返回查询标识。
*
* @return 查询标识
*/
@Override
public QueryId queryId() {
return queryId;
}
/**
* 返回结果列。
*
* @return 结果列
*/
@Override
public List<FederationColumn> columns() {
return columns;
}
/**
* 移动到下一行;读取结束时保留资源直至调用方关闭游标。
*
* @return 是否存在下一行
*/
@Override
public boolean next() {
ensureOpen();
try {
boolean present = resultSet.next();
ensureAllowedAfterRead();
if (present && firstRowObserved.compareAndSet(false, true)) {
observer.firstRowAvailable(System.nanoTime() - resultSetCreatedNanos);
}
return present;
} catch (SQLException exception) {
closeWithSuppressed(exception);
throw JdbcFailureClassifier.classify(
statementLifecycle,
exception,
"JDBC result read timed out",
"JDBC query was cancelled",
"failed to advance JDBC result cursor"
);
}
}
/**
* 读取当前行指定列。
*
* @param columnIndex 从 1 开始的列序号
* @return 列值
*/
@Override
public Object getObject(int columnIndex) {
ensureOpen();
try {
Object value = resultSet.getObject(columnIndex);
ensureAllowedAfterRead();
return value;
} catch (SQLException exception) {
closeWithSuppressed(exception);
throw JdbcFailureClassifier.classify(
statementLifecycle,
exception,
"JDBC result read timed out",
"JDBC query was cancelled",
"failed to read JDBC result column " + columnIndex
);
}
}
/**
* 以 JDBC 流读取二进制列。
*
* @param columnIndex 从 1 开始的列序号
* @return 二进制流SQL NULL 返回 null
*/
@Override
public InputStream getBinaryStream(int columnIndex) {
ensureOpen();
try {
InputStream stream = resultSet.getBinaryStream(columnIndex);
ensureAllowedAfterRead();
return stream == null ? null : new GuardedInputStream(stream, columnIndex);
} catch (SQLException exception) {
throw readFailure(columnIndex, exception);
}
}
/**
* 以 JDBC 流读取字符列。
*
* @param columnIndex 从 1 开始的列序号
* @return 字符流SQL NULL 返回 null
*/
@Override
public Reader getCharacterStream(int columnIndex) {
ensureOpen();
try {
Reader reader = resultSet.getCharacterStream(columnIndex);
ensureAllowedAfterRead();
return reader == null ? null : new GuardedReader(reader, columnIndex);
} catch (SQLException exception) {
throw readFailure(columnIndex, exception);
}
}
/**
* 复制当前行Engine 不缓存返回行。
*
* @return 当前行列值
*/
@Override
public List<Object> row() {
ensureOpen();
List<Object> row = new ArrayList<>(columns.size());
for (int index = 1; index <= columns.size(); index++) {
row.add(getObject(index));
}
return Collections.unmodifiableList(row);
}
private void ensureOpen() {
// 异步关闭可能先于消费线程到达,优先保留取消或超时终态语义。
try {
executionGuard.ensureAllowed();
} catch (RuntimeException exception) {
closeWithSuppressed(exception);
throw exception;
}
if (closed.get()) {
throw new FederationSqlException(
FederationSqlErrorCode.EXECUTION_FAILED,
"result cursor is closed"
);
}
}
private void ensureAllowedAfterRead() {
try {
executionGuard.ensureAllowed();
} catch (RuntimeException exception) {
closeWithSuppressed(exception);
throw exception;
}
}
/**
* 幂等关闭 JDBC 资源并最终释放准入许可与 Runtime lease。
*/
@Override
public void close() {
if (!closed.compareAndSet(false, true)) {
return;
}
FederationSqlException failure = null;
try {
resultSet.close();
} catch (SQLException exception) {
failure = closeFailure("ResultSet", exception);
}
try {
statementLifecycle.unregister(statement);
} catch (RuntimeException exception) {
failure = append(failure, closeFailure("Statement lifecycle", exception));
}
try {
statement.close();
} catch (SQLException exception) {
failure = append(failure, closeFailure("PreparedStatement", exception));
}
try {
connection.close();
} catch (SQLException exception) {
failure = append(failure, closeFailure("Connection", exception));
}
if (failure != null) {
throw failure;
}
}
private void closeWithSuppressed(Throwable original) {
try {
close();
} catch (RuntimeException closeException) {
original.addSuppressed(closeException);
}
}
/**
* 将流式列读取异常映射为统一错误并确定性关闭 JDBC 资源。
*
* @param columnIndex 列序号
* @param exception JDBC 或流读取异常
* @return 统一 Federation 异常
*/
private FederationSqlException readFailure(int columnIndex, Throwable exception) {
closeWithSuppressed(exception);
return JdbcFailureClassifier.classify(
statementLifecycle,
exception,
"JDBC result read timed out",
"JDBC query was cancelled",
"failed to stream JDBC result column " + columnIndex
);
}
/**
* 对二进制列的每次实际读取执行查询终态检查。
*/
private final class GuardedInputStream extends FilterInputStream {
private final int columnIndex;
/**
* 创建受查询生命周期保护的二进制流。
*
* @param delegate JDBC 驱动流
* @param columnIndex 列序号
*/
private GuardedInputStream(InputStream delegate, int columnIndex) {
super(delegate);
this.columnIndex = columnIndex;
}
/** {@inheritDoc} */
@Override
public int read() throws IOException {
ensureOpen();
try {
int value = super.read();
ensureAllowedAfterRead();
return value;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
/** {@inheritDoc} */
@Override
public int read(byte[] buffer, int offset, int length) throws IOException {
ensureOpen();
try {
int read = super.read(buffer, offset, length);
ensureAllowedAfterRead();
return read;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
/** {@inheritDoc} */
@Override
public long skip(long count) throws IOException {
ensureOpen();
try {
long skipped = super.skip(count);
ensureAllowedAfterRead();
return skipped;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
}
/**
* 对字符列的每次实际读取执行查询终态检查。
*/
private final class GuardedReader extends FilterReader {
private final int columnIndex;
/**
* 创建受查询生命周期保护的字符流。
*
* @param delegate JDBC 驱动 Reader
* @param columnIndex 列序号
*/
private GuardedReader(Reader delegate, int columnIndex) {
super(delegate);
this.columnIndex = columnIndex;
}
/** {@inheritDoc} */
@Override
public int read() throws IOException {
ensureOpen();
try {
int value = super.read();
ensureAllowedAfterRead();
return value;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
/** {@inheritDoc} */
@Override
public int read(char[] buffer, int offset, int length) throws IOException {
ensureOpen();
try {
int read = super.read(buffer, offset, length);
ensureAllowedAfterRead();
return read;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
/** {@inheritDoc} */
@Override
public long skip(long count) throws IOException {
ensureOpen();
try {
long skipped = super.skip(count);
ensureAllowedAfterRead();
return skipped;
} catch (IOException exception) {
throw readFailure(columnIndex, exception);
}
}
}
private static FederationSqlException closeFailure(String resource, Exception cause) {
return new FederationSqlException(
FederationSqlErrorCode.RESOURCE_CLOSE_FAILED,
"failed to close JDBC " + resource,
cause
);
}
private static FederationSqlException append(
FederationSqlException failure,
FederationSqlException next
) {
if (failure == null) {
return next;
}
failure.addSuppressed(next);
return failure;
}
}

View File

@@ -0,0 +1,212 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.adapter.AdapterDialectContext;
import com.easyagents.federation.sql.adapter.AdapterHints;
import com.easyagents.federation.sql.adapter.AdapterSchemaContext;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterProvider;
import com.easyagents.federation.sql.adapter.FederationStatisticsCollector;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.execute.FederationFragmentExplainer;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.util.Locale;
import java.util.Optional;
import java.util.Set;
import org.apache.calcite.adapter.jdbc.JdbcConvention;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.Schemas;
import org.apache.calcite.sql.SqlDialect;
import org.apache.calcite.sql.SqlDialectFactoryImpl;
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
import org.apache.calcite.sql.dialect.MysqlSqlDialect;
/**
* MySQL、PostgreSQL、Oracle 及显式实验 ANSI 数据库的默认 JDBC Adapter。
*/
public final class JdbcFederationSqlAdapterProvider implements FederationSqlAdapterProvider {
/** 默认 JDBC Adapter 标识。 */
public static final String ADAPTER_ID = "jdbc";
/** 允许未知数据库采用实验 ANSI 方言的 Definition 选项。 */
public static final String EXPERIMENTAL_ANSI_OPTION = "experimentalAnsi";
private static final Set<String> SUPPORTED_PRODUCTS = Set.of(
"mysql",
"postgresql",
"oracle",
"h2"
);
private final FederationFragmentExecutor executor = new JdbcFederationFragmentExecutor();
private final FederationFragmentExplainer explainer = new JdbcFederationFragmentExplainer();
private final FederationStatisticsCollector statisticsCollector =
new JdbcFederationStatisticsCollector();
/**
* 创建默认 JDBC Adapter Provider。
*/
public JdbcFederationSqlAdapterProvider() {
}
/**
* 返回默认 Adapter 标识。
*
* @return {@value #ADAPTER_ID}
*/
@Override
public String adapterId() {
return ADAPTER_ID;
}
/**
* 基于数据库产品名判断内建或实验 ANSI 支持。
*
* @param metadata JDBC 元数据
* @param hints Adapter 提示
* @return 是否支持
* @throws SQLException 元数据读取失败
*/
@Override
public boolean supports(DatabaseMetaData metadata, AdapterHints hints) throws SQLException {
return SUPPORTED_PRODUCTS.contains(normalize(metadata.getDatabaseProductName()))
|| hints.enabled(EXPERIMENTAL_ANSI_OPTION);
}
/**
* 返回与实际验证证据一致的兼容性状态。
*
* @param metadata JDBC 元数据
* @param hints Adapter 提示
* @return 兼容性说明
* @throws SQLException 元数据读取失败
*/
@Override
public AdapterCompatibility compatibility(DatabaseMetaData metadata, AdapterHints hints) throws SQLException {
String product = metadata.getDatabaseProductName();
String normalized = normalize(product);
AdapterCompatibilityStatus status;
String diagnostic;
if ("h2".equals(normalized)) {
status = AdapterCompatibilityStatus.VERIFIED;
diagnostic = "verified by module-level H2 integration tests";
} else if (SUPPORTED_PRODUCTS.contains(normalized)) {
status = AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED;
diagnostic = "dialect is supported by code; verify against the target database version before production";
} else if (hints.enabled(EXPERIMENTAL_ANSI_OPTION)) {
status = AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED;
diagnostic = "experimental ANSI mode is enabled for an unrecognized database";
} else {
status = AdapterCompatibilityStatus.UNSUPPORTED;
diagnostic = "database product is not recognized";
}
return new AdapterCompatibility(
status,
product,
metadata.getDatabaseProductVersion(),
metadata.getDriverName(),
metadata.getDriverVersion(),
diagnostic
);
}
/**
* 创建复用已探测 Dialect 和调用方 DataSource 的 JdbcSchema。
*
* @param context Schema 上下文
* @return Calcite JdbcSchema
*/
@Override
public Schema createSchema(AdapterSchemaContext context) {
if (!(context.schemaDefinition() instanceof JdbcSchemaDefinition definition)) {
throw new FederationSqlException(
FederationSqlErrorCode.INVALID_ARGUMENT,
"jdbc adapter requires JdbcSchemaDefinition"
);
}
JdbcConvention convention = JdbcConvention.of(
context.dialect(),
Schemas.subSchemaExpression(
context.parentSchema(),
definition.logicalName(),
JdbcSchema.class
),
context.sourceDefinition().sourceId().value() + "." + definition.logicalName()
);
Schema schema = new JdbcSchema(
context.handle().dataSource(),
context.dialect(),
convention,
definition.catalog(),
definition.physicalSchema()
);
// MySQL 表名可区分大小写而列名始终不区分大小写,需分别建模。
return context.dialect() instanceof MysqlSqlDialect
? new MysqlCaseInsensitiveColumnSchema(schema)
: schema;
}
/**
* 使用 Calcite 官方 DialectFactory 选择方言,未知数据库仅在显式 ANSI 模式下放行。
*
* @param context 方言上下文
* @return SqlDialect
* @throws SQLException 元数据读取失败
*/
@Override
public SqlDialect createDialect(AdapterDialectContext context) throws SQLException {
String product = normalize(context.metadata().getDatabaseProductName());
if (!SUPPORTED_PRODUCTS.contains(product)) {
AdapterHints hints = new AdapterHints(context.sourceDefinition().adapterOptions());
if (hints.enabled(EXPERIMENTAL_ANSI_OPTION)) {
return AnsiSqlDialect.DEFAULT;
}
throw new FederationSqlException(
FederationSqlErrorCode.ADAPTER_UNSUPPORTED,
"database product is not supported by jdbc adapter: "
+ context.metadata().getDatabaseProductName()
);
}
return SqlDialectFactoryImpl.INSTANCE.create(context.metadata());
}
/**
* 返回直接 JDBC 流式执行器。
*
* @return Fragment 执行器
*/
@Override
public FederationFragmentExecutor fragmentExecutor() {
return executor;
}
/**
* 返回 MySQL、PostgreSQL 和 H2 的显式物理 Explain 实现。
*
* @return JDBC 物理 Explain SPI
*/
@Override
public Optional<FederationFragmentExplainer> fragmentExplainer() {
return Optional.of(explainer);
}
/**
* 返回 MySQL 与 PostgreSQL 的内建目录统计采集器。
*
* <p>Oracle、H2 和实验 ANSI 数据库当前返回空统计,由引擎使用默认成本估算。</p>
*
* @return JDBC 统计采集 SPI
*/
@Override
public Optional<FederationStatisticsCollector> statisticsCollector() {
return Optional.of(statisticsCollector);
}
private static String normalize(String productName) {
return productName == null ? "" : productName.trim().toLowerCase(Locale.ROOT);
}
}

View File

@@ -0,0 +1,641 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.FederationStatisticsCollectionContext;
import com.easyagents.federation.sql.adapter.FederationStatisticsCollector;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.federation.FederationColumnStatistics;
import com.easyagents.federation.sql.federation.FederationStatisticsSnapshot;
import com.easyagents.federation.sql.federation.FederationStatisticsStatus;
import com.easyagents.federation.sql.federation.FederationTableStatistics;
import com.easyagents.federation.sql.source.FederationSchemaDefinition;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* MySQL 与 PostgreSQL 的批量 JDBC 目录统计采集器。
*/
final class JdbcFederationStatisticsCollector implements FederationStatisticsCollector {
private static final Logger LOG = LoggerFactory.getLogger(
JdbcFederationStatisticsCollector.class
);
/**
* 根据 JDBC 数据库产品分派内建统计采集逻辑。
*
* @param context 统计采集上下文
* @return 表统计映射
* @throws SQLException 目录或 JDBC 元数据读取失败
*/
@Override
public Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> collect(
FederationStatisticsCollectionContext context
) throws SQLException {
String product = normalize(
context.connection().getMetaData().getDatabaseProductName()
);
List<JdbcSchemaDefinition> schemas = jdbcSchemas(context);
return switch (product) {
case "mysql" -> collectMysql(context, schemas);
case "postgresql" -> collectPostgresql(context, schemas);
default -> Map.of();
};
}
/**
* 批量读取 MySQL INFORMATION_SCHEMA 表统计和主键。
*
* @param context 采集上下文
* @param schemas JDBC Schema 映射
* @return MySQL 表统计
* @throws SQLException 目录读取失败
*/
private Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> collectMysql(
FederationStatisticsCollectionContext context,
List<JdbcSchemaDefinition> schemas
) throws SQLException {
Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> statistics =
new LinkedHashMap<>();
for (JdbcSchemaDefinition schema : schemas) {
String catalog = textOr(schema.catalog(), context.connection().getCatalog());
if (catalog == null || catalog.isBlank()) {
continue;
}
Map<String, ColumnLayout> layouts = readColumnLayouts(
context.connection().getMetaData(),
catalog,
schema.physicalSchema()
);
Map<String, List<String>> primaryKeys = readMysqlPrimaryKeys(
context,
catalog
);
String sql = "SELECT TABLE_NAME, TABLE_ROWS, AVG_ROW_LENGTH "
+ "FROM INFORMATION_SCHEMA.TABLES "
+ "WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'";
try (PreparedStatement statement = context.connection().prepareStatement(sql)) {
statement.setQueryTimeout(context.queryTimeoutSeconds());
statement.setString(1, catalog);
try (ResultSet result = statement.executeQuery()) {
while (result.next()) {
String table = result.getString("TABLE_NAME");
ColumnLayout layout = layouts.getOrDefault(
normalize(table),
ColumnLayout.empty()
);
long averageWidth = result.getLong("AVG_ROW_LENGTH");
if (averageWidth <= 0L) {
averageWidth = layout.fallbackWidthBytes();
}
FederationStatisticsSnapshot.TableKey key =
new FederationStatisticsSnapshot.TableKey(
context.sourceDefinition().sourceId(),
schema.logicalName(),
table
);
statistics.put(key, new FederationTableStatistics(
Math.max(0D, result.getDouble("TABLE_ROWS")),
Math.max(1L, averageWidth),
context.collectedAt(),
"database-catalog:mysql",
Map.of(),
uniqueKey(primaryKeys.get(normalize(table))),
context.expiresAt(),
FederationStatisticsStatus.PARTIAL
));
}
}
}
}
return Map.copyOf(statistics);
}
/**
* 批量读取 PostgreSQL 表行数、列分布和主键统计。
*
* @param context 采集上下文
* @param schemas JDBC Schema 映射
* @return PostgreSQL 表统计
* @throws SQLException 表级目录读取失败
*/
private Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics>
collectPostgresql(
FederationStatisticsCollectionContext context,
List<JdbcSchemaDefinition> schemas
) throws SQLException {
Map<String, JdbcSchemaDefinition> schemasByPhysical = new LinkedHashMap<>();
Map<String, ColumnLayout> layouts = new LinkedHashMap<>();
for (JdbcSchemaDefinition schema : schemas) {
String physical = textOr(schema.physicalSchema(), context.connection().getSchema());
if (physical == null || physical.isBlank()) {
physical = "public";
}
schemasByPhysical.putIfAbsent(normalize(physical), schema);
Map<String, ColumnLayout> schemaLayouts = readColumnLayouts(
context.connection().getMetaData(),
schema.catalog(),
physical
);
String resolvedPhysical = physical;
schemaLayouts.forEach((table, layout) -> layouts.put(
tableKey(resolvedPhysical, table),
layout
));
}
if (schemasByPhysical.isEmpty()) {
return Map.of();
}
Map<String, TableEstimate> estimates = readPostgresqlTableEstimates(
context,
schemasByPhysical.keySet()
);
Map<String, Map<String, FederationColumnStatistics>> columns;
try {
columns = readPostgresqlColumnStatistics(
context,
schemasByPhysical.keySet(),
estimates
);
} catch (SQLException exception) {
LOG.warn(
"PostgreSQL column statistics are unavailable; retaining table estimates, sourceId={}",
context.sourceDefinition().sourceId(),
exception
);
columns = Map.of();
}
Map<String, List<String>> primaryKeys;
try {
primaryKeys = readPostgresqlPrimaryKeys(
context,
schemasByPhysical.keySet()
);
} catch (SQLException exception) {
LOG.warn(
"PostgreSQL primary-key statistics are unavailable, sourceId={}",
context.sourceDefinition().sourceId(),
exception
);
primaryKeys = Map.of();
}
Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> statistics =
new LinkedHashMap<>();
for (Map.Entry<String, TableEstimate> entry : estimates.entrySet()) {
TableEstimate estimate = entry.getValue();
JdbcSchemaDefinition schema = schemasByPhysical.get(normalize(estimate.schema()));
if (schema == null) {
continue;
}
ColumnLayout layout = layouts.getOrDefault(entry.getKey(), ColumnLayout.empty());
Map<String, FederationColumnStatistics> tableColumns = columns.getOrDefault(
entry.getKey(),
Map.of()
);
long averageWidth = Math.max(
layout.fallbackWidthBytes(),
averageColumnWidth(tableColumns)
);
boolean complete = !layout.columns().isEmpty()
&& containsAllIgnoreCase(tableColumns.keySet(), layout.columns());
FederationStatisticsSnapshot.TableKey key =
new FederationStatisticsSnapshot.TableKey(
context.sourceDefinition().sourceId(),
schema.logicalName(),
estimate.table()
);
statistics.put(key, new FederationTableStatistics(
estimate.estimatedRows(),
Math.max(1L, averageWidth),
context.collectedAt(),
"database-catalog:postgresql",
tableColumns,
uniqueKey(primaryKeys.get(entry.getKey())),
context.expiresAt(),
complete
? FederationStatisticsStatus.COMPLETE
: FederationStatisticsStatus.PARTIAL
));
}
return Map.copyOf(statistics);
}
/**
* 读取 Definition 中的 JDBC Schema 映射并拒绝不匹配的定义类型。
*
* @param context 采集上下文
* @return JDBC Schema 定义
*/
private List<JdbcSchemaDefinition> jdbcSchemas(
FederationStatisticsCollectionContext context
) {
List<JdbcSchemaDefinition> schemas = new ArrayList<>();
for (FederationSchemaDefinition schema : context.sourceDefinition().schemas()) {
if (!(schema instanceof JdbcSchemaDefinition jdbcSchema)) {
throw new FederationSqlException(
FederationSqlErrorCode.INVALID_ARGUMENT,
"jdbc statistics collector requires JdbcSchemaDefinition"
);
}
schemas.add(jdbcSchema);
}
return List.copyOf(schemas);
}
/**
* 通过 JDBC 元数据按 Schema 批量读取字段布局。
*
* @param metadata JDBC 元数据
* @param catalog 物理 Catalog
* @param schema 物理 Schema
* @return 按规范化表名索引的字段布局
* @throws SQLException 元数据读取失败
*/
private Map<String, ColumnLayout> readColumnLayouts(
DatabaseMetaData metadata,
String catalog,
String schema
) throws SQLException {
Map<String, MutableColumnLayout> layouts = new LinkedHashMap<>();
try (ResultSet result = metadata.getColumns(catalog, schema, "%", "%")) {
while (result.next()) {
String table = normalize(result.getString("TABLE_NAME"));
MutableColumnLayout layout = layouts.computeIfAbsent(
table,
ignored -> new MutableColumnLayout()
);
layout.columns.add(result.getString("COLUMN_NAME"));
layout.fallbackWidthBytes = saturatedAdd(
layout.fallbackWidthBytes,
estimatedJdbcWidth(result.getInt("DATA_TYPE"))
);
}
}
Map<String, ColumnLayout> frozen = new LinkedHashMap<>();
layouts.forEach((table, layout) -> frozen.put(
table,
new ColumnLayout(
Math.max(1L, layout.fallbackWidthBytes),
Set.copyOf(layout.columns)
)
));
return Map.copyOf(frozen);
}
/**
* 一次查询一个 MySQL Catalog 的全部主键字段。
*
* @param context 采集上下文
* @param catalog 物理 Catalog
* @return 按规范化表名索引的有序主键
* @throws SQLException 目录读取失败
*/
private Map<String, List<String>> readMysqlPrimaryKeys(
FederationStatisticsCollectionContext context,
String catalog
) throws SQLException {
String sql = "SELECT TABLE_NAME, COLUMN_NAME "
+ "FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE "
+ "WHERE TABLE_SCHEMA = ? AND CONSTRAINT_NAME = 'PRIMARY' "
+ "ORDER BY TABLE_NAME, ORDINAL_POSITION";
Map<String, List<String>> keys = new LinkedHashMap<>();
try (PreparedStatement statement = context.connection().prepareStatement(sql)) {
statement.setQueryTimeout(context.queryTimeoutSeconds());
statement.setString(1, catalog);
try (ResultSet result = statement.executeQuery()) {
while (result.next()) {
keys.computeIfAbsent(
normalize(result.getString("TABLE_NAME")),
ignored -> new ArrayList<>()
).add(result.getString("COLUMN_NAME"));
}
}
}
return freezeLists(keys);
}
/**
* 读取 PostgreSQL 表级近似行数。
*
* @param context 采集上下文
* @param schemas 物理 Schema
* @return 表级估算
* @throws SQLException 目录读取失败
*/
private Map<String, TableEstimate> readPostgresqlTableEstimates(
FederationStatisticsCollectionContext context,
Set<String> schemas
) throws SQLException {
String sql = "SELECT n.nspname AS schema_name, c.relname AS table_name, "
+ "GREATEST(c.reltuples, 0)::double precision AS estimated_rows "
+ "FROM pg_catalog.pg_class c "
+ "JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace "
+ "WHERE c.relkind IN ('r', 'p') AND lower(n.nspname) IN ("
+ placeholders(schemas.size()) + ")";
Map<String, TableEstimate> estimates = new LinkedHashMap<>();
try (PreparedStatement statement = context.connection().prepareStatement(sql)) {
statement.setQueryTimeout(context.queryTimeoutSeconds());
bind(statement, schemas);
try (ResultSet result = statement.executeQuery()) {
while (result.next()) {
String schema = result.getString("schema_name");
String table = result.getString("table_name");
estimates.put(
tableKey(schema, table),
new TableEstimate(
schema,
table,
Math.max(0D, result.getDouble("estimated_rows"))
)
);
}
}
}
return Map.copyOf(estimates);
}
/**
* 读取 PostgreSQL 列分布统计。
*
* @param context 采集上下文
* @param schemas 物理 Schema
* @param estimates 已读取的表级估算
* @return 按物理表索引的列统计
* @throws SQLException 目录读取失败
*/
private Map<String, Map<String, FederationColumnStatistics>>
readPostgresqlColumnStatistics(
FederationStatisticsCollectionContext context,
Set<String> schemas,
Map<String, TableEstimate> estimates
) throws SQLException {
String sql = "SELECT schemaname, tablename, attname, null_frac, n_distinct, avg_width "
+ "FROM pg_catalog.pg_stats WHERE lower(schemaname) IN ("
+ placeholders(schemas.size()) + ")";
Map<String, Map<String, FederationColumnStatistics>> columns = new LinkedHashMap<>();
try (PreparedStatement statement = context.connection().prepareStatement(sql)) {
statement.setQueryTimeout(context.queryTimeoutSeconds());
bind(statement, schemas);
try (ResultSet result = statement.executeQuery()) {
while (result.next()) {
String key = tableKey(
result.getString("schemaname"),
result.getString("tablename")
);
TableEstimate table = estimates.get(key);
if (table == null) {
continue;
}
double rawDistinct = result.getDouble("n_distinct");
double distinct = rawDistinct < 0D
? Math.abs(rawDistinct) * table.estimatedRows()
: rawDistinct;
if (!Double.isFinite(distinct)) {
distinct = 0D;
}
columns.computeIfAbsent(key, ignored -> new LinkedHashMap<>()).put(
result.getString("attname"),
new FederationColumnStatistics(
Math.max(0D, distinct),
Math.max(0D, Math.min(1D, result.getDouble("null_frac"))),
Math.max(0L, result.getLong("avg_width"))
)
);
}
}
}
Map<String, Map<String, FederationColumnStatistics>> frozen = new LinkedHashMap<>();
columns.forEach((table, values) -> frozen.put(table, Map.copyOf(values)));
return Map.copyOf(frozen);
}
/**
* 一次读取多个 PostgreSQL Schema 的主键字段。
*
* @param context 采集上下文
* @param schemas 物理 Schema
* @return 按物理表索引的主键字段
* @throws SQLException 目录读取失败
*/
private Map<String, List<String>> readPostgresqlPrimaryKeys(
FederationStatisticsCollectionContext context,
Set<String> schemas
) throws SQLException {
String sql = "SELECT n.nspname AS schema_name, c.relname AS table_name, "
+ "a.attname AS column_name "
+ "FROM pg_catalog.pg_index i "
+ "JOIN pg_catalog.pg_class c ON c.oid = i.indrelid "
+ "JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace "
+ "JOIN pg_catalog.pg_attribute a "
+ "ON a.attrelid = c.oid AND a.attnum = ANY(i.indkey) "
+ "WHERE i.indisprimary AND lower(n.nspname) IN ("
+ placeholders(schemas.size()) + ") "
+ "ORDER BY n.nspname, c.relname, a.attnum";
Map<String, List<String>> keys = new LinkedHashMap<>();
try (PreparedStatement statement = context.connection().prepareStatement(sql)) {
statement.setQueryTimeout(context.queryTimeoutSeconds());
bind(statement, schemas);
try (ResultSet result = statement.executeQuery()) {
while (result.next()) {
keys.computeIfAbsent(
tableKey(
result.getString("schema_name"),
result.getString("table_name")
),
ignored -> new ArrayList<>()
).add(result.getString("column_name"));
}
}
}
return freezeLists(keys);
}
/**
* 将可选主键转换为唯一键列表。
*
* @param primaryKey 主键字段
* @return 零个或一个唯一键
*/
private List<List<String>> uniqueKey(List<String> primaryKey) {
return primaryKey == null || primaryKey.isEmpty()
? List.of()
: List.of(List.copyOf(primaryKey));
}
/**
* 汇总列平均宽度并防止 long 溢出。
*
* @param columns 列统计
* @return 至少为 1 的平均宽度
*/
private long averageColumnWidth(Map<String, FederationColumnStatistics> columns) {
long width = 0L;
for (FederationColumnStatistics column : columns.values()) {
width = saturatedAdd(width, column.averageWidthBytes());
}
return Math.max(1L, width);
}
/**
* 判断列统计是否覆盖全部字段。
*
* @param available 已有列统计名称
* @param required JDBC 字段名称
* @return 完整覆盖时为 true
*/
private boolean containsAllIgnoreCase(Set<String> available, Set<String> required) {
Set<String> normalized = new LinkedHashSet<>();
available.forEach(value -> normalized.add(normalize(value)));
return required.stream().map(this::normalize).allMatch(normalized::contains);
}
/**
* 估算 JDBC 类型的保守内存宽度。
*
* @param jdbcType JDBC 类型
* @return 估算字节数
*/
private long estimatedJdbcWidth(int jdbcType) {
return switch (jdbcType) {
case Types.BOOLEAN, Types.BIT, Types.TINYINT -> 1L;
case Types.SMALLINT -> 2L;
case Types.INTEGER, Types.REAL, Types.FLOAT, Types.DATE -> 4L;
case Types.BIGINT, Types.DOUBLE, Types.TIMESTAMP,
Types.TIMESTAMP_WITH_TIMEZONE, Types.TIME,
Types.TIME_WITH_TIMEZONE -> 8L;
case Types.DECIMAL, Types.NUMERIC -> 16L;
case Types.BINARY, Types.VARBINARY, Types.LONGVARBINARY,
Types.BLOB, Types.CLOB, Types.NCLOB,
Types.LONGVARCHAR, Types.LONGNVARCHAR -> 64L;
default -> 32L;
};
}
/**
* 生成固定数量的 PreparedStatement 占位符。
*
* @param size 占位符数量
* @return 逗号分隔占位符
*/
private String placeholders(int size) {
return String.join(", ", Collections.nCopies(size, "?"));
}
/**
* 按稳定顺序绑定规范化 Schema。
*
* @param statement PreparedStatement
* @param schemas Schema 集合
* @throws SQLException 参数绑定失败
*/
private void bind(PreparedStatement statement, Set<String> schemas) throws SQLException {
int index = 1;
for (String schema : schemas) {
statement.setString(index++, normalize(schema));
}
}
/**
* 冻结可变列表映射。
*
* @param source 可变列表映射
* @return 不可变列表映射
*/
private Map<String, List<String>> freezeLists(Map<String, List<String>> source) {
Map<String, List<String>> frozen = new LinkedHashMap<>();
source.forEach((key, value) -> frozen.put(key, List.copyOf(value)));
return Map.copyOf(frozen);
}
/**
* 生成大小写不敏感的物理表索引键。
*
* @param schema 物理 Schema
* @param table 物理表
* @return 稳定索引键
*/
private String tableKey(String schema, String table) {
return normalize(schema) + '\u0000' + normalize(table);
}
/**
* 返回首个非空文本。
*
* @param primary 首选值
* @param fallback 备用值
* @return 可空结果
*/
private String textOr(String primary, String fallback) {
return primary == null || primary.isBlank() ? fallback : primary;
}
/**
* 规范化数据库产品名或标识符。
*
* @param value 原始值
* @return 小写非空值
*/
private String normalize(String value) {
return value == null ? "" : value.trim().toLowerCase(Locale.ROOT);
}
/**
* 饱和 long 加法。
*
* @param left 左值
* @param right 右值
* @return 不溢出的和
*/
private long saturatedAdd(long left, long right) {
return Long.MAX_VALUE - left < right ? Long.MAX_VALUE : left + right;
}
/**
* 单表字段布局。
*
* @param fallbackWidthBytes JDBC 类型估算行宽
* @param columns 字段名称
*/
private record ColumnLayout(long fallbackWidthBytes, Set<String> columns) {
/**
* 创建空布局。
*
* @return 保守空布局
*/
private static ColumnLayout empty() {
return new ColumnLayout(1L, Set.of());
}
}
/** 可变字段布局构造器。 */
private static final class MutableColumnLayout {
private long fallbackWidthBytes;
private final Set<String> columns = new LinkedHashSet<>();
}
/**
* PostgreSQL 表级估算。
*
* @param schema 物理 Schema
* @param table 物理表
* @param estimatedRows 估算行数
*/
private record TableEstimate(String schema, String table, double estimatedRows) {
}
}

View File

@@ -0,0 +1,38 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.source.FederationSchemaDefinition;
import java.util.Arrays;
import java.util.List;
/**
* JDBC Catalog/Schema 到逻辑 Schema 的映射定义。
*
* @param logicalName SQL 中使用的逻辑 Schema 名称
* @param catalog 物理 Catalog可为空
* @param physicalSchema 物理 Schema可为空
*/
public record JdbcSchemaDefinition(
String logicalName,
String catalog,
String physicalSchema
) implements FederationSchemaDefinition {
/**
* 校验逻辑名称并保留可空物理 Catalog/Schema。
*/
public JdbcSchemaDefinition {
if (logicalName == null || logicalName.isBlank()) {
throw new IllegalArgumentException("logicalName must not be blank");
}
}
/**
* 返回 JDBC Schema 映射的稳定校验和材料。
*
* @return 稳定材料
*/
@Override
public List<String> checksumFields() {
return Arrays.asList(catalog, physicalSchema);
}
}

View File

@@ -0,0 +1,308 @@
package com.easyagents.federation.sql.adapter.jdbc;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.adapter.jdbc.JdbcTable;
import org.apache.calcite.config.CalciteConnectionConfig;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rel.type.RelRecordType;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaVersion;
import org.apache.calcite.schema.Statistic;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.TranslatableTable;
import org.apache.calcite.schema.Wrapper;
import org.apache.calcite.schema.impl.DelegatingSchema;
import org.apache.calcite.schema.lookup.IgnoreCaseLookup;
import org.apache.calcite.schema.lookup.LikePattern;
import org.apache.calcite.schema.lookup.Lookup;
import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlNode;
/**
* 保留 MySQL 表名精确匹配,同时让列名遵循 MySQL 的大小写不敏感语义。
*/
final class MysqlCaseInsensitiveColumnSchema extends DelegatingSchema {
private final Lookup<Table> tableLookup;
/**
* 创建 MySQL 列名语义包装器。
*
* @param schema 原始 JDBC Schema
*/
MysqlCaseInsensitiveColumnSchema(Schema schema) {
super(Objects.requireNonNull(schema, "schema"));
Lookup<Table> sourceLookup = schema.tables();
if (schema instanceof JdbcSchema jdbcSchema) {
sourceLookup = new ExactJdbcTableLookup(jdbcSchema, sourceLookup);
}
this.tableLookup = sourceLookup.map((table, ignoredName) -> wrap(table));
}
/**
* 返回保持原始表名 Lookup 规则的包装表集合。
*
* @return 包装后的表 Lookup
*/
@Override
public Lookup<Table> tables() {
return tableLookup;
}
/**
* 按原始 Schema 规则精确获取表,再包装列类型。
*
* @param name 表名
* @return 包装表;不存在时返回 null
*/
@Override
public Table getTable(String name) {
return tableLookup.get(name);
}
/**
* 为 Schema 快照保留相同的列名语义。
*
* @param version Schema 版本
* @return 包装后的快照
*/
@Override
public Schema snapshot(SchemaVersion version) {
return new MysqlCaseInsensitiveColumnSchema(schema.snapshot(version));
}
private static Table wrap(Table table) {
return table instanceof MysqlCaseInsensitiveColumnTable
? table
: new MysqlCaseInsensitiveColumnTable(table);
}
/**
* 将 Calcite 的表名查找收紧为 JDBC 元数据层面的精确查找。
*/
private static final class ExactJdbcTableLookup extends IgnoreCaseLookup<Table> {
private final JdbcSchema jdbcSchema;
private final Lookup<Table> delegate;
private volatile boolean searchEscapeLoaded;
private String searchEscape;
private ExactJdbcTableLookup(JdbcSchema jdbcSchema, Lookup<Table> delegate) {
this.jdbcSchema = Objects.requireNonNull(jdbcSchema, "jdbcSchema");
this.delegate = Objects.requireNonNull(delegate, "delegate");
}
/**
* 转义 JDBC LIKE 通配字符后读取,并校验驱动返回的真实物理表名。
*
* @param name 精确表名
* @return 精确匹配的表;不存在时返回 null
*/
@Override
public Table get(String name) {
Table table = delegate.get(escapePattern(name));
if (table == null) {
return null;
}
JdbcTable jdbcTable = table instanceof JdbcTable direct
? direct
: table instanceof Wrapper wrapper
? wrapper.unwrap(JdbcTable.class)
: null;
return jdbcTable != null && name.equals(jdbcTable.jdbcTableName) ? table : null;
}
/**
* 返回符合 Calcite LIKE 语义的表名,过滤 JDBC 对下划线的额外通配匹配。
*
* @param pattern 表名模式
* @return 匹配名称集合
*/
@Override
public Set<String> getNames(LikePattern pattern) {
return delegate.getNames(pattern).stream()
.filter(pattern.matcher()::apply)
.collect(Collectors.toUnmodifiableSet());
}
private String escapePattern(String name) {
String escape = searchEscape();
if (escape == null || escape.isEmpty()) {
if (name.indexOf('_') >= 0 || name.indexOf('%') >= 0) {
throw new IllegalStateException(
"MySQL JDBC driver does not expose a metadata search escape"
);
}
return name;
}
return name
.replace(escape, escape + escape)
.replace("_", escape + "_")
.replace("%", escape + "%");
}
private String searchEscape() {
if (searchEscapeLoaded) {
return searchEscape;
}
synchronized (this) {
if (!searchEscapeLoaded) {
try (Connection connection = jdbcSchema.getDataSource().getConnection()) {
searchEscape = connection.getMetaData().getSearchStringEscape();
searchEscapeLoaded = true;
} catch (SQLException exception) {
throw new IllegalStateException(
"Failed to read MySQL JDBC metadata search escape",
exception
);
}
}
return searchEscape;
}
}
}
/**
* 仅调整行类型的字段查找规则,关系转换继续交由原始 JDBC Table 完成。
*/
private static final class MysqlCaseInsensitiveColumnTable
implements TranslatableTable, Wrapper {
private final Table delegate;
private MysqlCaseInsensitiveColumnTable(Table delegate) {
this.delegate = Objects.requireNonNull(delegate, "delegate");
}
/**
* 返回列名大小写不敏感的结构类型。
*
* @param typeFactory Calcite 类型工厂
* @return 包装后的结构类型
*/
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory) {
return new CaseInsensitiveRelRecordType(delegate.getRowType(typeFactory));
}
/**
* 复用原始表统计信息。
*
* @return 表统计信息
*/
@Override
public Statistic getStatistic() {
return delegate.getStatistic();
}
/**
* 复用原始 JDBC 表类型。
*
* @return JDBC 表类型
*/
@Override
public Schema.TableType getJdbcTableType() {
return delegate.getJdbcTableType();
}
/**
* 判断列是否为预聚合列。
*
* @param column 列名
* @return 原始表判断结果
*/
@Override
public boolean isRolledUp(String column) {
return delegate.isRolledUp(column);
}
/**
* 判断预聚合列能否用于聚合表达式。
*
* @param column 列名
* @param call SQL 调用
* @param parent 父节点
* @param config Calcite 连接配置
* @return 原始表判断结果
*/
@Override
public boolean rolledUpColumnValidInsideAgg(
String column,
SqlCall call,
SqlNode parent,
CalciteConnectionConfig config
) {
return delegate.rolledUpColumnValidInsideAgg(column, call, parent, config);
}
/**
* 交由原始 JDBC Table 生成关系节点,保留 JDBC Convention 与 SQL 下推。
*
* @param context 关系转换上下文
* @param relOptTable 规划器表
* @return 关系节点
* @throws IllegalStateException 原始表不支持关系转换
*/
@Override
public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {
if (!(delegate instanceof TranslatableTable translatableTable)) {
throw new IllegalStateException("MySQL JDBC table does not support relational translation");
}
return translatableTable.toRel(context, relOptTable);
}
/**
* 解包包装器或原始 JDBC Table 能力。
*
* @param type 目标类型
* @param <C> 目标类型参数
* @return 匹配实例;不存在时返回 null
*/
@Override
public <C> C unwrap(Class<C> type) {
if (type.isInstance(this)) {
return type.cast(this);
}
if (type.isInstance(delegate)) {
return type.cast(delegate);
}
return delegate instanceof Wrapper wrapper ? wrapper.unwrap(type) : null;
}
}
/**
* 始终以大小写不敏感方式解析 MySQL 列名的记录类型。
*/
private static final class CaseInsensitiveRelRecordType extends RelRecordType {
private CaseInsensitiveRelRecordType(RelDataType delegate) {
super(delegate.getStructKind(), delegate.getFieldList(), delegate.isNullable());
}
/**
* 按 MySQL 规则查找字段。
*
* @param fieldName 字段名
* @param caseSensitive Calcite 请求的匹配规则MySQL 列名语义下忽略
* @param elideRecord 是否递归省略嵌套记录层级
* @return 匹配字段;不存在时返回 null
*/
@Override
public RelDataTypeField getField(
String fieldName,
boolean caseSensitive,
boolean elideRecord
) {
return super.getField(fieldName, false, elideRecord);
}
}
}

View File

@@ -0,0 +1 @@
com.easyagents.federation.sql.adapter.jdbc.JdbcFederationSqlAdapterProvider

View File

@@ -0,0 +1,140 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.adapter.AdapterDialectContext;
import com.easyagents.federation.sql.adapter.AdapterHints;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.SourceId;
import java.lang.reflect.Proxy;
import java.sql.DatabaseMetaData;
import java.util.List;
import java.util.Map;
import org.apache.calcite.sql.SqlDialect;
import org.apache.calcite.sql.dialect.MysqlSqlDialect;
import org.apache.calcite.sql.dialect.OracleSqlDialect;
import org.apache.calcite.sql.dialect.PostgresqlSqlDialect;
import org.junit.Assert;
import org.junit.Test;
/**
* 默认 JDBC Adapter 的数据库识别与 Calcite 方言选择契约测试。
*/
public class JdbcDialectSelectionTest {
/**
* 验证 MySQL、PostgreSQL 和 Oracle 使用对应 Calcite 官方方言。
*
* @throws Exception 元数据读取失败
*/
@Test
public void shouldSelectBuiltInCalciteDialects() throws Exception {
JdbcFederationSqlAdapterProvider adapter = new JdbcFederationSqlAdapterProvider();
SqlDialect mysql = assertDialect(adapter, "MySQL", "`", MysqlSqlDialect.class);
SqlDialect postgresql = assertDialect(adapter, "PostgreSQL", "\"", PostgresqlSqlDialect.class);
assertDialect(adapter, "Oracle", "\"", OracleSqlDialect.class);
Assert.assertEquals(mysql.isCaseSensitive(), adapter.parserConfig(mysql).caseSensitive());
Assert.assertTrue(adapter.parserConfig(postgresql).caseSensitive());
}
/**
* 验证未知数据库默认拒绝,显式 ANSI 模式才以未验证状态放行。
*
* @throws Exception 元数据读取失败
*/
@Test
public void shouldRequireExplicitAnsiModeForUnknownDatabase() throws Exception {
JdbcFederationSqlAdapterProvider adapter = new JdbcFederationSqlAdapterProvider();
DatabaseMetaData metadata = metadata("UnknownDB", "\"");
Assert.assertFalse(adapter.supports(metadata, new AdapterHints(Map.of())));
AdapterHints experimental = new AdapterHints(Map.of(
JdbcFederationSqlAdapterProvider.EXPERIMENTAL_ANSI_OPTION,
"true"
));
Assert.assertTrue(adapter.supports(metadata, experimental));
Assert.assertEquals(
AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED,
adapter.compatibility(metadata, experimental).status()
);
}
private static SqlDialect assertDialect(
JdbcFederationSqlAdapterProvider adapter,
String product,
String quote,
Class<? extends SqlDialect> expectedType
) throws Exception {
DatabaseMetaData metadata = metadata(product, quote);
Assert.assertTrue(adapter.supports(metadata, new AdapterHints(Map.of())));
SqlDialect dialect = adapter.createDialect(new AdapterDialectContext(metadata, definition(Map.of())));
Assert.assertTrue(expectedType.isInstance(dialect));
Assert.assertEquals(
AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED,
adapter.compatibility(metadata, new AdapterHints(Map.of())).status()
);
return dialect;
}
private static FederationSourceDefinition definition(Map<String, String> options) {
return new FederationSourceDefinition(
new SourceId("source"),
1,
JdbcFederationSqlAdapterProvider.ADAPTER_ID,
List.of(new JdbcSchemaDefinition("app", null, null)),
options
);
}
private static DatabaseMetaData metadata(String product, String quote) {
return (DatabaseMetaData) Proxy.newProxyInstance(
JdbcDialectSelectionTest.class.getClassLoader(),
new Class<?>[] {DatabaseMetaData.class},
(proxy, method, arguments) -> switch (method.getName()) {
case "getDatabaseProductName" -> product;
case "getDatabaseProductVersion" -> "test-version";
case "getDatabaseMajorVersion" -> 1;
case "getDatabaseMinorVersion" -> 0;
case "getDriverName" -> "test-driver";
case "getDriverVersion" -> "1";
case "getIdentifierQuoteString" -> quote;
case "nullsAreSortedHigh" -> true;
case "nullsAreSortedAtEnd", "nullsAreSortedAtStart", "nullsAreSortedLow" -> false;
case "storesUpperCaseIdentifiers", "storesUpperCaseQuotedIdentifiers" -> false;
case "storesLowerCaseIdentifiers", "storesLowerCaseQuotedIdentifiers" -> false;
case "storesMixedCaseIdentifiers", "storesMixedCaseQuotedIdentifiers" -> true;
case "supportsMixedCaseIdentifiers", "supportsMixedCaseQuotedIdentifiers" -> true;
default -> defaultValue(method.getReturnType());
}
);
}
private static Object defaultValue(Class<?> type) {
if (!type.isPrimitive()) {
return null;
}
if (type == boolean.class) {
return false;
}
if (type == int.class) {
return 0;
}
if (type == long.class) {
return 0L;
}
if (type == short.class) {
return (short) 0;
}
if (type == byte.class) {
return (byte) 0;
}
if (type == float.class) {
return 0F;
}
if (type == double.class) {
return 0D;
}
if (type == char.class) {
return '\0';
}
return null;
}
}

View File

@@ -0,0 +1,774 @@
package com.easyagents.federation.sql.adapter.jdbc;
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.api.SqlQueryCommand;
import com.easyagents.federation.sql.compile.FederationSqlPlan;
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.FederationQueryMetricsSnapshot;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.SqlParameter;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationLogicalTableDefinition;
import com.easyagents.federation.sql.federation.FederationQueryMode;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
import com.easyagents.federation.sql.federation.FederationStatisticsSnapshot;
import com.easyagents.federation.sql.federation.FederationTableStatistics;
import com.easyagents.federation.sql.federation.FederationTableStatisticsProvider;
import com.easyagents.federation.sql.source.FederationDataSourceHandles;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.RuntimeFingerprint;
import com.easyagents.federation.sql.source.SourceApplyOptions;
import com.easyagents.federation.sql.source.SourceId;
import java.math.BigDecimal;
import java.time.Instant;
import java.sql.Connection;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
/**
* 多个独立 JDBC DataSource 的基础联邦查询集成测试。
*/
public class JdbcFederatedQueryEngineTest {
private static final SourceId SALES_SOURCE = new SourceId("sales-source");
private static final SourceId BILLING_SOURCE = new SourceId("billing-source");
private static final SourceId REGION_SOURCE = new SourceId("region-source");
private JdbcDataSource sales;
private JdbcDataSource billing;
private JdbcDataSource region;
private FederationSqlEngine engine;
private FederationQueryScopeDefinition scope;
private final AtomicReference<String> statisticsVersion =
new AtomicReference<>("stats-v1");
private final AtomicLong salesRowCount = new AtomicLong(3);
/**
* 创建三个独立 H2 数据库并登记物理数据源。
*
* @throws Exception 数据库初始化失败
*/
@Before
public void setUp() throws Exception {
Instant statisticsCollectedAt = Instant.now();
sales = dataSource("sales");
billing = dataSource("billing");
region = dataSource("region");
execute(sales,
"CREATE TABLE CUSTOMER (ID INT PRIMARY KEY, NAME VARCHAR(64) NOT NULL)",
"INSERT INTO CUSTOMER VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Carol')",
"CREATE TABLE TIME_EVENT (ID INT PRIMARY KEY, EVENT_TIME TIME(6) WITH TIME ZONE, "
+ "EVENT_AT TIMESTAMP(6) WITH TIME ZONE)",
"INSERT INTO TIME_EVENT VALUES (1, TIME WITH TIME ZONE '12:00:00.123456+08:00', "
+ "TIMESTAMP WITH TIME ZONE '2026-08-21 12:00:00.123456+08:00')",
"CREATE TABLE PRECISE_EVENT (ID INT PRIMARY KEY, EVENT_AT TIMESTAMP(6))",
"INSERT INTO PRECISE_EVENT VALUES "
+ "(1, TIMESTAMP '2026-08-21 12:00:00.123456')");
execute(billing,
"CREATE TABLE ORDER_ITEM (ID INT PRIMARY KEY, CUSTOMER_ID INT NOT NULL, AMOUNT DECIMAL(12,2), CODE VARCHAR(64))",
"INSERT INTO ORDER_ITEM VALUES (10, 1, 30.00, 'Alice'), (11, 1, 20.00, 'Alice'), (12, 2, 80.00, 'Bob')",
"CREATE TABLE TIME_EVENT (ID INT PRIMARY KEY, EVENT_TIME TIME(6) WITH TIME ZONE, "
+ "EVENT_AT TIMESTAMP(6) WITH TIME ZONE)",
"INSERT INTO TIME_EVENT VALUES (2, TIME WITH TIME ZONE '06:00:00.123456+02:00', "
+ "TIMESTAMP WITH TIME ZONE '2026-08-21 06:00:00.123456+02:00')",
"CREATE TABLE PRECISE_EVENT (ID INT PRIMARY KEY, EVENT_AT TIMESTAMP(6))",
"INSERT INTO PRECISE_EVENT VALUES "
+ "(2, TIMESTAMP '2026-08-21 08:30:00.654321')");
execute(region,
"CREATE TABLE CUSTOMER_REGION (CUSTOMER_ID INT PRIMARY KEY, REGION VARCHAR(64))",
"INSERT INTO CUSTOMER_REGION VALUES (1, 'North'), (2, 'South'), (3, 'West')");
RuntimeFingerprint fingerprint = new RuntimeFingerprint(
"H2", "2", "H2 JDBC Driver", "2", "1"
);
engine = FederationSqlEngines.builder()
.dataSourceResolver(definition -> FederationDataSourceHandles.shared(
dataSourceFor(definition.sourceId()),
fingerprint
))
.tableStatisticsProvider(() -> new FederationStatisticsSnapshot(
statisticsVersion.get(),
Map.of(
new FederationStatisticsSnapshot.TableKey(
SALES_SOURCE, "APP", "CUSTOMER"
),
new FederationTableStatistics(
salesRowCount.get(),
48,
statisticsCollectedAt,
"test-catalog"
),
new FederationStatisticsSnapshot.TableKey(
BILLING_SOURCE, "APP", "ORDER_ITEM"
),
new FederationTableStatistics(
3,
64,
statisticsCollectedAt,
"test-catalog"
),
new FederationStatisticsSnapshot.TableKey(
REGION_SOURCE, "APP", "CUSTOMER_REGION"
),
new FederationTableStatistics(
3,
32,
statisticsCollectedAt,
"test-catalog"
)
)
))
.federationExecutionPolicy(threeSourcePolicy())
.maximumPlanCacheEntries(32)
.build();
engine.sources().apply(definition(SALES_SOURCE), SourceApplyOptions.prewarmNow());
engine.sources().apply(definition(BILLING_SOURCE), SourceApplyOptions.prewarmNow());
engine.sources().apply(definition(REGION_SOURCE), SourceApplyOptions.prewarmNow());
scope = new FederationQueryScopeDefinition(
"sales-billing",
1,
Map.of(
"SALES",
FederationSourceBindingDefinition.of(SALES_SOURCE, 1),
"BILLING",
FederationSourceBindingDefinition.of(BILLING_SOURCE, 1)
),
"SALES",
FederationExecutionPolicy.basic()
);
}
/**
* 关闭 Engine。
*/
@After
public void tearDown() {
if (engine != null) {
engine.close();
}
}
/**
* 验证同一 Query Scope 中实际只引用一个源时保持完整单源下推。
*/
@Test
public void shouldRouteSingleReferencedSourceToDirectExecution() {
FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(
"SELECT NAME FROM SALES.APP.CUSTOMER ORDER BY ID",
scope
));
Assert.assertEquals(FederationQueryMode.SINGLE_SOURCE, plan.queryMode());
Assert.assertEquals(1, plan.fragments().size());
Assert.assertEquals(java.util.Set.of(SALES_SOURCE), plan.referencedSources());
}
/**
* 验证短逻辑表名、三段逻辑表名和跨源逻辑表 Join 共用底层映射。
*/
@Test
public void shouldExecuteLogicalTableNamesAcrossSources() {
FederationQueryScopeDefinition logicalScope =
FederationQueryScopeDefinition.virtual(
"logical-sales-billing",
2,
scope.bindings(),
"SALES",
List.of(
FederationLogicalTableDefinition.of(
"customers", "SALES", "APP", "CUSTOMER"
),
FederationLogicalTableDefinition.of(
"order_lines", "BILLING", "APP", "ORDER_ITEM"
)
),
FederationExecutionPolicy.basic()
);
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT customers.NAME FROM customers ORDER BY customers.ID",
logicalScope,
List.of()
))) {
Assert.assertTrue(cursor.next());
Assert.assertEquals("Alice", cursor.getObject(1));
}
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT customers.NAME FROM SALES.APP.customers "
+ "ORDER BY SALES.APP.customers.ID",
logicalScope,
List.of()
))) {
Assert.assertTrue(cursor.next());
Assert.assertEquals("Alice", cursor.getObject(1));
}
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT c.NAME, o.AMOUNT FROM customers c "
+ "JOIN order_lines o ON c.ID = o.CUSTOMER_ID "
+ "ORDER BY o.ID",
logicalScope,
List.of()
))) {
Assert.assertTrue(cursor.next());
Assert.assertEquals(List.of("Alice", new BigDecimal("30.00")), cursor.row());
}
}
/**
* 验证全限定 SQL 不依赖未引用默认 Binding 的运行状态。
*/
@Test
public void shouldNotAcquireUnavailableDefaultBindingWhenOnlyAnotherSourceIsReferenced() {
FederationQueryScopeDefinition unavailableDefault =
FederationQueryScopeDefinition.virtual(
"unavailable-default",
1,
Map.of(
"OFFLINE", FederationSourceBindingDefinition.of(
new SourceId("offline-source"), 1
),
"BILLING", FederationSourceBindingDefinition.of(BILLING_SOURCE, 1)
),
"OFFLINE",
FederationExecutionPolicy.basic()
);
FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(
"SELECT AMOUNT FROM BILLING.APP.ORDER_ITEM",
unavailableDefault
));
Assert.assertEquals(FederationQueryMode.SINGLE_SOURCE, plan.queryMode());
Assert.assertEquals(java.util.Set.of(BILLING_SOURCE), plan.referencedSources());
FederationSqlPlan ctePlan = engine.compile(SqlCompileRequest.of(
"WITH x AS (SELECT AMOUNT FROM BILLING.APP.ORDER_ITEM) SELECT * FROM x",
unavailableDefault
));
Assert.assertEquals(FederationQueryMode.SINGLE_SOURCE, ctePlan.queryMode());
Assert.assertEquals(java.util.Set.of(BILLING_SOURCE), ctePlan.referencedSources());
}
/**
* 验证跨源等值 Join、聚合、全局排序和查询指标。
*/
@Test
public void shouldJoinAggregateAndSortAcrossTwoSources() {
String sql = "SELECT c.ID, SUM(o.AMOUNT) AS TOTAL "
+ "FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "GROUP BY c.ID ORDER BY TOTAL DESC";
FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(sql, scope));
Assert.assertEquals(FederationQueryMode.FEDERATED, plan.queryMode());
Assert.assertEquals(2, plan.fragments().size());
Assert.assertEquals(
"the smaller estimated input should become the Hash Join build side",
SALES_SOURCE,
plan.fragments().get(1).sourceId()
);
List<List<Object>> rows = new ArrayList<>();
FederationQueryMetricsSnapshot finalMetrics;
try (FederationResultCursor cursor = engine.query(
SqlQueryCommand.of(sql, scope, List.of())
)) {
while (cursor.next()) {
rows.add(cursor.row());
}
finalMetrics = cursor.metrics();
}
Assert.assertEquals(2, rows.size());
Assert.assertEquals(2, rows.get(0).get(0));
Assert.assertEquals(new BigDecimal("80.00"), rows.get(0).get(1));
Assert.assertEquals(1, rows.get(1).get(0));
Assert.assertEquals(new BigDecimal("50.00"), rows.get(1).get(1));
Assert.assertTrue(finalMetrics.complete());
Assert.assertEquals(2, finalMetrics.returnedRows());
Assert.assertTrue(finalMetrics.intermediateRows() >= 6);
Assert.assertEquals(2, finalMetrics.fragments().size());
}
/**
* 验证三个独立 JDBC 数据源可由同一计划完成 Join 并返回稳定结果。
*/
@Test
public void shouldExecuteJoinAcrossThreeSources() {
FederationQueryScopeDefinition threeSourceScope =
new FederationQueryScopeDefinition(
"sales-billing-region",
1,
Map.of(
"SALES", FederationSourceBindingDefinition.of(SALES_SOURCE, 1),
"BILLING", FederationSourceBindingDefinition.of(BILLING_SOURCE, 1),
"REGION", FederationSourceBindingDefinition.of(REGION_SOURCE, 1)
),
"SALES",
threeSourcePolicy()
);
String sql = "SELECT c.NAME, o.AMOUNT, r.REGION "
+ "FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "JOIN REGION.APP.CUSTOMER_REGION r ON c.ID = r.CUSTOMER_ID "
+ "ORDER BY o.ID";
FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(sql, threeSourceScope));
Assert.assertEquals(FederationQueryMode.FEDERATED, plan.queryMode());
Assert.assertEquals(3, plan.referencedSources().size());
Assert.assertEquals(3, plan.fragments().size());
List<List<Object>> rows = new ArrayList<>();
try (FederationResultCursor cursor = engine.query(
SqlQueryCommand.of(sql, threeSourceScope, List.of())
)) {
while (cursor.next()) {
rows.add(cursor.row());
}
}
Assert.assertEquals(3, rows.size());
Assert.assertEquals(
List.of("Alice", new BigDecimal("30.00"), "North"),
rows.get(0)
);
Assert.assertEquals(
List.of("Bob", new BigDecimal("80.00"), "South"),
rows.get(2)
);
}
/**
* 验证两个分片分别绑定原始查询中的动态参数。
*/
@Test
public void shouldMapParametersIntoDifferentFragments() {
String sql = "SELECT c.NAME, o.AMOUNT "
+ "FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "WHERE c.ID > ? AND o.AMOUNT > ? ORDER BY o.AMOUNT";
FederationSqlPlan plan = engine.compile(new SqlCompileRequest(
sql,
scope,
List.of(Types.INTEGER, Types.DECIMAL),
"default"
));
Assert.assertEquals(2, plan.fragments().size());
Assert.assertEquals(List.of(0), plan.fragments().get(0).parameterMapping());
Assert.assertEquals(List.of(1), plan.fragments().get(1).parameterMapping());
Assert.assertTrue(plan.fragments().stream()
.allMatch(fragment -> fragment.executableSql().toUpperCase().contains("WHERE")));
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
sql,
scope,
List.of(
new SqlParameter(Types.INTEGER, 0),
new SqlParameter(Types.DECIMAL, new BigDecimal("25.00"))
)
))) {
Assert.assertTrue(cursor.next());
Assert.assertEquals(List.of("Alice", new BigDecimal("30.00")), cursor.row());
Assert.assertTrue(cursor.next());
Assert.assertEquals(List.of("Bob", new BigDecimal("80.00")), cursor.row());
Assert.assertFalse(cursor.next());
}
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT NAME FROM SALES.APP.CUSTOMER ORDER BY ID "
+ "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
scope,
List.of(
new SqlParameter(Types.INTEGER, 1),
new SqlParameter(Types.INTEGER, 1)
)
))) {
Assert.assertTrue(cursor.next());
Assert.assertEquals("Bob", cursor.getObject(1));
Assert.assertFalse(cursor.next());
}
}
/**
* 验证 LEFT JOIN、UNION ALL、CTE 和全局分页使用同一联邦执行入口。
*/
@Test
public void shouldExecuteBasicFederatedOperators() {
List<List<Object>> leftRows = query(
"SELECT c.NAME, o.AMOUNT FROM SALES.APP.CUSTOMER c "
+ "LEFT JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "ORDER BY c.ID, o.ID"
);
Assert.assertEquals(4, leftRows.size());
Assert.assertEquals("Carol", leftRows.get(3).get(0));
Assert.assertNull(leftRows.get(3).get(1));
List<List<Object>> unionRows = query(
"SELECT ID FROM SALES.APP.CUSTOMER "
+ "UNION ALL SELECT CUSTOMER_ID FROM BILLING.APP.ORDER_ITEM ORDER BY ID"
);
Assert.assertEquals(List.of(1, 1, 1, 2, 2, 3),
unionRows.stream().map(row -> row.get(0)).toList());
List<List<Object>> cteRows = query(
"WITH large_orders AS ("
+ "SELECT CUSTOMER_ID, AMOUNT FROM BILLING.APP.ORDER_ITEM WHERE AMOUNT >= 50"
+ ") SELECT c.NAME, o.AMOUNT FROM SALES.APP.CUSTOMER c "
+ "JOIN large_orders o ON c.ID = o.CUSTOMER_ID "
+ "ORDER BY o.AMOUNT DESC FETCH NEXT 1 ROWS ONLY"
);
Assert.assertEquals(List.of(List.of("Bob", new BigDecimal("80.00"))), cteRows);
List<List<Object>> residualRows = query(
"SELECT c.ID, c.NAME FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "WHERE c.ID < o.ID ORDER BY c.ID, o.ID"
);
Assert.assertEquals(List.of("Alice", "Alice", "Bob"),
residualRows.stream().map(row -> row.get(1)).toList());
}
/**
* 验证本地联邦算子以 UTC Offset 类型返回 JDBC 4.2 时区值。
*/
@Test
public void shouldPreserveTimezoneSemanticsAcrossFederatedUnion() {
List<List<Object>> rows = query(
"SELECT ID, EVENT_TIME, EVENT_AT FROM SALES.APP.TIME_EVENT "
+ "UNION ALL SELECT ID, EVENT_TIME, EVENT_AT FROM BILLING.APP.TIME_EVENT "
+ "ORDER BY ID"
);
Assert.assertEquals(2, rows.size());
Assert.assertEquals(
java.time.OffsetTime.parse("04:00:00.123456Z"),
rows.get(0).get(1)
);
Assert.assertEquals(
java.time.OffsetDateTime.parse("2026-08-21T04:00:00.123456Z"),
rows.get(0).get(2)
);
Assert.assertEquals(
java.time.OffsetTime.parse("04:00:00.123456Z"),
rows.get(1).get(1)
);
Assert.assertEquals(
java.time.OffsetDateTime.parse("2026-08-21T04:00:00.123456Z"),
rows.get(1).get(2)
);
List<List<Object>> joined = query(
"SELECT s.ID, b.ID FROM SALES.APP.TIME_EVENT s "
+ "JOIN BILLING.APP.TIME_EVENT b ON s.EVENT_AT = b.EVENT_AT"
);
Assert.assertEquals(List.of(List.of(1, 2)), joined);
}
/**
* 验证逻辑 Explain 不访问数据库 Optimizer物理 Explain 显式返回每个分片原生计划。
*/
@Test
public void shouldExplainLogicalAndPhysicalFederatedPlans() {
String sql = "SELECT c.NAME, o.AMOUNT FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID";
SqlCompileRequest compileRequest = SqlCompileRequest.of(sql, scope);
SqlExplainResult logical = engine.explain(new SqlExplainRequest(
compileRequest,
SqlExplainLevel.LOGICAL
));
Assert.assertEquals(FederationQueryMode.FEDERATED, logical.queryMode());
Assert.assertEquals(2, logical.fragments().size());
Assert.assertTrue(logical.fragments().stream()
.allMatch(fragment -> fragment.physicalExplain() == null));
Assert.assertTrue(logical.fragments().stream().allMatch(fragment ->
fragment.costEstimate().estimatedRows() >= 0
&& fragment.costEstimate().estimatedRowWidthBytes() > 0
&& fragment.costEstimate().estimatedTransferBytes() >= 0
&& fragment.costEstimate().statisticsSource().contains("test-catalog")
&& "stats-v1".equals(
fragment.costEstimate().statisticsSnapshotVersion()
)
&& !fragment.pushedDownOperators().isEmpty()
));
SqlExplainResult physical = engine.explain(new SqlExplainRequest(compileRequest));
Assert.assertEquals(SqlExplainLevel.PHYSICAL, physical.level());
Assert.assertEquals(2, physical.fragments().size());
Assert.assertTrue(physical.fragments().stream().allMatch(fragment ->
fragment.physicalExplain() != null
&& fragment.physicalExplain().available()
&& !fragment.physicalExplain().nativePlan().isBlank()
));
}
/**
* 验证纯快照版本变化不扰动计划,实际引用表统计变化才触发重编译。
*/
@Test
public void shouldRecompileWhenStatisticsSnapshotChanges() {
SqlCompileRequest request = SqlCompileRequest.of(
"SELECT NAME FROM SALES.APP.CUSTOMER ORDER BY ID",
scope
);
FederationSqlPlan first = engine.compile(request);
FederationSqlPlan cacheHit = engine.compile(request);
Assert.assertSame(first.relRoot(), cacheHit.relRoot());
statisticsVersion.set("stats-v2");
FederationSqlPlan versionOnly = engine.compile(request);
Assert.assertSame(first.relRoot(), versionOnly.relRoot());
salesRowCount.set(4);
FederationSqlPlan refreshed = engine.compile(request);
Assert.assertNotSame(first.relRoot(), refreshed.relRoot());
Assert.assertEquals(
"stats-v2",
refreshed.fragments().get(0).costEstimate().statisticsSnapshotVersion()
);
}
/**
* 验证不支持的跨源算子和中间结果预算超限均返回稳定错误。
*/
@Test
public void shouldRejectUnsupportedOperatorsAndExceededBudget() {
assertCompileError(
"SELECT c.NAME FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID < o.CUSTOMER_ID",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT ID FROM SALES.APP.CUSTOMER "
+ "UNION SELECT CUSTOMER_ID FROM BILLING.APP.ORDER_ITEM",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.ID FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.NAME = o.CODE",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.ID FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "AND c.NAME < o.CODE",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.ID FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "AND c.ID < o.ID",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.ID FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "AND c.NAME LIKE o.CODE",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.NAME FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "ORDER BY c.NAME",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertCompileError(
"SELECT c.NAME, COUNT(*) FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID "
+ "GROUP BY c.NAME",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
assertQueryError(
"SELECT EVENT_AT FROM SALES.APP.PRECISE_EVENT "
+ "UNION ALL SELECT EVENT_AT FROM BILLING.APP.PRECISE_EVENT",
FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED
);
FederationQueryScopeDefinition strictScope = FederationQueryScopeDefinition.virtual(
"strict-budget",
2,
scope.bindings(),
scope.defaultBinding(),
new FederationExecutionPolicy(2, 8, 2, 1, 1024, 60_000)
);
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT c.NAME, o.AMOUNT FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID",
strictScope,
List.of()
))) {
cursor.next();
Assert.fail("intermediate row budget should reject the query");
} catch (FederationSqlException exception) {
Assert.assertEquals(
FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
exception.errorCode()
);
}
FederationQueryScopeDefinition localExpansionScope =
FederationQueryScopeDefinition.virtual(
"local-expansion-budget",
3,
scope.bindings(),
scope.defaultBinding(),
new FederationExecutionPolicy(2, 8, 2, 7, 64_000, 60_000)
);
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT COUNT(*) FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID",
localExpansionScope,
List.of()
))) {
cursor.next();
Assert.fail("local join expansion should consume the intermediate row budget");
} catch (FederationSqlException exception) {
Assert.assertEquals(
FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
exception.errorCode()
);
}
Assert.assertFalse(query(
"SELECT c.NAME FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID"
).isEmpty());
FederationQueryScopeDefinition singleFragmentSlot =
FederationQueryScopeDefinition.virtual(
"single-fragment-slot",
3,
scope.bindings(),
scope.defaultBinding(),
new FederationExecutionPolicy(2, 8, 1, 100, 64_000, 2_000)
);
try (FederationResultCursor cursor = engine.query(SqlQueryCommand.of(
"SELECT c.NAME FROM SALES.APP.CUSTOMER c "
+ "JOIN BILLING.APP.ORDER_ITEM o ON c.ID = o.CUSTOMER_ID",
singleFragmentSlot,
List.of()
))) {
Assert.assertTrue(cursor.next());
}
}
private List<List<Object>> query(String sql) {
List<List<Object>> rows = new ArrayList<>();
try (FederationResultCursor cursor = engine.query(
SqlQueryCommand.of(sql, scope, List.of())
)) {
while (cursor.next()) {
rows.add(cursor.row());
}
}
return rows;
}
private void assertCompileError(String sql, FederationSqlErrorCode expected) {
try {
engine.compile(SqlCompileRequest.of(sql, scope));
Assert.fail("SQL should have been rejected: " + sql);
} catch (FederationSqlException exception) {
Assert.assertEquals(expected, exception.errorCode());
}
}
/**
* 断言联邦查询在执行阶段返回指定稳定错误。
*
* @param sql 待执行 SQL
* @param expected 预期错误码
*/
private void assertQueryError(String sql, FederationSqlErrorCode expected) {
try (FederationResultCursor cursor = engine.query(
SqlQueryCommand.of(sql, scope, List.of())
)) {
cursor.next();
Assert.fail("SQL execution should have been rejected: " + sql);
} catch (FederationSqlException exception) {
Assert.assertEquals(expected, exception.errorCode());
}
}
private static JdbcDataSource dataSource(String name) {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL(
"jdbc:h2:mem:federation_" + name + '_' + System.nanoTime()
+ ";DB_CLOSE_DELAY=-1"
);
return dataSource;
}
/**
* 根据物理源选择测试数据库。
*
* @param sourceId 物理数据源标识
* @return 对应测试数据源
*/
private JdbcDataSource dataSourceFor(SourceId sourceId) {
if (sourceId.equals(SALES_SOURCE)) {
return sales;
}
if (sourceId.equals(BILLING_SOURCE)) {
return billing;
}
if (sourceId.equals(REGION_SOURCE)) {
return region;
}
throw new IllegalArgumentException("unknown test source: " + sourceId.value());
}
/**
* 返回允许三源执行且限制并发分片数的测试策略。
*
* @return 三源执行策略
*/
private static FederationExecutionPolicy threeSourcePolicy() {
return new FederationExecutionPolicy(
3,
8,
2,
100_000,
64L * 1024L * 1024L,
60_000
);
}
private static void execute(JdbcDataSource dataSource, String... statements)
throws Exception {
try (Connection connection = dataSource.getConnection();
Statement statement = connection.createStatement()) {
for (String sql : statements) {
statement.execute(sql);
}
}
}
private static FederationSourceDefinition definition(SourceId sourceId) {
return new FederationSourceDefinition(
sourceId,
1,
JdbcFederationSqlAdapterProvider.ADAPTER_ID,
List.of(new JdbcSchemaDefinition("APP", null, "PUBLIC")),
Map.of()
);
}
}

View File

@@ -0,0 +1,756 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationExecutionGuard;
import com.easyagents.federation.sql.execute.FederationExecutionObserver;
import com.easyagents.federation.sql.execute.FederationFragmentExecutionContext;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.SqlExecutionOptions;
import com.easyagents.federation.sql.execute.StatementLifecycle;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.Reader;
import java.io.StringReader;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
import java.sql.Statement;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;
import javax.sql.DataSource;
import org.junit.Assert;
import org.junit.Test;
/**
* JDBC 分片执行器的连接获取错误边界测试。
*/
public class JdbcFederationFragmentExecutorTest {
/**
* 验证连接池等待跨过查询截止时间时保留统一查询超时错误。
*/
@Test
public void shouldPreferQueryDeadlineOverConnectionPoolTimeout() {
AtomicInteger checks = new AtomicInteger();
FederationExecutionGuard guard = new FederationExecutionGuard() {
@Override
public void ensureAllowed() {
if (checks.incrementAndGet() > 1) {
throw new FederationSqlException(
FederationSqlErrorCode.QUERY_TIMEOUT,
"query deadline reached"
);
}
}
@Override
public long remainingNanos() {
return 1L;
}
};
FederationSqlException failure = expectFailure(context(guard));
Assert.assertEquals(FederationSqlErrorCode.QUERY_TIMEOUT, failure.errorCode());
}
/**
* 验证截止时间仍有效时保留连接获取超时分类。
*/
@Test
public void shouldReportConnectionAcquisitionTimeoutBeforeQueryDeadline() {
FederationSqlException failure = expectFailure(
context(FederationExecutionGuard.none())
);
Assert.assertEquals(
FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT,
failure.errorCode()
);
}
/**
* 验证显式取消先到达时,驱动的 SQLTimeoutException 不会覆盖取消终态。
*/
@Test
public void shouldPreserveCancellationWhenDriverReportsExecutionTimeout() {
TerminalLifecycle lifecycle = new TerminalLifecycle(false, true);
AtomicBoolean statementClosed = new AtomicBoolean();
AtomicBoolean connectionClosed = new AtomicBoolean();
FederationSqlException failure = expectFailure(context(
FederationExecutionGuard.none(),
executionTimeoutDataSource(statementClosed, connectionClosed),
lifecycle
));
Assert.assertEquals(FederationSqlErrorCode.QUERY_CANCELLED, failure.errorCode());
Assert.assertTrue(lifecycle.unregistered.get());
Assert.assertTrue(statementClosed.get());
Assert.assertTrue(connectionClosed.get());
}
/**
* 验证结果读取阶段同样保留已经先到达的显式取消终态。
*/
@Test
public void shouldPreserveCancellationWhenDriverReportsResultTimeout() {
TerminalLifecycle lifecycle = new TerminalLifecycle(false, true);
JdbcFederationResultCursor cursor = failingCursor(
lifecycle,
new SQLTimeoutException("driver reported timeout after cancel")
);
FederationSqlException failure = expectCursorFailure(cursor);
Assert.assertEquals(FederationSqlErrorCode.QUERY_CANCELLED, failure.errorCode());
Assert.assertTrue(lifecycle.unregistered.get());
}
/**
* 验证统一超时先到达时,驱动普通异常仍保持超时终态。
*/
@Test
public void shouldPreserveTimeoutWhenDriverReportsGenericReadFailure() {
TerminalLifecycle lifecycle = new TerminalLifecycle(true, true);
JdbcFederationResultCursor cursor = failingCursor(
lifecycle,
new SQLException("statement was closed by timeout task")
);
FederationSqlException failure = expectCursorFailure(cursor);
Assert.assertEquals(FederationSqlErrorCode.QUERY_TIMEOUT, failure.errorCode());
Assert.assertTrue(lifecycle.unregistered.get());
}
/**
* 验证二进制流取得后发生取消时,后续流读取立即终止并释放 JDBC 资源。
*
* @throws Exception 流读取失败
*/
@Test
public void shouldStopBinaryStreamReadAfterCancellation() throws Exception {
AtomicBoolean cancelled = new AtomicBoolean();
TerminalLifecycle lifecycle = new TerminalLifecycle(false, false);
JdbcFederationResultCursor cursor = streamingCursor(
lifecycle,
cancellationGuard(cancelled)
);
InputStream stream = cursor.getBinaryStream(1);
cancelled.set(true);
FederationSqlException failure = expectStreamFailure(stream);
Assert.assertEquals(FederationSqlErrorCode.QUERY_CANCELLED, failure.errorCode());
Assert.assertTrue(lifecycle.unregistered.get());
}
/**
* 验证字符流取得后发生超时时,后续流读取立即终止并释放 JDBC 资源。
*
* @throws Exception 流读取失败
*/
@Test
public void shouldStopCharacterStreamReadAfterTimeout() throws Exception {
AtomicBoolean timedOut = new AtomicBoolean();
TerminalLifecycle lifecycle = new TerminalLifecycle(false, false);
JdbcFederationResultCursor cursor = streamingCursor(
lifecycle,
timeoutGuard(timedOut)
);
Reader reader = cursor.getCharacterStream(2);
timedOut.set(true);
FederationSqlException failure = expectStreamFailure(reader);
Assert.assertEquals(FederationSqlErrorCode.QUERY_TIMEOUT, failure.errorCode());
Assert.assertTrue(lifecycle.unregistered.get());
}
/**
* 验证阻塞的结果读取可由 Statement.cancel 解阻,并确定性释放全部 JDBC 资源。
*
* @throws Exception 并发测试等待失败
*/
@Test
public void shouldCancelBlockingResultReadAndCloseAllResources() throws Exception {
CountDownLatch readStarted = new CountDownLatch(1);
CountDownLatch cancelSignal = new CountDownLatch(1);
AtomicBoolean resultSetClosed = new AtomicBoolean();
AtomicBoolean statementClosed = new AtomicBoolean();
AtomicBoolean connectionClosed = new AtomicBoolean();
CancellableLifecycle lifecycle = new CancellableLifecycle();
DataSource dataSource = blockingReadDataSource(
readStarted,
cancelSignal,
resultSetClosed,
statementClosed,
connectionClosed
);
JdbcFederationResultCursor cursor = (JdbcFederationResultCursor)
new JdbcFederationFragmentExecutor().execute(context(
FederationExecutionGuard.none(),
dataSource,
lifecycle
));
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
Future<FederationSqlException> read = executor.submit(
() -> expectCursorFailure(cursor)
);
Assert.assertTrue(readStarted.await(2, TimeUnit.SECONDS));
lifecycle.requestCancellation();
FederationSqlException failure = read.get(2, TimeUnit.SECONDS);
Assert.assertEquals(
FederationSqlErrorCode.QUERY_CANCELLED,
failure.errorCode()
);
Assert.assertTrue(lifecycle.unregistered.get());
Assert.assertTrue(resultSetClosed.get());
Assert.assertTrue(statementClosed.get());
Assert.assertTrue(connectionClosed.get());
} finally {
executor.shutdownNow();
executor.awaitTermination(2, TimeUnit.SECONDS);
}
}
private static FederationSqlException expectFailure(
FederationFragmentExecutionContext context
) {
try {
new JdbcFederationFragmentExecutor().execute(context);
Assert.fail("expected connection acquisition to fail");
return null;
} catch (FederationSqlException exception) {
return exception;
}
}
private static FederationFragmentExecutionContext context(
FederationExecutionGuard guard
) {
return context(guard, new FailingDataSource(), new TerminalLifecycle(false, false));
}
private static FederationFragmentExecutionContext context(
FederationExecutionGuard guard,
DataSource dataSource,
StatementLifecycle lifecycle
) {
return new FederationFragmentExecutionContext(
QueryId.create(),
"SELECT 1",
List.of(),
SqlExecutionOptions.defaults(),
dataSource,
new AdapterCompatibility(
AdapterCompatibilityStatus.VERIFIED,
"test",
"1",
"test",
"1",
"test"
),
Map.of(),
lifecycle,
null,
guard
);
}
private static DataSource executionTimeoutDataSource(
AtomicBoolean statementClosed,
AtomicBoolean connectionClosed
) {
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {PreparedStatement.class},
(proxy, method, arguments) -> {
if ("executeQuery".equals(method.getName())) {
throw new SQLTimeoutException("driver reported timeout after cancel");
}
if ("close".equals(method.getName())) {
statementClosed.set(true);
}
return defaultValue(method.getReturnType());
}
);
Connection connection = (Connection) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {Connection.class},
(proxy, method, arguments) -> {
if ("prepareStatement".equals(method.getName())) {
return statement;
}
if ("close".equals(method.getName())) {
connectionClosed.set(true);
}
return defaultValue(method.getReturnType());
}
);
return dataSource(connection);
}
/**
* 创建在 ResultSet.next 中等待 Statement.cancel 的 JDBC 代理。
*
* @param readStarted 结果读取已开始信号
* @param cancelSignal Statement 已取消信号
* @param resultSetClosed ResultSet 关闭标记
* @param statementClosed Statement 关闭标记
* @param connectionClosed Connection 关闭标记
* @return 可执行阻塞读取的 DataSource
*/
private static DataSource blockingReadDataSource(
CountDownLatch readStarted,
CountDownLatch cancelSignal,
AtomicBoolean resultSetClosed,
AtomicBoolean statementClosed,
AtomicBoolean connectionClosed
) {
Object metadata = Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {java.sql.ResultSetMetaData.class},
(proxy, method, arguments) -> defaultValue(method.getReturnType())
);
ResultSet resultSet = (ResultSet) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {ResultSet.class},
(proxy, method, arguments) -> {
if ("getMetaData".equals(method.getName())) {
return metadata;
}
if ("next".equals(method.getName())) {
readStarted.countDown();
try {
if (!cancelSignal.await(2, TimeUnit.SECONDS)) {
throw new SQLException("test cancellation did not arrive");
}
} catch (InterruptedException exception) {
Thread.currentThread().interrupt();
throw new SQLException("blocking read was interrupted", exception);
}
throw new SQLException("driver read cancelled");
}
if ("close".equals(method.getName())) {
resultSetClosed.set(true);
}
return defaultValue(method.getReturnType());
}
);
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {PreparedStatement.class},
(proxy, method, arguments) -> {
if ("executeQuery".equals(method.getName())) {
return resultSet;
}
if ("cancel".equals(method.getName())) {
cancelSignal.countDown();
}
if ("close".equals(method.getName())) {
statementClosed.set(true);
}
return defaultValue(method.getReturnType());
}
);
Connection connection = (Connection) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {Connection.class},
(proxy, method, arguments) -> {
if ("isReadOnly".equals(method.getName())) {
return true;
}
if ("prepareStatement".equals(method.getName())) {
return statement;
}
if ("close".equals(method.getName())) {
connectionClosed.set(true);
}
return defaultValue(method.getReturnType());
}
);
return dataSource(connection);
}
private static JdbcFederationResultCursor failingCursor(
StatementLifecycle lifecycle,
SQLException readFailure
) {
ResultSet resultSet = (ResultSet) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {ResultSet.class},
(proxy, method, arguments) -> {
if ("next".equals(method.getName())) {
throw readFailure;
}
return defaultValue(method.getReturnType());
}
);
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {PreparedStatement.class},
(proxy, method, arguments) -> defaultValue(method.getReturnType())
);
Connection connection = (Connection) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {Connection.class},
(proxy, method, arguments) -> defaultValue(method.getReturnType())
);
return new JdbcFederationResultCursor(
QueryId.create(),
List.of(),
resultSet,
statement,
connection,
lifecycle
);
}
/**
* 创建可返回二进制流和字符流的测试游标。
*
* @param lifecycle Statement 生命周期
* @param guard 查询终态检查器
* @return 测试游标
*/
private static JdbcFederationResultCursor streamingCursor(
StatementLifecycle lifecycle,
FederationExecutionGuard guard
) {
ResultSet resultSet = (ResultSet) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {ResultSet.class},
(proxy, method, arguments) -> {
if ("getBinaryStream".equals(method.getName())) {
return new ByteArrayInputStream(new byte[] {1, 2, 3});
}
if ("getCharacterStream".equals(method.getName())) {
return new StringReader("streamed value");
}
return defaultValue(method.getReturnType());
}
);
PreparedStatement statement = (PreparedStatement) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {PreparedStatement.class},
(proxy, method, arguments) -> defaultValue(method.getReturnType())
);
Connection connection = (Connection) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {Connection.class},
(proxy, method, arguments) -> defaultValue(method.getReturnType())
);
return new JdbcFederationResultCursor(
QueryId.create(),
List.of(),
resultSet,
statement,
connection,
lifecycle,
guard,
FederationExecutionObserver.none()
);
}
/**
* 创建由布尔终态驱动的取消检查器。
*
* @param cancelled 是否已取消
* @return 取消检查器
*/
private static FederationExecutionGuard cancellationGuard(AtomicBoolean cancelled) {
return terminalGuard(
cancelled,
FederationSqlErrorCode.QUERY_CANCELLED,
"query was cancelled"
);
}
/**
* 创建由布尔终态驱动的超时检查器。
*
* @param timedOut 是否已超时
* @return 超时检查器
*/
private static FederationExecutionGuard timeoutGuard(AtomicBoolean timedOut) {
return terminalGuard(
timedOut,
FederationSqlErrorCode.QUERY_TIMEOUT,
"query deadline reached"
);
}
/**
* 创建固定错误语义的查询终态检查器。
*
* @param terminal 是否进入终态
* @param errorCode 终态错误码
* @param message 错误消息
* @return 查询终态检查器
*/
private static FederationExecutionGuard terminalGuard(
AtomicBoolean terminal,
FederationSqlErrorCode errorCode,
String message
) {
return new FederationExecutionGuard() {
@Override
public void ensureAllowed() {
if (terminal.get()) {
throw new FederationSqlException(errorCode, message);
}
}
@Override
public long remainingNanos() {
return Long.MAX_VALUE;
}
};
}
/**
* 读取二进制流并捕获预期的统一异常。
*
* @param stream 测试流
* @return 捕获的统一异常
* @throws Exception 非预期读取错误
*/
private static FederationSqlException expectStreamFailure(InputStream stream)
throws Exception {
try {
stream.read();
Assert.fail("expected binary stream read to fail");
return null;
} catch (FederationSqlException exception) {
return exception;
}
}
/**
* 读取字符流并捕获预期的统一异常。
*
* @param reader 测试 Reader
* @return 捕获的统一异常
* @throws Exception 非预期读取错误
*/
private static FederationSqlException expectStreamFailure(Reader reader)
throws Exception {
try {
reader.read();
Assert.fail("expected character stream read to fail");
return null;
} catch (FederationSqlException exception) {
return exception;
}
}
private static FederationSqlException expectCursorFailure(
JdbcFederationResultCursor cursor
) {
try {
cursor.next();
Assert.fail("expected result read to fail");
return null;
} catch (FederationSqlException exception) {
return exception;
}
}
private static DataSource dataSource(Connection connection) {
return (DataSource) Proxy.newProxyInstance(
JdbcFederationFragmentExecutorTest.class.getClassLoader(),
new Class<?>[] {DataSource.class},
(proxy, method, arguments) -> {
if ("getConnection".equals(method.getName())) {
return connection;
}
if ("getParentLogger".equals(method.getName())) {
return Logger.getGlobal();
}
return defaultValue(method.getReturnType());
}
);
}
private static Object defaultValue(Class<?> type) {
if (!type.isPrimitive()) {
return null;
}
if (type == boolean.class) {
return false;
}
if (type == byte.class) {
return (byte) 0;
}
if (type == short.class) {
return (short) 0;
}
if (type == int.class) {
return 0;
}
if (type == long.class) {
return 0L;
}
if (type == float.class) {
return 0F;
}
if (type == double.class) {
return 0D;
}
if (type == char.class) {
return '\0';
}
return null;
}
/** 记录测试所需的查询终态和注销动作。 */
private static final class TerminalLifecycle implements StatementLifecycle {
private final boolean timedOut;
private final boolean cancelled;
private final AtomicBoolean unregistered = new AtomicBoolean();
/**
* 创建固定终态的生命周期。
*
* @param timedOut 是否已超时
* @param cancelled 是否已取消
*/
private TerminalLifecycle(boolean timedOut, boolean cancelled) {
this.timedOut = timedOut;
this.cancelled = cancelled;
}
/** {@inheritDoc} */
@Override
public void register(Statement statement) {
}
/** {@inheritDoc} */
@Override
public void unregister(Statement statement) {
unregistered.set(true);
}
/** {@inheritDoc} */
@Override
public boolean cancellationRequested() {
return cancelled;
}
/** {@inheritDoc} */
@Override
public boolean timeoutRequested() {
return timedOut;
}
}
/** 可从测试线程触发 Statement.cancel 的生命周期。 */
private static final class CancellableLifecycle implements StatementLifecycle {
private final AtomicReference<Statement> statement = new AtomicReference<>();
private final AtomicBoolean cancelled = new AtomicBoolean();
private final AtomicBoolean unregistered = new AtomicBoolean();
/** {@inheritDoc} */
@Override
public void register(Statement candidate) {
statement.set(candidate);
}
/** {@inheritDoc} */
@Override
public void unregister(Statement candidate) {
statement.compareAndSet(candidate, null);
unregistered.set(true);
}
/** {@inheritDoc} */
@Override
public boolean cancellationRequested() {
return cancelled.get();
}
/**
* 标记查询取消并调用已登记 Statement 的取消入口。
*
* @throws SQLException JDBC 取消失败
*/
private void requestCancellation() throws SQLException {
cancelled.set(true);
Statement active = statement.get();
if (active != null) {
active.cancel();
}
}
}
/** 始终返回瞬时连接池超时的测试 DataSource。 */
private static final class FailingDataSource implements DataSource {
@Override
public Connection getConnection() throws SQLException {
throw new SQLTransientConnectionException("pool timeout");
}
@Override
public Connection getConnection(String username, String password) throws SQLException {
throw new SQLTransientConnectionException("pool timeout");
}
@Override
public PrintWriter getLogWriter() {
return null;
}
@Override
public void setLogWriter(PrintWriter out) {
}
@Override
public void setLoginTimeout(int seconds) {
}
@Override
public int getLoginTimeout() {
return 0;
}
@Override
public Logger getParentLogger() {
return Logger.getGlobal();
}
@Override
public <T> T unwrap(Class<T> iface) throws SQLException {
throw new SQLException("unsupported");
}
@Override
public boolean isWrapperFor(Class<?> iface) {
return false;
}
}
}

View File

@@ -0,0 +1,152 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationExecutionGuard;
import com.easyagents.federation.sql.execute.FederationFragmentExplainContext;
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
import javax.sql.DataSource;
import org.junit.Assert;
import org.junit.Test;
/**
* JDBC 物理 Explain 的连接获取错误边界测试。
*/
public class JdbcFederationFragmentExplainerTest {
/**
* 验证连接池以运行时异常拒绝连接时返回精确连接获取失败错误。
*/
@Test
public void shouldMapRuntimeConnectionFailure() {
FederationSqlException failure = expectFailure(context(
FederationExecutionGuard.none(),
new IllegalStateException("pool is closed")
));
Assert.assertEquals(
FederationSqlErrorCode.CONNECTION_ACQUISITION_FAILED,
failure.errorCode()
);
}
/**
* 验证连接池失败返回时,已经到达的统一截止时间优先于连接错误。
*/
@Test
public void shouldPreferDeadlineOverRuntimeConnectionFailure() {
AtomicInteger checks = new AtomicInteger();
FederationExecutionGuard guard = new FederationExecutionGuard() {
@Override
public void ensureAllowed() {
if (checks.incrementAndGet() > 1) {
throw new FederationSqlException(
FederationSqlErrorCode.QUERY_TIMEOUT,
"query deadline reached"
);
}
}
@Override
public long remainingNanos() {
return 1L;
}
};
FederationSqlException failure = expectFailure(context(
guard,
new IllegalStateException("pool is closed")
));
Assert.assertEquals(FederationSqlErrorCode.QUERY_TIMEOUT, failure.errorCode());
}
private static FederationSqlException expectFailure(
FederationFragmentExplainContext context
) {
try {
new JdbcFederationFragmentExplainer().explain(context);
Assert.fail("physical Explain should fail");
return null;
} catch (FederationSqlException exception) {
return exception;
}
}
private static FederationFragmentExplainContext context(
FederationExecutionGuard guard,
RuntimeException failure
) {
return new FederationFragmentExplainContext(
"SELECT 1",
List.of(),
failingDataSource(failure),
new AdapterCompatibility(
AdapterCompatibilityStatus.VERIFIED,
"MySQL",
"8",
"test-driver",
"1",
"test"
),
Map.of(),
5,
guard
);
}
private static DataSource failingDataSource(RuntimeException failure) {
return new DataSource() {
@Override
public Connection getConnection() {
throw failure;
}
@Override
public Connection getConnection(String username, String password) {
throw failure;
}
@Override
public <T> T unwrap(Class<T> iface) throws SQLException {
throw new SQLException("not a wrapper");
}
@Override
public boolean isWrapperFor(Class<?> iface) {
return false;
}
@Override
public PrintWriter getLogWriter() {
return null;
}
@Override
public void setLogWriter(PrintWriter out) {
}
@Override
public void setLoginTimeout(int seconds) {
}
@Override
public int getLoginTimeout() {
return 0;
}
@Override
public Logger getParentLogger() {
return Logger.getGlobal();
}
};
}
}

View File

@@ -0,0 +1,347 @@
package com.easyagents.federation.sql.adapter.jdbc;
import com.easyagents.federation.sql.adapter.FederationStatisticsCollectionContext;
import com.easyagents.federation.sql.api.FederationSqlEngine;
import com.easyagents.federation.sql.api.FederationSqlEngines;
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.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
import com.easyagents.federation.sql.federation.FederationStatisticsSnapshot;
import com.easyagents.federation.sql.federation.FederationTableStatistics;
import com.easyagents.federation.sql.source.FederationDataSourceHandles;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.RuntimeFingerprint;
import com.easyagents.federation.sql.source.SourceApplyOptions;
import com.easyagents.federation.sql.source.SourceId;
import com.mysql.cj.jdbc.MysqlDataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.postgresql.ds.PGSimpleDataSource;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
/**
* 本机 MySQL 与 PostgreSQL 的 Adapter 内建统计采集验证。
*/
public class JdbcFederationStatisticsIntegrationTest {
private static final Duration STATISTICS_TTL = Duration.ofMinutes(30);
private static final SourceId MYSQL_SOURCE = new SourceId("mysql-statistics");
private static final SourceId POSTGRESQL_SOURCE = new SourceId(
"postgresql-statistics"
);
/**
* 验证 MySQL 目录统计可由 JDBC Adapter 自动采集。
*
* @throws Exception JDBC 连接或目录读取失败
*/
@Test
public void shouldCollectMysqlStatistics() throws Exception {
Assume.assumeTrue(Boolean.getBoolean("federation.integration.enabled"));
String database = System.getProperty(
"federation.mysql.database",
"data-sheet"
);
String url = "jdbc:mysql://"
+ System.getProperty("federation.mysql.host", "127.0.0.1")
+ ':' + Integer.getInteger("federation.mysql.port", 33306)
+ '/' + database
+ "?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai";
FederationSourceDefinition definition = definition(
"mysql-statistics",
"MAIN",
database,
null
);
try (Connection connection = DriverManager.getConnection(
url,
System.getProperty("federation.mysql.username", "root"),
System.getProperty("federation.mysql.password", "root")
)) {
Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> statistics =
collect(definition, connection);
FederationTableStatistics outlet = statistics.get(
new FederationStatisticsSnapshot.TableKey(
definition.sourceId(),
"MAIN",
"outlet"
)
);
Assert.assertNotNull(outlet);
assertUsable(outlet, "database-catalog:mysql");
}
}
/**
* 验证 PostgreSQL 目录统计可由 JDBC Adapter 自动采集。
*
* @throws Exception JDBC 连接或目录读取失败
*/
@Test
public void shouldCollectPostgresqlStatistics() throws Exception {
Assume.assumeTrue(Boolean.getBoolean("federation.integration.enabled"));
String database = System.getProperty(
"federation.pg.database",
"harmony_adapter"
);
String url = "jdbc:postgresql://"
+ System.getProperty("federation.pg.host", "127.0.0.1")
+ ':' + Integer.getInteger("federation.pg.port", 54329)
+ '/' + database;
FederationSourceDefinition definition = definition(
"postgresql-statistics",
"MAIN",
database,
"public"
);
try (Connection connection = DriverManager.getConnection(
url,
System.getProperty("federation.pg.username", "harmony"),
System.getProperty("federation.pg.password", "harmony")
)) {
Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> statistics =
collect(definition, connection);
Assert.assertFalse(statistics.isEmpty());
Assert.assertTrue(statistics.keySet().stream().allMatch(key ->
definition.sourceId().equals(key.sourceId())
&& "main".equals(key.schema())
));
statistics.values().forEach(value ->
assertUsable(value, "database-catalog:postgresql")
);
}
}
/**
* 验证 Engine 默认启用 Adapter 统计,并将结果交给 Explain 成本估算。
*/
@Test
public void shouldExposeAutomaticallyCollectedStatisticsThroughExplain() {
Assume.assumeTrue(Boolean.getBoolean("federation.integration.enabled"));
String mysqlDatabase = System.getProperty(
"federation.mysql.database",
"data-sheet"
);
String postgresqlDatabase = System.getProperty(
"federation.pg.database",
"harmony_adapter"
);
FederationSourceDefinition mysqlDefinition = definition(
MYSQL_SOURCE.value(),
"main",
mysqlDatabase,
null
);
FederationSourceDefinition postgresqlDefinition = definition(
POSTGRESQL_SOURCE.value(),
"main",
postgresqlDatabase,
"public"
);
Map<SourceId, DataSource> dataSources = Map.of(
MYSQL_SOURCE,
mysqlDataSource(mysqlDatabase),
POSTGRESQL_SOURCE,
postgresqlDataSource(postgresqlDatabase)
);
try (FederationSqlEngine engine = FederationSqlEngines.builder()
.dataSourceResolver(definition -> FederationDataSourceHandles.shared(
dataSources.get(definition.sourceId()),
new RuntimeFingerprint("test", "1", "jdbc", "1", "1")
))
.build()) {
engine.sources().apply(mysqlDefinition, SourceApplyOptions.prewarmNow());
engine.sources().apply(
postgresqlDefinition,
SourceApplyOptions.prewarmNow()
);
FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
"automatic-statistics",
1,
Map.of(
"mysql",
FederationSourceBindingDefinition.of(MYSQL_SOURCE, 1),
"pg",
FederationSourceBindingDefinition.of(POSTGRESQL_SOURCE, 1)
),
"mysql",
FederationExecutionPolicy.basic()
);
SqlExplainResult mysqlExplain = explain(
engine,
scope,
"SELECT * FROM mysql.main.outlet"
);
SqlExplainResult postgresqlExplain = explain(
engine,
scope,
"SELECT * FROM pg.main.artifact"
);
assertExplainStatistics(mysqlExplain, "database-catalog:mysql");
assertExplainStatistics(
postgresqlExplain,
"database-catalog:postgresql"
);
}
}
/**
* 使用 Adapter 统计采集器读取当前连接。
*
* @param definition 物理源定义
* @param connection JDBC 连接
* @return 不可变表统计
* @throws Exception 目录读取失败
*/
private Map<FederationStatisticsSnapshot.TableKey, FederationTableStatistics> collect(
FederationSourceDefinition definition,
Connection connection
) throws Exception {
Instant collectedAt = Instant.now();
return new JdbcFederationStatisticsCollector().collect(
new FederationStatisticsCollectionContext(
definition,
connection,
collectedAt,
collectedAt.plus(STATISTICS_TTL),
5
)
);
}
/**
* 创建单 Schema JDBC 数据源定义。
*
* @param sourceId 物理源标识
* @param logicalSchema 逻辑 Schema
* @param catalog 物理 Catalog
* @param physicalSchema 物理 Schema
* @return 数据源定义
*/
private FederationSourceDefinition definition(
String sourceId,
String logicalSchema,
String catalog,
String physicalSchema
) {
return new FederationSourceDefinition(
new SourceId(sourceId),
1,
JdbcFederationSqlAdapterProvider.ADAPTER_ID,
List.of(new JdbcSchemaDefinition(
logicalSchema,
catalog,
physicalSchema
)),
Map.of()
);
}
/**
* 创建本机 MySQL 测试 DataSource。
*
* @param database 数据库名称
* @return MySQL DataSource
*/
private DataSource mysqlDataSource(String database) {
MysqlDataSource dataSource = new MysqlDataSource();
dataSource.setUrl(
"jdbc:mysql://"
+ System.getProperty("federation.mysql.host", "127.0.0.1")
+ ':' + Integer.getInteger("federation.mysql.port", 33306)
+ '/' + database
+ "?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai"
);
dataSource.setUser(System.getProperty("federation.mysql.username", "root"));
dataSource.setPassword(System.getProperty("federation.mysql.password", "root"));
return dataSource;
}
/**
* 创建本机 PostgreSQL 测试 DataSource。
*
* @param database 数据库名称
* @return PostgreSQL DataSource
*/
private DataSource postgresqlDataSource(String database) {
PGSimpleDataSource dataSource = new PGSimpleDataSource();
dataSource.setServerNames(new String[]{
System.getProperty("federation.pg.host", "127.0.0.1")
});
dataSource.setPortNumbers(new int[]{
Integer.getInteger("federation.pg.port", 54329)
});
dataSource.setDatabaseName(database);
dataSource.setUser(System.getProperty("federation.pg.username", "harmony"));
dataSource.setPassword(System.getProperty("federation.pg.password", "harmony"));
return dataSource;
}
/**
* 执行逻辑 Explain。
*
* @param engine 联邦 SQL Engine
* @param scope 查询范围
* @param sql SQL
* @return Explain 结果
*/
private SqlExplainResult explain(
FederationSqlEngine engine,
FederationQueryScopeDefinition scope,
String sql
) {
return engine.explain(new SqlExplainRequest(
SqlCompileRequest.of(sql, scope),
SqlExplainLevel.LOGICAL
));
}
/**
* 断言 Explain 已使用自动采集的数据库统计。
*
* @param explain Explain 结果
* @param source 预期统计来源
*/
private void assertExplainStatistics(SqlExplainResult explain, String source) {
Assert.assertEquals(1, explain.fragments().size());
Assert.assertFalse(explain.fragments().get(0).costEstimate().statisticsMissing());
Assert.assertTrue(
explain.fragments().get(0).costEstimate().statisticsSource().contains(source)
);
Assert.assertTrue(
explain.fragments().get(0).costEstimate().estimatedRowWidthBytes() > 0L
);
}
/**
* 断言采集结果包含优化器可使用的基础统计。
*
* @param statistics 表统计
* @param source 预期统计来源
*/
private void assertUsable(FederationTableStatistics statistics, String source) {
Assert.assertTrue(statistics.estimatedRows() >= 0D);
Assert.assertTrue(statistics.averageRowWidthBytes() > 0L);
Assert.assertEquals(source, statistics.source());
Assert.assertNotNull(statistics.collectedAt());
Assert.assertTrue(statistics.expiresAt().isAfter(statistics.collectedAt()));
}
}

View File

@@ -0,0 +1,185 @@
package com.easyagents.federation.sql.adapter.jdbc;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.adapter.jdbc.JdbcTable;
import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.Wrapper;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.sql.dialect.MysqlSqlDialect;
import org.apache.calcite.sql.type.SqlTypeName;
import org.junit.Assert;
import org.junit.Test;
/**
* MySQL 表名与列名大小写语义包装器测试。
*/
public class MysqlCaseInsensitiveColumnSchemaTest {
/**
* 验证 JDBC 元数据 LIKE 模式不会把下划线表名解析到近似表。
*/
@Test
public void shouldEscapeJdbcMetadataPatternAndRequireExactPhysicalTable() {
DataSource dataSource = metadataDataSource();
JdbcSchema jdbcSchema = new JdbcSchema(
dataSource,
MysqlSqlDialect.DEFAULT,
null,
"app",
null
);
JdbcTable rawTable = ((Wrapper) jdbcSchema.tables().get("order_item"))
.unwrap(JdbcTable.class);
Assert.assertEquals("order0item", rawTable.jdbcTableName);
Schema schema = new MysqlCaseInsensitiveColumnSchema(jdbcSchema);
JdbcTable exactTable = ((Wrapper) schema.getTable("order_item"))
.unwrap(JdbcTable.class);
Assert.assertEquals("order_item", exactTable.jdbcTableName);
}
/**
* 验证大小写不同的表保持独立,同时列名可忽略大小写查找。
*/
@Test
public void shouldKeepExactTableNamesAndMatchColumnsIgnoringCase() {
Schema schema = new MysqlCaseInsensitiveColumnSchema(new AbstractSchema() {
@Override
protected Map<String, Table> getTableMap() {
return Map.of(
"orders", table("id"),
"Orders", table("different_column")
);
}
});
RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
Table lowerCaseTable = schema.getTable("orders");
Table upperCaseTable = schema.getTable("Orders");
Assert.assertNotNull(lowerCaseTable);
Assert.assertNotNull(upperCaseTable);
Assert.assertNull(schema.getTable("ORDERS"));
RelDataType lowerCaseRow = lowerCaseTable.getRowType(typeFactory);
RelDataType upperCaseRow = upperCaseTable.getRowType(typeFactory);
Assert.assertNotNull(lowerCaseRow.getField("ID", true, false));
Assert.assertEquals("id", lowerCaseRow.getField("ID", true, false).getName());
Assert.assertNotNull(upperCaseRow.getField("DIFFERENT_COLUMN", true, false));
Assert.assertNull(upperCaseRow.getField("ID", true, false));
}
private static Table table(String columnName) {
return new AbstractTable() {
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory) {
return typeFactory.builder()
.add(columnName, SqlTypeName.INTEGER)
.build();
}
};
}
private static DataSource metadataDataSource() {
DatabaseMetaData metadata = (DatabaseMetaData) Proxy.newProxyInstance(
MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
new Class<?>[] {DatabaseMetaData.class},
(proxy, method, arguments) -> switch (method.getName()) {
case "getSearchStringEscape" -> "\\";
case "getJDBCMajorVersion" -> 4;
case "getJDBCMinorVersion" -> 2;
case "getDatabaseProductName" -> "MySQL";
case "getTables" -> tableResultSet((String) arguments[2]);
default -> defaultValue(method.getReturnType());
}
);
Connection connection = (Connection) Proxy.newProxyInstance(
MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
new Class<?>[] {Connection.class},
(proxy, method, arguments) -> switch (method.getName()) {
case "getMetaData" -> metadata;
case "getCatalog" -> "app";
case "getSchema" -> null;
case "close" -> null;
case "isClosed" -> false;
default -> defaultValue(method.getReturnType());
}
);
return (DataSource) Proxy.newProxyInstance(
MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
new Class<?>[] {DataSource.class},
(proxy, method, arguments) -> switch (method.getName()) {
case "getConnection" -> connection;
default -> defaultValue(method.getReturnType());
}
);
}
private static ResultSet tableResultSet(String pattern) {
List<String> tableNames = switch (pattern) {
case "order_item" -> List.of("order0item", "order_item");
case "order\\_item" -> List.of("order_item");
case "%" -> List.of("order0item", "order_item");
default -> List.of();
};
int[] cursor = {-1};
return (ResultSet) Proxy.newProxyInstance(
MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
new Class<?>[] {ResultSet.class},
(proxy, method, arguments) -> switch (method.getName()) {
case "next" -> ++cursor[0] < tableNames.size();
case "getString" -> switch ((Integer) arguments[0]) {
case 1 -> "app";
case 2 -> null;
case 3 -> tableNames.get(cursor[0]);
case 4 -> "TABLE";
default -> null;
};
case "close" -> null;
default -> defaultValue(method.getReturnType());
}
);
}
private static Object defaultValue(Class<?> type) {
if (!type.isPrimitive()) {
return null;
}
if (type == boolean.class) {
return false;
}
if (type == int.class) {
return 0;
}
if (type == long.class) {
return 0L;
}
if (type == short.class) {
return (short) 0;
}
if (type == byte.class) {
return (byte) 0;
}
if (type == float.class) {
return 0F;
}
if (type == double.class) {
return 0D;
}
if (type == char.class) {
return '\0';
}
return null;
}
}