package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.api.FederationSqlEngine;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.api.FederationCleanupMetrics;
import com.easyagents.federation.sql.api.SqlExecutionContext;
import com.easyagents.federation.sql.api.SqlCompletionRequest;
import com.easyagents.federation.sql.api.SqlCompletionResult;
import com.easyagents.federation.sql.api.SqlQueryCommand;
import com.easyagents.federation.sql.compile.FederationSqlPlan;
import com.easyagents.federation.sql.compile.FederationFragmentExplain;
import com.easyagents.federation.sql.compile.FederationSqlPolicy;
import com.easyagents.federation.sql.compile.SqlCompileRequest;
import com.easyagents.federation.sql.compile.SqlExplainRequest;
import com.easyagents.federation.sql.compile.SqlExplainResult;
import com.easyagents.federation.sql.compile.SqlExplainLevel;
import com.easyagents.federation.sql.execute.FederationFragmentExplainContext;
import com.easyagents.federation.sql.execute.FederationFragmentExecutionContext;
import com.easyagents.federation.sql.execute.FederationQueryAdmissionController;
import com.easyagents.federation.sql.execute.FederationQueryPermit;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.FederationPhysicalExplain;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.QueryAdmissionRequest;
import com.easyagents.federation.sql.execute.SqlParameter;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationFragmentPlan;
import com.easyagents.federation.sql.federation.FederationJoinOptimization;
import com.easyagents.federation.sql.federation.FederationQueryMode;
import com.easyagents.federation.sql.federation.FederationSourceRuntimeIdentity;
import com.easyagents.federation.sql.federation.FederationStatisticsSnapshot;
import com.easyagents.federation.sql.federation.FederationStatisticsStatus;
import com.easyagents.federation.sql.federation.FederationTableStatisticsProvider;
import com.easyagents.federation.sql.source.FederationSourceManager;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import org.apache.calcite.plan.RelOptUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 默认 FederationSqlEngine，实现 Query Scope 编译、单源/联邦自动路由和资源治理。
 */
public final class DefaultFederationSqlEngine implements FederationSqlEngine {

    private static final Logger LOG = LoggerFactory.getLogger(DefaultFederationSqlEngine.class);

    private final DefaultFederationSourceManager sourceManager;
    private final FederationQueryAdmissionController admissionController;
    private final BoundedPlanCache planCache;
    private final CalciteFederationSqlCompiler compiler;
    private final AdapterFederationStatisticsProvider automaticStatisticsProvider;
    private final NodeMemoryAdmissionController nodeMemoryAdmission;
    private final CalciteSqlCompleter completer = new CalciteSqlCompleter();
    private final QueryCancellationRegistry cancellations = new QueryCancellationRegistry();
    private final ScheduledThreadPoolExecutor deadlineScheduler = deadlineScheduler();
    private final Map<FederationSqlPlan, AuthorizedPlan> authorizedPlans =
        Collections.synchronizedMap(new WeakHashMap<>());
    private final Map<QueryId, ManagedFederationResultCursor> openCursors =
        new ConcurrentHashMap<>();
    private final AtomicBoolean closed = new AtomicBoolean();
    private final ReentrantReadWriteLock lifecycleLock = new ReentrantReadWriteLock();
    private final Lock operationLock = lifecycleLock.readLock();
    private final Lock closeLock = lifecycleLock.writeLock();

    /**
     * 创建使用默认冷编译并发和联邦资源上限的 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存上限
     * @param crossSourceEnabled 是否允许联邦执行
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        boolean crossSourceEnabled
    ) {
        this(
            sourceManager,
            admissionController,
            policies,
            maximumPlanCacheEntries,
            Math.min(
                maximumPlanCacheEntries,
                Math.max(1, Math.min(8, Runtime.getRuntime().availableProcessors()))
            ),
            crossSourceEnabled,
            FederationExecutionPolicy.basic(),
            64L * 1024L * 1024L,
            Duration.ofMinutes(30),
            null,
            512L * 1024L * 1024L
        );
    }

    /**
     * 创建可独立限制计划容量与冷编译并发的 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存上限
     * @param maximumConcurrentCompilations 最大并发冷编译数
     * @param crossSourceEnabled 是否允许联邦执行
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        int maximumConcurrentCompilations,
        boolean crossSourceEnabled
    ) {
        this(
            sourceManager,
            admissionController,
            policies,
            maximumPlanCacheEntries,
            maximumConcurrentCompilations,
            crossSourceEnabled,
            FederationExecutionPolicy.basic(),
            64L * 1024L * 1024L,
            Duration.ofMinutes(30),
            null,
            512L * 1024L * 1024L
        );
    }

    /**
     * 创建完整配置的 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存上限
     * @param maximumConcurrentCompilations 最大并发冷编译数
     * @param crossSourceEnabled 是否允许联邦执行
     * @param executionPolicy Engine 联邦资源硬上限
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        int maximumConcurrentCompilations,
        boolean crossSourceEnabled,
        FederationExecutionPolicy executionPolicy
    ) {
        this(
            sourceManager,
            admissionController,
            policies,
            maximumPlanCacheEntries,
            maximumConcurrentCompilations,
            crossSourceEnabled,
            executionPolicy,
            64L * 1024L * 1024L,
            Duration.ofMinutes(30),
            null,
            512L * 1024L * 1024L
        );
    }

    /**
     * 创建包含计划缓存权重与 TTL 的完整配置 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存条目上限
     * @param maximumConcurrentCompilations 最大并发冷编译数
     * @param crossSourceEnabled 是否允许联邦执行
     * @param executionPolicy Engine 联邦资源硬上限
     * @param maximumPlanCacheWeightBytes 计划缓存估算权重上限
     * @param planCacheTimeToLive 计划缓存条目存活时间
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        int maximumConcurrentCompilations,
        boolean crossSourceEnabled,
        FederationExecutionPolicy executionPolicy,
        long maximumPlanCacheWeightBytes,
        Duration planCacheTimeToLive
    ) {
        this(
            sourceManager,
            admissionController,
            policies,
            maximumPlanCacheEntries,
            maximumConcurrentCompilations,
            crossSourceEnabled,
            executionPolicy,
            maximumPlanCacheWeightBytes,
            planCacheTimeToLive,
            null,
            512L * 1024L * 1024L
        );
    }

    /**
     * 创建包含计划缓存与外部统计 SPI 的完整配置 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存条目上限
     * @param maximumConcurrentCompilations 最大并发冷编译数
     * @param crossSourceEnabled 是否允许联邦执行
     * @param executionPolicy Engine 联邦资源硬上限
     * @param maximumPlanCacheWeightBytes 计划缓存估算权重上限
     * @param planCacheTimeToLive 计划缓存条目存活时间
     * @param statisticsProvider 调用方托管的联邦表统计 Provider；null 时启用 Adapter 自动采集
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        int maximumConcurrentCompilations,
        boolean crossSourceEnabled,
        FederationExecutionPolicy executionPolicy,
        long maximumPlanCacheWeightBytes,
        Duration planCacheTimeToLive,
        FederationTableStatisticsProvider statisticsProvider
    ) {
        this(
            sourceManager,
            admissionController,
            policies,
            maximumPlanCacheEntries,
            maximumConcurrentCompilations,
            crossSourceEnabled,
            executionPolicy,
            maximumPlanCacheWeightBytes,
            planCacheTimeToLive,
            statisticsProvider,
            512L * 1024L * 1024L
        );
    }

    /**
     * 创建包含节点中间结果内存准入的完整配置 Engine。
     *
     * @param sourceManager 数据源管理器
     * @param admissionController 查询准入控制器
     * @param policies SQL 策略
     * @param maximumPlanCacheEntries 计划缓存条目上限
     * @param maximumConcurrentCompilations 最大并发冷编译数
     * @param crossSourceEnabled 是否允许联邦执行
     * @param executionPolicy Engine 联邦资源硬上限
     * @param maximumPlanCacheWeightBytes 计划缓存估算权重上限
     * @param planCacheTimeToLive 计划缓存条目存活时间
     * @param statisticsProvider 调用方托管的联邦表统计 Provider；null 时启用 Adapter 自动采集
     * @param maximumNodeIntermediateBytes 节点本地中间结果总预留上限
     */
    public DefaultFederationSqlEngine(
        DefaultFederationSourceManager sourceManager,
        FederationQueryAdmissionController admissionController,
        List<FederationSqlPolicy> policies,
        int maximumPlanCacheEntries,
        int maximumConcurrentCompilations,
        boolean crossSourceEnabled,
        FederationExecutionPolicy executionPolicy,
        long maximumPlanCacheWeightBytes,
        Duration planCacheTimeToLive,
        FederationTableStatisticsProvider statisticsProvider,
        long maximumNodeIntermediateBytes
    ) {
        this.sourceManager = sourceManager;
        this.admissionController = admissionController;
        this.nodeMemoryAdmission = new NodeMemoryAdmissionController(
            maximumNodeIntermediateBytes
        );
        this.planCache = new BoundedPlanCache(
            maximumPlanCacheEntries,
            maximumConcurrentCompilations,
            maximumPlanCacheWeightBytes,
            planCacheTimeToLive
        );
        this.sourceManager.addRuntimeClosedListener(planCache::invalidateRuntime);
        this.automaticStatisticsProvider = statisticsProvider == null
            ? new AdapterFederationStatisticsProvider()
            : null;
        if (automaticStatisticsProvider != null) {
            this.sourceManager.addRuntimeClosedListener(automaticStatisticsProvider::invalidate);
        }
        FederationTableStatisticsProvider effectiveStatisticsProvider =
            automaticStatisticsProvider == null
                ? statisticsProvider
                : automaticStatisticsProvider;
        this.compiler = new CalciteFederationSqlCompiler(
            policies,
            crossSourceEnabled,
            executionPolicy,
            effectiveStatisticsProvider
        );
    }

    /** {@inheritDoc} */
    @Override
    public FederationSourceManager sources() {
        operationLock.lock();
        try {
            ensureOpen();
            return sourceManager;
        } finally {
            operationLock.unlock();
        }
    }

    /** {@inheritDoc} */
    @Override
    public SqlCompletionResult complete(SqlCompletionRequest request) {
        if (request == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "completion request must be provided"
            );
        }
        ensureOpen();
        Set<String> bindings = new LinkedHashSet<>(request.queryScope().bindings().keySet());
        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.acquire(
            sourceManager,
            request.queryScope(),
            bindings
        )) {
            return completer.complete(request, snapshot);
        }
    }

    /** {@inheritDoc} */
    @Override
    public FederationSqlPlan compile(SqlCompileRequest request) {
        requireCompileRequest(request);
        QueryDeadline deadline = QueryDeadline.compile(Duration.ofMillis(
            compiler.effectivePolicy(request.queryScope()).maximumExecutionTimeMillis()
        ));
        return compileInternal(request, deadline, false).plan();
    }

    /**
     * 编译并登记一个授权计划。
     *
     * @param request 编译请求
     * @param deadline 编译或查询共享截止时间
     * @param waitForStatistics 是否在截止时间内等待自动统计刷新
     * @return 编译结果及授权信息
     */
    private CompilationResult compileInternal(
        SqlCompileRequest request,
        QueryDeadline deadline,
        boolean waitForStatistics
    ) {
        requireCompileRequest(request);
        long started = System.nanoTime();
        ensureOpen();
        deadline.ensureAllowed();
        Set<String> candidateBindings = compiler.discoverBindings(request);
        deadline.ensureAllowed();
        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.acquire(
            sourceManager,
            request.queryScope(),
            candidateBindings
        )) {
            if (automaticStatisticsProvider != null) {
                CompletableFuture<Void> refresh =
                    automaticStatisticsProvider.refreshIfNeeded(snapshot);
                if (waitForStatistics) {
                    awaitStatistics(refresh, deadline);
                }
            }
            FederationStatisticsSnapshot statisticsSnapshot = compiler.statisticsSnapshot();
            PlanCacheKey key = new PlanCacheKey(
                request.sql(),
                request.queryScope().checksum(),
                request.parameterJdbcTypes(),
                runtimeIdentityFingerprint(snapshot.identities()),
                compiler.policyFingerprint(),
                "table-scoped-v2",
                request.policyVersion()
            );
            BoundedPlanCache.LookupResult lookup = planCache.getOrCompileWithStatus(
                key,
                () -> compiler.compile(request, snapshot, statisticsSnapshot),
                deadline,
                cached -> statisticsCompatible(cached, statisticsSnapshot),
                statisticsSnapshot.capturedAt()
            );
            if (!(lookup.plan() instanceof DefaultFederationSqlPlan cachedPlan)) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.EXECUTION_FAILED,
                    "plan cache returned an unsupported plan implementation"
                );
            }
            DefaultFederationSqlPlan plan = cachedPlan.issuedCopy();
            deadline.ensureAllowed();
            compiler.validatePolicies(request, plan);
            AuthorizedPlan authorized = new AuthorizedPlan(
                request,
                lookup.cacheHit(),
                System.nanoTime() - started
            );
            operationLock.lock();
            try {
                ensureOpen();
                authorizedPlans.put(plan, authorized);
                return new CompilationResult(plan, authorized);
            } finally {
                operationLock.unlock();
            }
        }
    }

    /**
     * 在请求截止时间内等待显式 Explain 所需的自动统计。
     *
     * @param refresh 自动统计刷新任务
     * @param deadline 请求截止时间
     * @throws FederationSqlException 等待超时或线程被中断时抛出
     */
    private void awaitStatistics(
        CompletableFuture<Void> refresh,
        QueryDeadline deadline
    ) {
        deadline.ensureAllowed();
        try {
            refresh.get(Math.max(1L, deadline.remainingNanos()), TimeUnit.NANOSECONDS);
            deadline.ensureAllowed();
        } catch (TimeoutException exception) {
            deadline.ensureAllowed();
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_COMPILE_TIMEOUT,
                "statistics refresh exceeded the SQL compilation deadline",
                exception
            );
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw new FederationSqlException(
                FederationSqlErrorCode.EXPLAIN_FAILED,
                "statistics refresh was interrupted",
                exception
            );
        } catch (ExecutionException exception) {
            // 自动统计是优化提示，异常时继续使用已冻结统计或默认估算。
            LOG.warn("automatic federation statistics refresh failed", exception.getCause());
        }
    }

    /**
     * 判断缓存计划引用的统计是否仍与当前快照一致。
     *
     * @param cached 缓存计划
     * @param current 当前统计快照
     * @return 统计指纹一致时为 true
     */
    private static boolean statisticsCompatible(
        FederationSqlPlan cached,
        FederationStatisticsSnapshot current
    ) {
        if (!(cached instanceof DefaultFederationSqlPlan plan)) {
            return false;
        }
        return current.select(plan.statisticsTables()).fingerprint()
            .equals(plan.statisticsFingerprint());
    }

    /** {@inheritDoc} */
    @Override
    public FederationResultCursor execute(
        FederationSqlPlan plan,
        SqlExecutionContext context
    ) {
        if (plan == null || context == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "plan and execution context must be provided"
            );
        }
        QueryCancellationRegistry.QueryRegistration registration =
            cancellations.begin(context.queryId());
        QueryDeadline deadline = QueryDeadline.query(
            QueryDeadline.deadlineAfter(Duration.ofMillis(effectiveExecutionTimeoutMillis(
                compiler.effectivePolicy(plan.queryScope()),
                context
            ))),
            registration
        );
        try {
            return executeRegistered(
                plan,
                context,
                registration,
                requireAuthorizedPlan(plan),
                deadline
            );
        } catch (RuntimeException exception) {
            closeAndSuppress(registration, exception);
            throw exception;
        }
    }

    private FederationResultCursor executeRegistered(
        FederationSqlPlan publicPlan,
        SqlExecutionContext context,
        QueryCancellationRegistry.QueryRegistration registration,
        AuthorizedPlan authorized,
        QueryDeadline queryDeadline
    ) {
        FederationQueryScopeSnapshot runtimeSnapshot = null;
        FederationQueryPermit permit = null;
        NodeMemoryAdmissionController.Permit memoryPermit = null;
        CompositeQueryResources resources = null;
        AutoCloseable deadline = null;
        FederationExecutionSession session = null;
        FederationResultCursor queryCursor = null;
        try {
            if (!(publicPlan instanceof DefaultFederationSqlPlan plan)) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.INVALID_ARGUMENT,
                    "plan was not issued by the federation SQL engine"
                );
            }
            registration.ensureNotCancelled();
            queryDeadline.ensureAllowed();
            compiler.validatePolicies(authorized.request(), publicPlan);
            registration.ensureNotCancelled();
            if (!plan.executable()) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.CROSS_SOURCE_EXECUTION_UNSUPPORTED,
                    "the compiled plan is explain-only"
                );
            }
            if (plan.parameterCount() != context.parameters().size()) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.PARAMETER_COUNT_MISMATCH,
                    "plan expects " + plan.parameterCount() + " parameters but received "
                        + context.parameters().size()
                );
            }
            requireMatchingParameterTypes(plan.parameterJdbcTypes(), context.parameters());

            Set<String> actualBindings = plan.sourceRuntimeIdentities().stream()
                .map(FederationSourceRuntimeIdentity::bindingName)
                .collect(Collectors.toCollection(LinkedHashSet::new));
            runtimeSnapshot = FederationQueryScopeSnapshot.acquire(
                sourceManager,
                plan.queryScope(),
                actualBindings
            );
            registration.ensureNotCancelled();
            requireMatchingRuntimes(plan, runtimeSnapshot);
            FederationQueryMetricsTracker metrics = new FederationQueryMetricsTracker(
                context.queryId(),
                plan.queryMode(),
                authorized.planCacheHit(),
                authorized.planningNanos(),
                plan.fragments()
            );
            long admissionStarted = System.nanoTime();
            Duration admissionTimeout = boundedAdmissionTimeout(
                context.admissionTimeout(),
                queryDeadline
            );
            try {
                permit = admissionController.acquire(new QueryAdmissionRequest(
                    plan.referencedSources().stream().toList(),
                    context.queryId(),
                    admissionTimeout,
                    registration::cancellationRequested
                ));
            } catch (FederationSqlException exception) {
                if (exception.errorCode() == FederationSqlErrorCode.QUERY_ADMISSION_TIMEOUT) {
                    // 总查询时限先到时，终态应稳定为 QUERY_TIMEOUT。
                    queryDeadline.ensureAllowed();
                }
                throw exception;
            }
            metrics.recordAdmissionWait(System.nanoTime() - admissionStarted);
            registration.ensureNotCancelled();
            queryDeadline.ensureAllowed();

            FederationQueryScopeSnapshot executionSnapshot = runtimeSnapshot;
            FederationExecutionPolicy effectivePolicy = compiler.effectivePolicy(
                plan.queryScope()
            );
            memoryPermit = plan.queryMode() == FederationQueryMode.FEDERATED
                ? nodeMemoryAdmission.acquire(
                    effectivePolicy.maximumIntermediateBytes(),
                    queryDeadline
                )
                : NodeMemoryAdmissionController.none();
            deadline = scheduleExecutionDeadline(
                queryDeadline,
                context.queryId(),
                metrics,
                queryDeadline.remainingNanos()
            );
            resources = new CompositeQueryResources(
                registration,
                runtimeSnapshot,
                permit,
                memoryPermit,
                deadline
            );
            runtimeSnapshot = null;
            permit = null;
            memoryPermit = null;
            deadline = null;
            if (plan.queryMode() == FederationQueryMode.SINGLE_SOURCE) {
                queryCursor = executeSingleSource(
                    plan,
                    context,
                    executionSnapshot,
                    registration,
                    metrics,
                    queryDeadline
                );
            } else {
                // 资源组合已经持有同一快照；会话只借用它并在游标关闭时清理分片。
                session = new FederationExecutionSession(
                    plan,
                    context,
                    executionSnapshot,
                    registration,
                    compiler.effectivePolicy(plan.queryScope()),
                    metrics,
                    queryDeadline
                );
                queryCursor = new FederatedResultCursor(
                    plan,
                    context.queryId(),
                    session,
                    metrics,
                    context.options().maxRows()
                );
                session = null;
            }
            return manageCursor(queryCursor, resources, registration, context.queryId());
        } catch (RuntimeException exception) {
            closeAndSuppress(queryCursor, exception);
            closeAndSuppress(session, exception);
            closeAndSuppress(resources, exception);
            closeAndSuppress(deadline, exception);
            closeAndSuppress(memoryPermit, exception);
            closeAndSuppress(permit, exception);
            closeAndSuppress(runtimeSnapshot, exception);
            closeAndSuppress(registration, exception);
            throw exception;
        }
    }

    private FederationResultCursor executeSingleSource(
        DefaultFederationSqlPlan plan,
        SqlExecutionContext context,
        FederationQueryScopeSnapshot runtimeSnapshot,
        QueryCancellationRegistry.QueryRegistration registration,
        FederationQueryMetricsTracker metrics,
        QueryDeadline queryDeadline
    ) {
        FederationFragmentPlan fragment = plan.fragments().get(0);
        SourceRuntime runtime = runtimeSnapshot.runtime(fragment.bindingName());
        requireMatchingRuntime(plan.sourceRuntimeIdentities().get(0), runtime);
        FederationResultCursor adapterCursor = runtime.adapter().fragmentExecutor().execute(
            new FederationFragmentExecutionContext(
                context.queryId(),
                fragment.executableSql(),
                remapParameters(fragment.parameterMapping(), context.parameters()),
                context.options(),
                runtime.handle().dataSource(),
                runtime.compatibility(),
                runtime.definition().adapterOptions(),
                registration.statementLifecycle(),
                metrics.fragmentObserver(fragment.fragmentId()),
                queryDeadline
            )
        );
        if (adapterCursor == null || !context.queryId().equals(adapterCursor.queryId())) {
            FederationSqlException invalid = new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "adapter returned an invalid query cursor"
            );
            closeAndSuppress(adapterCursor, invalid);
            throw invalid;
        }
        return new MetricsFederationResultCursor(adapterCursor, metrics);
    }

    private FederationResultCursor manageCursor(
        FederationResultCursor queryCursor,
        CompositeQueryResources resources,
        QueryCancellationRegistry.QueryRegistration registration,
        QueryId queryId
    ) {
        operationLock.lock();
        try {
            ensureOpen();
            AtomicReference<ManagedFederationResultCursor> reference = new AtomicReference<>();
            ManagedFederationResultCursor managed = new ManagedFederationResultCursor(
                queryCursor,
                resources,
                () -> openCursors.remove(queryId, reference.get())
            );
            reference.set(managed);
            ManagedFederationResultCursor previous = openCursors.putIfAbsent(queryId, managed);
            if (previous != null) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.EXECUTION_FAILED,
                    "query id is already active: " + queryId.value()
                );
            }
            if (registration.cancellationRequested()) {
                try {
                    registration.ensureNotCancelled();
                } catch (FederationSqlException terminal) {
                    if (registration.timeoutRequested()) {
                        managed.markTimedOut();
                    } else {
                        managed.markCancelled();
                    }
                    closeAndSuppress(managed, terminal);
                    throw terminal;
                }
            }
            return managed;
        } finally {
            operationLock.unlock();
        }
    }

    /** {@inheritDoc} */
    @Override
    public FederationResultCursor query(SqlQueryCommand command) {
        if (command == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "query command must be provided"
            );
        }
        ensureOpen();
        List<Integer> parameterTypes = command.parameters().stream()
            .map(SqlParameter::jdbcType)
            .toList();
        SqlCompileRequest compileRequest = new SqlCompileRequest(
            command.sql(),
            command.queryScope(),
            parameterTypes,
            command.policyVersion()
        );
        SqlExecutionContext executionContext = new SqlExecutionContext(
            command.queryId(),
            command.parameters(),
            command.options(),
            Duration.ofMillis(command.admissionTimeoutMillis())
        );
        long absoluteDeadline = QueryDeadline.deadlineAfter(Duration.ofMillis(
            effectiveExecutionTimeoutMillis(
                compiler.effectivePolicy(command.queryScope()),
                executionContext
            )
        ));
        for (int attempt = 0; attempt < 2; attempt++) {
            QueryCancellationRegistry.QueryRegistration registration =
                cancellations.begin(command.queryId());
            QueryDeadline deadline = QueryDeadline.query(absoluteDeadline, registration);
            try {
                CompilationResult compilation = compileInternal(
                    compileRequest,
                    deadline,
                    false
                );
                registration.ensureNotCancelled();
                return executeRegistered(
                    compilation.plan(),
                    executionContext,
                    registration,
                    compilation.authorized(),
                    deadline
                );
            } catch (RuntimeException exception) {
                closeAndSuppress(registration, exception);
                if (attempt == 0 && isPlanStale(exception)) {
                    continue;
                }
                throw exception;
            }
        }
        throw new IllegalStateException("query retry loop completed unexpectedly");
    }

    private static boolean isPlanStale(RuntimeException exception) {
        return exception instanceof FederationSqlException federationException
            && federationException.errorCode() == FederationSqlErrorCode.PLAN_STALE;
    }

    /** {@inheritDoc} */
    @Override
    public SqlExplainResult explain(SqlExplainRequest request) {
        if (request == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "Explain request must be provided"
            );
        }
        QueryDeadline deadline = QueryDeadline.compile(Duration.ofMillis(
            compiler.effectivePolicy(request.compileRequest().queryScope())
                .maximumExecutionTimeMillis()
        ));
        CompilationResult compilation = compileInternal(
            request.compileRequest(),
            deadline,
            true
        );
        FederationSqlPlan plan = compilation.plan();
        List<FederationFragmentExplain> fragments = request.level() == SqlExplainLevel.PHYSICAL
            ? explainPhysicalFragments(request, plan, deadline)
            : explainLogicalFragments(plan);
        return new SqlExplainResult(
            request.level(),
            plan.queryMode(),
            explainStatisticsStatus(plan.fragments()),
            plan.fragments().stream()
                .allMatch(fragment -> fragment.costEstimate().estimateAvailable()),
            estimatedTransferBytes(plan.fragments()),
            estimatedLocalMemoryBytes(plan.joinOptimizations()),
            plan.joinOptimizations(),
            plan.normalizedSql(),
            plan.executableSql(),
            RelOptUtil.toString(plan.relRoot().rel),
            ((DefaultFederationSqlPlan) plan).executionRelationalPlan(),
            fragments,
            plan.referencedSources(),
            plan.compatibility(),
            plan.executable(),
            compilation.authorized().planCacheHit(),
            plan.queryMode() == FederationQueryMode.SINGLE_SOURCE
                ? "single-source plan is executable"
                : "federated plan is executable with " + plan.fragments().size() + " fragments"
        );
    }

    private static FederationStatisticsStatus explainStatisticsStatus(
        List<FederationFragmentPlan> fragments
    ) {
        boolean anyComplete = false;
        boolean anyMissing = false;
        boolean anyPartial = false;
        for (FederationFragmentPlan fragment : fragments) {
            FederationStatisticsStatus status = fragment.costEstimate().statisticsStatus();
            if (status == FederationStatisticsStatus.STALE) {
                return FederationStatisticsStatus.STALE;
            }
            anyComplete |= status == FederationStatisticsStatus.COMPLETE;
            anyMissing |= status == FederationStatisticsStatus.MISSING;
            anyPartial |= status == FederationStatisticsStatus.PARTIAL;
        }
        if (anyPartial || anyComplete && anyMissing) {
            return FederationStatisticsStatus.PARTIAL;
        }
        return anyComplete ? FederationStatisticsStatus.COMPLETE : FederationStatisticsStatus.MISSING;
    }

    private static double estimatedTransferBytes(List<FederationFragmentPlan> fragments) {
        double total = 0D;
        for (FederationFragmentPlan fragment : fragments) {
            double next = fragment.costEstimate().estimatedTransferBytes();
            total = total > Double.MAX_VALUE - next ? Double.MAX_VALUE : total + next;
        }
        return total;
    }

    private static long estimatedLocalMemoryBytes(
        List<FederationJoinOptimization> optimizations
    ) {
        long peak = 0L;
        for (FederationJoinOptimization optimization : optimizations) {
            long next = (long) Math.min(
                Long.MAX_VALUE,
                Math.ceil(optimization.estimatedBuildBytes())
            );
            peak = Math.max(peak, next);
        }
        return peak;
    }

    private List<FederationFragmentExplain> explainLogicalFragments(FederationSqlPlan plan) {
        return plan.fragments().stream()
            .map(fragment -> fragmentExplain(plan, fragment, null))
            .toList();
    }

    private List<FederationFragmentExplain> explainPhysicalFragments(
        SqlExplainRequest request,
        FederationSqlPlan publicPlan,
        QueryDeadline deadline
    ) {
        if (!(publicPlan instanceof DefaultFederationSqlPlan plan)) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "plan was not issued by the federation SQL engine"
            );
        }
        Set<String> bindings = plan.sourceRuntimeIdentities().stream()
            .map(FederationSourceRuntimeIdentity::bindingName)
            .collect(Collectors.toCollection(LinkedHashSet::new));
        QueryId explainQueryId = QueryId.create();
        List<SqlParameter> originalParameters = explainParameters(request);
        FederationQueryPermit permit;
        Duration admissionTimeout = boundedAdmissionTimeout(
            Duration.ofSeconds(5),
            deadline
        );
        try {
            permit = admissionController.acquire(new QueryAdmissionRequest(
                plan.referencedSources().stream().toList(),
                explainQueryId,
                admissionTimeout,
                () -> deadline.remainingNanos() <= 0L
            ));
        } catch (FederationSqlException exception) {
            if (exception.errorCode() == FederationSqlErrorCode.QUERY_ADMISSION_TIMEOUT
                || exception.errorCode() == FederationSqlErrorCode.QUERY_CANCELLED) {
                deadline.ensureAllowed();
            }
            throw exception;
        }
        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.acquire(
                 sourceManager,
                 plan.queryScope(),
                 bindings
             );
             FederationQueryPermit ignored = permit) {
            deadline.ensureAllowed();
            requireMatchingRuntimes(plan, snapshot);
            List<FederationFragmentExplain> results = new ArrayList<>(plan.fragments().size());
            for (FederationFragmentPlan fragment : plan.fragments()) {
                SourceRuntime runtime = snapshot.runtime(fragment.bindingName());
                FederationPhysicalExplain physical = runtime.adapter().fragmentExplainer()
                    .map(explainer -> explainer.explain(new FederationFragmentExplainContext(
                        fragment.executableSql(),
                        remapParameters(fragment.parameterMapping(), originalParameters),
                        runtime.handle().dataSource(),
                        runtime.compatibility(),
                        runtime.definition().adapterOptions(),
                        15,
                        deadline
                    )))
                    .orElseGet(() -> FederationPhysicalExplain.unavailable(
                        "adapter does not provide physical Explain"
                    ));
                results.add(fragmentExplain(plan, fragment, physical));
            }
            return List.copyOf(results);
        }
    }

    private static FederationFragmentExplain fragmentExplain(
        FederationSqlPlan plan,
        FederationFragmentPlan fragment,
        FederationPhysicalExplain physical
    ) {
        String adapterId = plan.sourceRuntimeIdentities().stream()
            .filter(identity -> identity.bindingName().equals(fragment.bindingName()))
            .map(FederationSourceRuntimeIdentity::adapterId)
            .findFirst()
            .orElse("");
        return new FederationFragmentExplain(
            fragment.fragmentId(),
            fragment.bindingName(),
            fragment.sourceId(),
            adapterId,
            fragment.executableSql(),
            fragment.parameterMapping(),
            fragment.columns(),
            fragment.costEstimate(),
            fragment.pushedDownOperators(),
            physical
        );
    }

    private static List<SqlParameter> explainParameters(SqlExplainRequest request) {
        return request.compileRequest().parameterJdbcTypes().stream()
            .map(jdbcType -> new SqlParameter(jdbcType, null))
            .toList();
    }

    private static void requireMatchingParameterTypes(
        List<Integer> expectedTypes,
        List<SqlParameter> parameters
    ) {
        if (expectedTypes.isEmpty()) {
            return;
        }
        for (int index = 0; index < expectedTypes.size(); index++) {
            if (expectedTypes.get(index) != parameters.get(index).jdbcType()) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.PARAMETER_COUNT_MISMATCH,
                    "parameter " + index + " JDBC type does not match the compiled plan"
                );
            }
        }
    }

    /** {@inheritDoc} */
    @Override
    public boolean cancel(QueryId queryId) {
        ensureOpen();
        QueryCancellationRegistry.CancellationOutcome outcome =
            cancellations.cancelOutcome(queryId);
        boolean found = outcome.found();
        ManagedFederationResultCursor cursor = openCursors.get(queryId);
        if (cursor != null) {
            found = true;
            if (outcome.reason() == QueryCancellationRegistry.TerminationReason.TIMED_OUT) {
                cursor.markTimedOut();
            } else {
                cursor.markCancelled();
            }
            cancellations.submitCleanup(cursor::close);
        }
        return found;
    }

    /** {@inheritDoc} */
    @Override
    public FederationCleanupMetrics cleanupMetrics() {
        return cancellations.cleanupMetrics();
    }

    private void ensureOpen() {
        if (closed.get()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.ENGINE_CLOSED,
                "federation SQL engine is closed"
            );
        }
    }

    /**
     * 有界提交活动游标清理，再关闭取消通道、数据源 Runtime、计划缓存和准入控制器。
     */
    @Override
    public void close() {
        closeLock.lock();
        try {
            closed.compareAndSet(false, true);
            RuntimeException failure = null;
            deadlineScheduler.shutdownNow();
            for (ManagedFederationResultCursor cursor : new ArrayList<>(openCursors.values())) {
                cursor.markCancelled();
                cancellations.submitCleanup(cursor::close);
            }
            try {
                cancellations.close();
            } catch (RuntimeException exception) {
                failure = exception;
            }
            try {
                sourceManager.close();
            } catch (RuntimeException exception) {
                failure = append(failure, exception);
            }
            if (automaticStatisticsProvider != null) {
                try {
                    automaticStatisticsProvider.close();
                } catch (RuntimeException exception) {
                    failure = append(failure, exception);
                }
            }
            planCache.close();
            authorizedPlans.clear();
            try {
                admissionController.close();
            } catch (RuntimeException exception) {
                failure = append(failure, exception);
            }
            if (failure != null) {
                throw failure;
            }
        } finally {
            closeLock.unlock();
        }
    }

    private AuthorizedPlan requireAuthorizedPlan(FederationSqlPlan plan) {
        if (!(plan instanceof DefaultFederationSqlPlan)) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "plan was not issued by the federation SQL engine"
            );
        }
        AuthorizedPlan authorized = authorizedPlans.get(plan);
        if (authorized == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "plan was not compiled by this engine instance"
            );
        }
        return authorized;
    }

    private static void requireMatchingRuntimes(
        DefaultFederationSqlPlan plan,
        FederationQueryScopeSnapshot snapshot
    ) {
        for (FederationSourceRuntimeIdentity identity : plan.sourceRuntimeIdentities()) {
            requireMatchingRuntime(identity, snapshot.runtime(identity.bindingName()));
        }
    }

    private static void requireMatchingRuntime(
        FederationSourceRuntimeIdentity identity,
        SourceRuntime runtime
    ) {
        boolean matches = runtime.definition().sourceId().equals(identity.sourceId())
            && runtime.definition().revision() == identity.sourceRevision()
            && runtime.sourceChecksum().equals(identity.sourceChecksum())
            && runtime.adapter().adapterId().equals(identity.adapterId())
            && runtime.runtimeFingerprint().equals(identity.runtimeFingerprint());
        if (!matches) {
            throw new FederationSqlException(
                FederationSqlErrorCode.PLAN_STALE,
                "plan runtime identity is stale; recompile on the current node"
            );
        }
    }

    private static List<SqlParameter> remapParameters(
        List<Integer> mapping,
        List<SqlParameter> parameters
    ) {
        if (mapping.isEmpty()) {
            return List.of();
        }
        List<SqlParameter> remapped = new ArrayList<>(mapping.size());
        for (Integer sourceIndex : mapping) {
            remapped.add(parameters.get(sourceIndex));
        }
        return List.copyOf(remapped);
    }

    private static String runtimeIdentityFingerprint(
        List<FederationSourceRuntimeIdentity> identities
    ) {
        StringBuilder value = new StringBuilder();
        identities.stream()
            .sorted(java.util.Comparator.comparing(FederationSourceRuntimeIdentity::bindingName))
            .forEach(identity -> {
                appendField(value, identity.bindingName());
                appendField(value, identity.sourceId().value());
                appendField(value, Long.toString(identity.sourceRevision()));
                appendField(value, identity.sourceChecksum());
                appendField(value, identity.adapterId());
                appendField(value, identity.runtimeFingerprint());
            });
        return value.toString();
    }

    private static void appendField(StringBuilder target, String value) {
        target.append(value.length()).append(':').append(value);
    }

    private AutoCloseable scheduleExecutionDeadline(
        QueryDeadline deadline,
        QueryId queryId,
        FederationQueryMetricsTracker metrics,
        long timeoutNanos
    ) {
        ScheduledFuture<?> future = deadlineScheduler.schedule(() -> {
            QueryCancellationRegistry.TerminationReason reason = deadline.requestTimeout();
            if (reason != QueryCancellationRegistry.TerminationReason.TIMED_OUT) {
                return;
            }
            metrics.markTimedOut();
            ManagedFederationResultCursor cursor = openCursors.get(queryId);
            if (cursor != null) {
                cursor.markTimedOut();
                cancellations.submitCleanup(cursor::close);
            }
        }, Math.max(1L, timeoutNanos), TimeUnit.NANOSECONDS);
        return () -> future.cancel(false);
    }

    private static Duration boundedAdmissionTimeout(
        Duration requested,
        QueryDeadline deadline
    ) {
        deadline.ensureAllowed();
        long requestedNanos;
        try {
            requestedNanos = requested.toNanos();
        } catch (ArithmeticException exception) {
            requestedNanos = Long.MAX_VALUE;
        }
        return Duration.ofNanos(Math.max(
            1L,
            Math.min(requestedNanos, deadline.remainingNanos())
        ));
    }

    private static long effectiveExecutionTimeoutMillis(
        FederationExecutionPolicy policy,
        SqlExecutionContext context
    ) {
        long policyMillis = policy.maximumExecutionTimeMillis();
        int requestSeconds = context.options().queryTimeoutSeconds();
        if (requestSeconds == 0) {
            return policyMillis;
        }
        long requestMillis = TimeUnit.SECONDS.toMillis(requestSeconds);
        return Math.max(1L, Math.min(policyMillis, requestMillis));
    }

    private static void requireCompileRequest(SqlCompileRequest request) {
        if (request == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "compile request must be provided"
            );
        }
    }

    private static ScheduledThreadPoolExecutor deadlineScheduler() {
        ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1, runnable -> {
            Thread thread = new Thread(runnable, "easy-agents-federation-deadline");
            thread.setDaemon(true);
            return thread;
        });
        scheduler.setRemoveOnCancelPolicy(true);
        return scheduler;
    }

    private static void closeAndSuppress(AutoCloseable closeable, RuntimeException original) {
        if (closeable == null) {
            return;
        }
        try {
            closeable.close();
        } catch (Exception closeException) {
            original.addSuppressed(closeException);
        }
    }

    private static RuntimeException append(RuntimeException failure, RuntimeException next) {
        if (failure == null) {
            return next;
        }
        failure.addSuppressed(next);
        return failure;
    }

    private record AuthorizedPlan(
        SqlCompileRequest request,
        boolean planCacheHit,
        long planningNanos
    ) {
    }

    private record CompilationResult(
        FederationSqlPlan plan,
        AuthorizedPlan authorized
    ) {
    }
}
