package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.compile.FederationSqlPlan;
import com.easyagents.federation.sql.compile.FederationSqlPolicy;
import com.easyagents.federation.sql.compile.SqlCompileRequest;
import com.easyagents.federation.sql.compile.SqlPolicyContext;
import com.easyagents.federation.sql.execute.FederationColumn;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationCostEstimate;
import com.easyagents.federation.sql.federation.FederationFragmentPlan;
import com.easyagents.federation.sql.federation.FederationJoinAlgorithm;
import com.easyagents.federation.sql.federation.FederationJoinOptimization;
import com.easyagents.federation.sql.federation.FederationJoinSelectionReason;
import com.easyagents.federation.sql.federation.FederationQueryMode;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
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.FederationTableStatistics;
import com.easyagents.federation.sql.federation.FederationTableStatisticsProvider;
import com.easyagents.federation.sql.source.FederationSchemaDefinition;
import com.easyagents.federation.sql.source.SourceId;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.calcite.adapter.enumerable.EnumerableConvention;
import org.apache.calcite.adapter.enumerable.EnumerableInterpretable;
import org.apache.calcite.adapter.enumerable.EnumerableRel;
import org.apache.calcite.adapter.enumerable.EnumerableRules;
import org.apache.calcite.config.Lex;
import org.apache.calcite.jdbc.CalcitePrepare;
import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.plan.hep.HepMatchOrder;
import org.apache.calcite.plan.hep.HepPlanner;
import org.apache.calcite.plan.hep.HepProgramBuilder;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.prepare.RelOptTableImpl;
import org.apache.calcite.rel.RelHomogeneousShuttle;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rel.RelVisitor;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.Sort;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.core.Union;
import org.apache.calcite.rel.core.Values;
import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.rel.rel2sql.RelToSqlConverter;
import org.apache.calcite.rel.rules.CoreRules;
import org.apache.calcite.rel.rules.HyperGraph;
import org.apache.calcite.rel.rules.JoinCommuteRule;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexDynamicParam;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexShuttle;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.runtime.Bindable;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Schemas;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlDataTypeSpec;
import org.apache.calcite.sql.SqlDynamicParam;
import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlJoin;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.SqlOrderBy;
import org.apache.calcite.sql.SqlSelect;
import org.apache.calcite.sql.SqlWith;
import org.apache.calcite.sql.SqlWithItem;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.type.SqlTypeUtil;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.util.SqlBasicVisitor;
import org.apache.calcite.sql.util.SqlOperatorTables;
import org.apache.calcite.sql.util.SqlShuttle;
import org.apache.calcite.sql.util.SqlString;
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.Frameworks;
import org.apache.calcite.tools.Planner;
import org.apache.calcite.tools.Programs;
import org.apache.calcite.tools.RelConversionException;
import org.apache.calcite.tools.ValidationException;

/**
 * 为每次冷编译创建独立 Planner，并直接以 Calcite RelNode 生成单源或联邦计划。
 */
final class CalciteFederationSqlCompiler {

    private static final List<RelOptRule> LOCAL_ENUMERABLE_RULES = List.of(
        EnumerableRules.ENUMERABLE_TABLE_SCAN_RULE,
        EnumerableRules.ENUMERABLE_CALC_RULE,
        EnumerableRules.ENUMERABLE_JOIN_RULE,
        EnumerableRules.ENUMERABLE_AGGREGATE_RULE,
        EnumerableRules.ENUMERABLE_SORT_RULE,
        EnumerableRules.ENUMERABLE_LIMIT_SORT_RULE,
        EnumerableRules.ENUMERABLE_LIMIT_RULE,
        EnumerableRules.ENUMERABLE_UNION_RULE,
        EnumerableRules.ENUMERABLE_VALUES_RULE
    );

    private final List<FederationSqlPolicy> policies;
    private final String policyFingerprint;
    private final boolean crossSourceEnabled;
    private final FederationExecutionPolicy enginePolicy;
    private final FederationTableStatisticsProvider statisticsProvider;

    /**
     * 创建 Calcite 编译器。
     *
     * @param policies SQL 策略
     * @param crossSourceEnabled 是否允许联邦执行
     * @param enginePolicy Engine 资源硬上限
     */
    CalciteFederationSqlCompiler(
        List<FederationSqlPolicy> policies,
        boolean crossSourceEnabled,
        FederationExecutionPolicy enginePolicy,
        FederationTableStatisticsProvider statisticsProvider
    ) {
        this.policies = List.copyOf(policies);
        this.policyFingerprint = policyFingerprint(this.policies);
        this.crossSourceEnabled = crossSourceEnabled;
        this.enginePolicy = enginePolicy == null ? FederationExecutionPolicy.basic() : enginePolicy;
        this.statisticsProvider = statisticsProvider == null
            ? FederationTableStatisticsProvider.none()
            : statisticsProvider;
    }

    /**
     * 创建不注入外部统计的兼容编译器。
     *
     * @param policies SQL 策略
     * @param crossSourceEnabled 是否允许联邦执行
     * @param enginePolicy Engine 资源硬上限
     */
    CalciteFederationSqlCompiler(
        List<FederationSqlPolicy> policies,
        boolean crossSourceEnabled,
        FederationExecutionPolicy enginePolicy
    ) {
        this(
            policies,
            crossSourceEnabled,
            enginePolicy,
            FederationTableStatisticsProvider.none()
        );
    }

    /**
     * 创建使用默认资源上限的兼容编译器。
     *
     * @param policies SQL 策略
     * @param crossSourceEnabled 是否允许联邦执行
     */
    CalciteFederationSqlCompiler(
        List<FederationSqlPolicy> policies,
        boolean crossSourceEnabled
    ) {
        this(policies, crossSourceEnabled, FederationExecutionPolicy.basic());
    }

    /**
     * 使用已构建单源 Runtime 的兼容编译入口。
     *
     * @param request 编译请求
     * @param runtime 单源 Runtime
     * @param knownSources 旧接口已知源集合；单源范围内不再使用
     * @return 编译计划
     */
    FederationSqlPlan compile(
        SqlCompileRequest request,
        SourceRuntime runtime,
        Set<SourceId> knownSources
    ) {
        try (FederationQueryScopeSnapshot snapshot =
                 FederationQueryScopeSnapshot.borrowedSingle(request.queryScope(), runtime)) {
            return compile(request, snapshot);
        }
    }

    /**
     * 返回当前 Engine 策略实现与版本的稳定指纹。
     *
     * @return 策略指纹
     */
    String policyFingerprint() {
        return policyFingerprint;
    }

    /**
     * 返回当前外部统计快照版本，用于隔离成本计划缓存。
     *
     * @return 稳定版本字符串
     */
    FederationStatisticsSnapshot statisticsSnapshot() {
        FederationStatisticsSnapshot snapshot = statisticsProvider.snapshot();
        return snapshot == null
            ? FederationStatisticsSnapshot.empty()
            : snapshot;
    }

    /**
     * 返回 Engine 与调用方 Scope 收敛后的有效资源策略。
     *
     * @param scope 查询范围
     * @return 有效策略
     */
    FederationExecutionPolicy effectivePolicy(FederationQueryScopeDefinition scope) {
        return enginePolicy.intersect(scope.executionPolicy());
    }

    /**
     * 使用默认 Binding 方言预解析 SQL，确定需要加载的候选 Binding。
     *
     * <p>最终来源仍以校验后的 TableScan 为准；该结果只用于减少无关 Runtime 初始化。</p>
     *
     * @param request 编译请求
     * @return 候选 Binding 名称
     */
    Set<String> discoverBindings(SqlCompileRequest request) {
        try {
            SqlNode parsed = org.apache.calcite.sql.parser.SqlParser.create(
                request.sql(),
                org.apache.calcite.sql.parser.SqlParser.config()
                    .withLex(Lex.ORACLE)
            ).parseQuery();
            requireReadOnly(parsed);
            Set<String> bindings = new LinkedHashSet<>();
            discoverBindings(parsed, request.queryScope(), Set.of(), bindings);
            if (bindings.isEmpty()) {
                bindings.add(request.queryScope().defaultBinding());
            }
            return Set.copyOf(bindings);
        } catch (SqlParseException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_PARSE_FAILED,
                "SQL parsing failed",
                exception
            );
        }
    }

    /**
     * 对已编译计划执行当前策略校验；缓存命中和执行前均会调用。
     *
     * @param request 原始编译请求
     * @param plan 已编译计划
     */
    void validatePolicies(SqlCompileRequest request, FederationSqlPlan plan) {
        SqlPolicyContext context = new SqlPolicyContext(
            request,
            plan.sqlNode(),
            plan.relRoot(),
            plan.referencedSources()
        );
        for (FederationSqlPolicy policy : policies) {
            policy.validate(context);
        }
    }

    /**
     * 编译单源完整下推计划或两源基础联邦计划。
     *
     * @param request 编译请求
     * @param snapshot 候选 Runtime 快照
     * @return 编译计划
     */
    FederationSqlPlan compile(
        SqlCompileRequest request,
        FederationQueryScopeSnapshot snapshot
    ) {
        return compile(request, snapshot, statisticsSnapshot());
    }

    /**
     * 使用调用方已捕获的统计快照编译，确保计划诊断、成本数据与缓存键一致。
     *
     * @param request 编译请求
     * @param snapshot 候选 Runtime 快照
     * @param statisticsSnapshot 冻结的统计快照
     * @return 编译计划
     */
    FederationSqlPlan compile(
        SqlCompileRequest request,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        SourceRuntime plannerRuntime = plannerRuntime(request.queryScope(), snapshot);
        UnifiedSchema unifiedSchema = buildUnifiedSchema(request.queryScope(), snapshot);
        List<RelOptRule> plannerRules = uniquePhysicalSources(snapshot) == 1
            ? plannerRuntime.adapter().plannerRules()
            : List.of();
        Frameworks.ConfigBuilder configBuilder = Frameworks.newConfigBuilder()
            .defaultSchema(unifiedSchema.defaultSchema())
            .operatorTable(operatorTable(snapshot))
            .parserConfig(plannerRuntime.adapter().parserConfig(plannerRuntime.dialect()))
            .typeSystem(plannerRuntime.adapter().typeSystem());
        if (!plannerRules.isEmpty()) {
            configBuilder.programs(Programs.ofRules(plannerRules));
        }
        FrameworkConfig config = configBuilder.build();

        try (Planner planner = Frameworks.getPlanner(config)) {
            SqlNode parsed = planner.parse(request.sql());
            requireReadOnly(parsed);
            int dynamicParameterCount = countDynamicParameters(parsed);
            requireMatchingParameterTypes(request, dynamicParameterCount);
            SqlNode resolved = LogicalTableSqlResolver.resolve(parsed, request.queryScope());
            SqlNode typedSql = applyDeclaredParameterTypes(resolved, request, plannerRuntime);
            SqlNode validated = planner.validate(typedSql);
            RelRoot relRoot = planner.rel(validated);
            if (!request.parameterJdbcTypes().isEmpty()) {
                relRoot = relRoot.withRel(stripSyntheticParameterCasts(relRoot.rel));
            }
            if (!plannerRules.isEmpty()) {
                relRoot = relRoot.withRel(planner.transform(
                    0,
                    relRoot.rel.getTraitSet(),
                    relRoot.rel
                ));
            }

            RelNode queryRoot = applyStatistics(
                relRoot.project(),
                request.queryScope(),
                snapshot,
                statisticsSnapshot
            );
            FederationStatisticsMetadata.install(queryRoot.getCluster());
            RelSourceAnalysis initialSourceAnalysis = analyzeSources(
                queryRoot,
                request.queryScope(),
                snapshot
            );
            validateCrossSourceJoinConditions(queryRoot, initialSourceAnalysis);
            queryRoot = optimizeForFederation(queryRoot);
            RelSourceAnalysis sourceAnalysis = analyzeSources(
                queryRoot,
                request.queryScope(),
                snapshot
            );
            if (sourceAnalysis.sources(queryRoot).size() > 1) {
                queryRoot = optimizeJoinBuildSides(
                    queryRoot,
                    sourceAnalysis,
                    request.queryScope(),
                    snapshot,
                    statisticsSnapshot
                );
                // Join 交换会生成新的 Join/Project 节点，来源身份需按新树重新计算。
                sourceAnalysis = analyzeSources(queryRoot, request.queryScope(), snapshot);
            }
            Set<SourceId> referencedSources = sourceAnalysis.sources(queryRoot);
            if (referencedSources.isEmpty()) {
                referencedSources = Set.of(request.sourceId());
            }
            Set<FederationStatisticsSnapshot.TableKey> statisticsTables =
                referencedStatisticsTables(
                    queryRoot,
                    request.queryScope(),
                    snapshot
                );
            FederationStatisticsSnapshot.Selection statisticsSelection =
                statisticsSnapshot.select(statisticsTables);
            FederationExecutionPolicy effectivePolicy = effectivePolicy(request.queryScope());
            if (referencedSources.size() > effectivePolicy.maximumReferencedSources()) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
                    "query references " + referencedSources.size()
                        + " physical sources but the limit is "
                        + effectivePolicy.maximumReferencedSources()
                );
            }
            if (referencedSources.size() == 1) {
                return compileSingleSource(
                    request,
                    snapshot,
                    validated,
                    relRoot,
                    queryRoot,
                    dynamicParameterCount,
                    referencedSources,
                    sourceAnalysis,
                    statisticsSnapshot,
                    statisticsTables,
                    statisticsSelection
                );
            }
            if (!crossSourceEnabled) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.CROSS_SOURCE_DISABLED,
                    "cross-source SQL execution is disabled"
                );
            }
            return compileFederated(
                request,
                snapshot,
                plannerRuntime,
                validated,
                relRoot,
                queryRoot,
                dynamicParameterCount,
                referencedSources,
                sourceAnalysis,
                effectivePolicy,
                statisticsSnapshot,
                statisticsTables,
                statisticsSelection
            );
        } catch (FederationSqlException exception) {
            throw exception;
        } catch (SqlParseException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_PARSE_FAILED,
                "SQL parsing failed",
                exception
            );
        } catch (ValidationException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_VALIDATION_FAILED,
                "SQL validation failed",
                exception
            );
        } catch (RelConversionException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_COMPILE_FAILED,
                "SQL relational conversion failed",
                exception
            );
        } catch (RuntimeException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_COMPILE_FAILED,
                "SQL compilation failed",
                exception
            );
        }
    }

    private FederationSqlPlan compileSingleSource(
        SqlCompileRequest request,
        FederationQueryScopeSnapshot snapshot,
        SqlNode validated,
        RelRoot relRoot,
        RelNode queryRoot,
        int parameterCount,
        Set<SourceId> referencedSources,
        RelSourceAnalysis sourceAnalysis,
        FederationStatisticsSnapshot statisticsSnapshot,
        Set<FederationStatisticsSnapshot.TableKey> statisticsTables,
        FederationStatisticsSnapshot.Selection statisticsSelection
    ) {
        ensureNoEnumerableResidual(queryRoot);
        SourceId sourceId = referencedSources.iterator().next();
        String bindingName = sourceAnalysis.bindingForSource(sourceId);
        SourceRuntime runtime = snapshot.runtime(bindingName);
        SqlString targetSql = toSql(queryRoot, runtime);
        List<Integer> mapping = parameterMapping(targetSql, parameterCount, true);
        FederationFragmentPlan fragment = new FederationFragmentPlan(
            "fragment-1",
            bindingName,
            sourceId,
            targetSql.getSql(),
            mapping,
            columns(queryRoot.getRowType()),
            estimateCost(
                queryRoot,
                sourceId,
                bindingName,
                request.queryScope(),
                statisticsSnapshot
            ),
            pushedDownOperators(queryRoot)
        );
        CompiledFederationFragment compiled = new CompiledFederationFragment(
            fragment,
            queryRoot.getRowType()
        );
        List<FederationSourceRuntimeIdentity> identities = identitiesForBindings(
            snapshot,
            Set.of(bindingName)
        );
        return new DefaultFederationSqlPlan(
            request.queryScope(),
            FederationQueryMode.SINGLE_SOURCE,
            validated.toSqlString(runtime.dialect()).getSql(),
            targetSql.getSql(),
            validated,
            relRoot.withRel(queryRoot),
            parameterCount,
            request.parameterJdbcTypes(),
            mapping,
            List.of(fragment),
            List.of(),
            identities,
            columns(relRoot.validatedRowType),
            referencedSources,
            runtime.compatibility(),
            true,
            Map.of(fragment.fragmentId(), compiled),
            RelOptUtil.toString(queryRoot),
            null,
            null,
            statisticsTables,
            statisticsSelection.fingerprint(),
            statisticsSelection.validUntil()
        );
    }

    private FederationSqlPlan compileFederated(
        SqlCompileRequest request,
        FederationQueryScopeSnapshot snapshot,
        SourceRuntime plannerRuntime,
        SqlNode validated,
        RelRoot relRoot,
        RelNode queryRoot,
        int parameterCount,
        Set<SourceId> referencedSources,
        RelSourceAnalysis sourceAnalysis,
        FederationExecutionPolicy effectivePolicy,
        FederationStatisticsSnapshot statisticsSnapshot,
        Set<FederationStatisticsSnapshot.TableKey> statisticsTables,
        FederationStatisticsSnapshot.Selection statisticsSelection
    ) {
        List<RelNode> fragmentRoots = maximalSingleSourceFragments(queryRoot, sourceAnalysis);
        if (fragmentRoots.size() > effectivePolicy.maximumFragments()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
                "query requires " + fragmentRoots.size() + " fragments but the limit is "
                    + effectivePolicy.maximumFragments()
            );
        }
        IdentityHashMap<RelNode, CompiledFederationFragment> fragmentsByRoot =
            new IdentityHashMap<>();
        LinkedHashMap<String, CompiledFederationFragment> compiledById = new LinkedHashMap<>();
        int sequence = 1;
        for (RelNode fragmentRoot : fragmentRoots) {
            SourceId sourceId = sourceAnalysis.sources(fragmentRoot).iterator().next();
            String bindingName = sourceAnalysis.bindingForSource(sourceId);
            SourceRuntime runtime = snapshot.runtime(bindingName);
            SqlString fragmentSql = toSql(fragmentRoot, runtime);
            List<Integer> mapping = parameterMapping(fragmentSql, parameterCount, false);
            FederationFragmentPlan fragmentPlan = new FederationFragmentPlan(
                "fragment-" + sequence++,
                bindingName,
                sourceId,
                fragmentSql.getSql(),
                mapping,
                columns(fragmentRoot.getRowType()),
                estimateCost(
                    fragmentRoot,
                    sourceId,
                    bindingName,
                    request.queryScope(),
                    statisticsSnapshot
                ),
                pushedDownOperators(fragmentRoot)
            );
            CompiledFederationFragment compiled = new CompiledFederationFragment(
                fragmentPlan,
                fragmentRoot.getRowType()
            );
            fragmentsByRoot.put(fragmentRoot, compiled);
            compiledById.put(fragmentPlan.fragmentId(), compiled);
        }
        LocalLogicalPlan localPlan = replaceFragments(queryRoot, fragmentsByRoot);
        validateLocalOperators(localPlan.root());
        Bindable<?> bindable = compileLocalBindable(localPlan.root());
        List<FederationFragmentPlan> publicFragments = compiledById.values().stream()
            .map(CompiledFederationFragment::plan)
            .toList();
        List<FederationJoinOptimization> joinOptimizations = joinOptimizations(
            queryRoot,
            sourceAnalysis,
            request.queryScope(),
            snapshot,
            statisticsSnapshot
        );
        Set<String> actualBindings = publicFragments.stream()
            .map(FederationFragmentPlan::bindingName)
            .collect(Collectors.toCollection(LinkedHashSet::new));
        return new DefaultFederationSqlPlan(
            request.queryScope(),
            FederationQueryMode.FEDERATED,
            validated.toSqlString(plannerRuntime.dialect()).getSql(),
            request.sql(),
            validated,
            relRoot.withRel(queryRoot),
            parameterCount,
            request.parameterJdbcTypes(),
            List.of(),
            publicFragments,
            joinOptimizations,
            identitiesForBindings(snapshot, actualBindings),
            columns(relRoot.validatedRowType),
            referencedSources,
            plannerRuntime.compatibility(),
            true,
            compiledById,
            RelOptUtil.toString(localPlan.root()),
            bindable,
            localPlan.rootSchema(),
            statisticsTables,
            statisticsSelection.fingerprint(),
            statisticsSelection.validUntil()
        );
    }

    private static UnifiedSchema buildUnifiedSchema(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        SchemaPlus root = CalciteSchema.createRootSchema(true, false).plus();
        SchemaPlus defaultSchema = root;
        for (Map.Entry<String, SourceRuntime> entry : snapshot.runtimesByBinding().entrySet()) {
            String bindingName = entry.getKey();
            SourceRuntime runtime = entry.getValue();
            FederationSourceBindingDefinition binding = scope.bindings().get(bindingName);
            SchemaPlus bindingSchema = root.add(bindingName, new AbstractSchema());
            Map<String, String> mappings = binding.schemaMappings();
            if (mappings.isEmpty()) {
                LinkedHashMap<String, String> sameName = new LinkedHashMap<>();
                for (FederationSchemaDefinition definition : runtime.definition().schemas()) {
                    sameName.put(definition.logicalName(), definition.logicalName());
                }
                mappings = sameName;
            }
            SchemaPlus onlySchema = null;
            for (Map.Entry<String, String> mapping : mappings.entrySet()) {
                SchemaPlus sourceSchema = runtime.rootSchema()
                    .getSubSchema(runtime.definition().sourceId().value());
                sourceSchema = sourceSchema == null
                    ? null
                    : sourceSchema.getSubSchema(mapping.getValue());
                if (sourceSchema == null) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.INVALID_QUERY_SCOPE,
                        "binding " + bindingName + " maps an unknown source schema: "
                            + mapping.getValue()
                    );
                }
                Schema mounted = sourceSchema.unwrap(Schema.class);
                onlySchema = bindingSchema.add(mapping.getKey(), mounted);
            }
            if (bindingName.equals(scope.defaultBinding())) {
                defaultSchema = mappings.size() == 1 ? onlySchema : bindingSchema;
            }
        }
        return new UnifiedSchema(defaultSchema);
    }

    private static SqlOperatorTable operatorTable(FederationQueryScopeSnapshot snapshot) {
        List<SqlOperatorTable> tables = snapshot.runtimesByBinding().values().stream()
            .map(runtime -> runtime.adapter().operatorTable())
            .distinct()
            .toList();
        return tables.isEmpty()
            ? SqlStdOperatorTable.instance()
            : SqlOperatorTables.chain(tables);
    }

    private static SourceRuntime plannerRuntime(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        SourceRuntime defaultRuntime = snapshot.runtimesByBinding().get(scope.defaultBinding());
        return defaultRuntime == null
            ? snapshot.runtimesByBinding().values().iterator().next()
            : defaultRuntime;
    }

    private static long uniquePhysicalSources(FederationQueryScopeSnapshot snapshot) {
        return snapshot.runtimesByBinding().values().stream()
            .map(runtime -> runtime.definition().sourceId())
            .distinct()
            .count();
    }

    private static RelSourceAnalysis analyzeSources(
        RelNode root,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        IdentityHashMap<RelNode, Set<SourceId>> sources = new IdentityHashMap<>();
        Map<SourceId, String> bindingsBySource = new LinkedHashMap<>();
        snapshot.runtimesByBinding().forEach((binding, runtime) ->
            bindingsBySource.putIfAbsent(runtime.definition().sourceId(), binding)
        );
        analyzeNode(root, scope, snapshot, sources, bindingsBySource);
        return new RelSourceAnalysis(sources, bindingsBySource);
    }

    private static Set<SourceId> analyzeNode(
        RelNode node,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        IdentityHashMap<RelNode, Set<SourceId>> cache,
        Map<SourceId, String> bindingsBySource
    ) {
        Set<SourceId> cached = cache.get(node);
        if (cached != null) {
            return cached;
        }
        LinkedHashSet<SourceId> found = new LinkedHashSet<>();
        if (node instanceof TableScan scan) {
            String bindingName = bindingFromQualifiedName(
                scan.getTable().getQualifiedName(),
                scope,
                snapshot
            );
            SourceRuntime runtime = snapshot.runtime(bindingName);
            found.add(runtime.definition().sourceId());
            bindingsBySource.putIfAbsent(runtime.definition().sourceId(), bindingName);
        } else {
            for (RelNode input : node.getInputs()) {
                found.addAll(analyzeNode(input, scope, snapshot, cache, bindingsBySource));
            }
        }
        Set<SourceId> immutable = Set.copyOf(found);
        cache.put(node, immutable);
        return immutable;
    }

    private FederationCostEstimate estimateCost(
        RelNode fragmentRoot,
        SourceId sourceId,
        String bindingName,
        FederationQueryScopeDefinition scope,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        List<TableScan> scans = new ArrayList<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof TableScan scan) {
                    scans.add(scan);
                }
                super.visit(node, ordinal, parent);
            }
        }.go(fragmentRoot);

        var metadata = fragmentRoot.getCluster().getMetadataQuery();
        double calciteRows = safeEstimate(metadata.getRowCount(fragmentRoot), 100D);
        double scale = 1D;
        long providedWidth = 0;
        int providedFields = 0;
        int totalScanFields = 0;
        int missingStatistics = 0;
        int partialStatistics = 0;
        boolean staleStatistics = false;
        Instant oldestCollection = null;
        LinkedHashSet<String> sources = new LinkedHashSet<>();
        for (TableScan scan : scans) {
            TableReference reference = tableReference(scan, bindingName, scope);
            FederationTableStatistics statistics = statisticsSnapshot.statistics(
                sourceId,
                reference.schema(),
                reference.table()
            );
            totalScanFields += scan.getRowType().getFieldCount();
            if (statistics == null) {
                missingStatistics++;
                continue;
            }
            FederationStatisticsStatus status = statistics.effectiveStatus(
                statisticsSnapshot.capturedAt()
            );
            if (status == FederationStatisticsStatus.MISSING) {
                missingStatistics++;
                continue;
            }
            if (status == FederationStatisticsStatus.STALE) {
                staleStatistics = true;
                missingStatistics++;
                sources.add(statistics.source());
                if (oldestCollection == null
                    || statistics.collectedAt().isBefore(oldestCollection)) {
                    oldestCollection = statistics.collectedAt();
                }
                // 过期值只用于诊断，不参与行数、行宽或 Join 构建侧决策。
                continue;
            } else if (status == FederationStatisticsStatus.PARTIAL) {
                partialStatistics++;
            }
            double defaultTableRows = Math.max(
                1D,
                safeEstimate(metadata.getRowCount(scan), 100D)
            );
            scale = saturatingMultiply(scale, statistics.estimatedRows() / defaultTableRows);
            providedWidth += statistics.averageRowWidthBytes();
            providedFields += scan.getRowType().getFieldCount();
            sources.add(statistics.source());
            if (oldestCollection == null || statistics.collectedAt().isBefore(oldestCollection)) {
                oldestCollection = statistics.collectedAt();
            }
        }

        double estimatedRows = Math.max(0D, saturatingMultiply(calciteRows, scale));
        Double calciteAverageRowSize = metadata.getAverageRowSize(fragmentRoot);
        long rowWidth = calciteAverageRowSize == null || !Double.isFinite(calciteAverageRowSize)
            ? Math.max(1L, fragmentRoot.getRowType().getFieldCount() * 16L)
            : Math.max(1L, (long) Math.ceil(calciteAverageRowSize));
        if (providedFields > 0 && totalScanFields > 0) {
            double projectedRatio = Math.min(
                1D,
                (double) fragmentRoot.getRowType().getFieldCount() / totalScanFields
            );
            long providedEstimate = Math.max(
                1L,
                (long) Math.ceil(providedWidth * projectedRatio)
            );
            rowWidth = missingStatistics > 0
                ? Math.max(rowWidth, providedEstimate)
                : providedEstimate;
        }
        double transferBytes = saturatingMultiply(estimatedRows, rowWidth);
        boolean statisticsMissing = scans.isEmpty() || missingStatistics > 0 || sources.isEmpty();
        FederationStatisticsStatus statisticsStatus;
        if (staleStatistics) {
            statisticsStatus = FederationStatisticsStatus.STALE;
        } else if (sources.isEmpty()) {
            statisticsStatus = FederationStatisticsStatus.MISSING;
        } else if (missingStatistics > 0 || partialStatistics > 0) {
            statisticsStatus = FederationStatisticsStatus.PARTIAL;
        } else {
            statisticsStatus = FederationStatisticsStatus.COMPLETE;
        }
        return new FederationCostEstimate(
            estimatedRows,
            rowWidth,
            transferBytes,
            sources.isEmpty() ? "calcite-default" : String.join(",", sources),
            statisticsSnapshot.version(),
            oldestCollection == null ? Instant.EPOCH : oldestCollection,
            statisticsMissing,
            statisticsStatus
        );
    }

    private RelNode applyStatistics(
        RelNode root,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        return root.accept(new RelHomogeneousShuttle() {
            @Override
            public RelNode visit(TableScan scan) {
                if (!(scan instanceof LogicalTableScan)) {
                    return scan;
                }
                String bindingName = bindingFromQualifiedName(
                    scan.getTable().getQualifiedName(),
                    scope,
                    snapshot
                );
                SourceRuntime runtime = snapshot.runtime(bindingName);
                TableReference reference = tableReference(scan, bindingName, scope);
                FederationTableStatistics statistics = statisticsSnapshot.statistics(
                    runtime.definition().sourceId(),
                    reference.schema(),
                    reference.table()
                );
                FederationStatisticsStatus status = statistics == null
                    ? FederationStatisticsStatus.MISSING
                    : statistics.effectiveStatus(statisticsSnapshot.capturedAt());
                if (status == FederationStatisticsStatus.MISSING
                    || status == FederationStatisticsStatus.STALE) {
                    return scan;
                }
                return new FederationStatisticsTableScan(
                    scan.getCluster(),
                    scan.getTraitSet(),
                    scan.getHints(),
                    scan.getTable(),
                    statistics,
                    runtime.definition().sourceId()
                );
            }
        });
    }

    private static TableReference tableReference(
        TableScan scan,
        String bindingName,
        FederationQueryScopeDefinition scope
    ) {
        List<String> qualifiedName = scan.getTable().getQualifiedName();
        String table = qualifiedName.isEmpty()
            ? ""
            : qualifiedName.get(qualifiedName.size() - 1);
        String logicalSchema = qualifiedName.size() < 2
            ? ""
            : qualifiedName.get(qualifiedName.size() - 2);
        FederationSourceBindingDefinition binding = scope.bindings().get(bindingName);
        String physicalSchema = binding == null
            ? logicalSchema
            : binding.schemaMappings().getOrDefault(logicalSchema, logicalSchema);
        return new TableReference(physicalSchema, table);
    }

    private static Set<FederationStatisticsSnapshot.TableKey> referencedStatisticsTables(
        RelNode root,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        LinkedHashSet<FederationStatisticsSnapshot.TableKey> tables = new LinkedHashSet<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof TableScan scan) {
                    String binding = bindingFromQualifiedName(
                        scan.getTable().getQualifiedName(),
                        scope,
                        snapshot
                    );
                    TableReference reference = tableReference(scan, binding, scope);
                    tables.add(new FederationStatisticsSnapshot.TableKey(
                        snapshot.runtime(binding).definition().sourceId(),
                        reference.schema(),
                        reference.table()
                    ));
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        return Set.copyOf(tables);
    }

    private static List<String> pushedDownOperators(RelNode root) {
        LinkedHashSet<String> operators = new LinkedHashSet<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                operators.add(node.getRelTypeName());
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        return List.copyOf(operators);
    }

    private static double safeEstimate(Double value, double fallback) {
        return value == null || !Double.isFinite(value) || value < 0 ? fallback : value;
    }

    private static double saturatingMultiply(double left, double right) {
        if (left == 0D || right == 0D) {
            return 0D;
        }
        if (!Double.isFinite(left) || !Double.isFinite(right)
            || left > Double.MAX_VALUE / right) {
            return Double.MAX_VALUE;
        }
        return left * right;
    }

    private RelNode optimizeJoinBuildSides(
        RelNode node,
        RelSourceAnalysis analysis,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        List<RelNode> rewrittenInputs = node.getInputs().stream()
            .map(input -> optimizeJoinBuildSides(
                input,
                analysis,
                scope,
                snapshot,
                statisticsSnapshot
            ))
            .toList();
        RelNode rewritten = rewrittenInputs.equals(node.getInputs())
            ? node
            : node.copy(node.getTraitSet(), rewrittenInputs);
        if (!(node instanceof Join originalJoin)
            || !(rewritten instanceof Join rewrittenJoin)
            || rewrittenJoin.getJoinType() != JoinRelType.INNER
            || rewrittenJoin.analyzeCondition().pairs().isEmpty()) {
            return rewritten;
        }
        Set<SourceId> leftSources = analysis.sources(originalJoin.getLeft());
        Set<SourceId> rightSources = analysis.sources(originalJoin.getRight());
        LinkedHashSet<SourceId> joinedSources = new LinkedHashSet<>(leftSources);
        joinedSources.addAll(rightSources);
        if (leftSources.isEmpty() || rightSources.isEmpty() || joinedSources.size() <= 1) {
            return rewritten;
        }
        FederationCostEstimate leftCost = estimateJoinInputCost(
            rewrittenJoin.getLeft(),
            leftSources,
            analysis,
            scope,
            snapshot,
            statisticsSnapshot
        );
        FederationCostEstimate rightCost = estimateJoinInputCost(
            rewrittenJoin.getRight(),
            rightSources,
            analysis,
            scope,
            snapshot,
            statisticsSnapshot
        );
        // EnumerableHashJoin 构建右侧 Hash Table；表级估算可信且收益明确时才交换。
        if (!hasUsableBuildEstimate(leftCost)
            || !hasUsableBuildEstimate(rightCost)
            || leftCost.estimatedTransferBytes() >= rightCost.estimatedTransferBytes()) {
            return rewritten;
        }
        RelNode swapped = JoinCommuteRule.swap(rewrittenJoin, true);
        return swapped == null ? rewritten : swapped;
    }

    private List<FederationJoinOptimization> joinOptimizations(
        RelNode root,
        RelSourceAnalysis analysis,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        List<FederationJoinOptimization> optimizations = new ArrayList<>();
        collectJoinOptimizations(
            root,
            analysis,
            scope,
            snapshot,
            statisticsSnapshot,
            optimizations
        );
        return List.copyOf(optimizations);
    }

    private void collectJoinOptimizations(
        RelNode node,
        RelSourceAnalysis analysis,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot,
        List<FederationJoinOptimization> target
    ) {
        for (RelNode input : node.getInputs()) {
            collectJoinOptimizations(
                input,
                analysis,
                scope,
                snapshot,
                statisticsSnapshot,
                target
            );
        }
        if (node instanceof Join join && !join.analyzeCondition().pairs().isEmpty()) {
            appendJoinOptimization(
                join,
                analysis,
                scope,
                snapshot,
                statisticsSnapshot,
                target
            );
        }
    }

    private void appendJoinOptimization(
        Join join,
        RelSourceAnalysis analysis,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot,
        List<FederationJoinOptimization> target
    ) {
        Set<SourceId> leftSources = analysis.sources(join.getLeft());
        Set<SourceId> rightSources = analysis.sources(join.getRight());
        LinkedHashSet<SourceId> joinedSources = new LinkedHashSet<>(leftSources);
        joinedSources.addAll(rightSources);
        if (leftSources.isEmpty() || rightSources.isEmpty() || joinedSources.size() <= 1) {
            return;
        }
        List<String> leftBindings = bindingsForSources(leftSources, analysis);
        List<String> rightBindings = bindingsForSources(rightSources, analysis);
        FederationCostEstimate leftCost = estimateJoinInputCost(
            join.getLeft(),
            leftSources,
            analysis,
            scope,
            snapshot,
            statisticsSnapshot
        );
        FederationCostEstimate rightCost = estimateJoinInputCost(
            join.getRight(),
            rightSources,
            analysis,
            scope,
            snapshot,
            statisticsSnapshot
        );
        FederationJoinSelectionReason reason;
        if (join.getJoinType() != JoinRelType.INNER) {
            reason = FederationJoinSelectionReason.JOIN_SEMANTICS;
        } else if (hasUsableBuildEstimate(leftCost)
            && hasUsableBuildEstimate(rightCost)) {
            reason = FederationJoinSelectionReason.SMALLER_BUILD_SIDE;
        } else {
            reason = FederationJoinSelectionReason.INCOMPLETE_STATISTICS;
        }
        target.add(new FederationJoinOptimization(
            target.size() + 1,
            leftBindings,
            rightBindings,
            String.join(" + ", rightBindings),
            FederationJoinAlgorithm.HASH_JOIN,
            reason,
            rightCost.estimatedTransferBytes()
        ));
    }

    private FederationCostEstimate estimateJoinInputCost(
        RelNode input,
        Set<SourceId> sources,
        RelSourceAnalysis analysis,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot,
        FederationStatisticsSnapshot statisticsSnapshot
    ) {
        if (sources.size() == 1) {
            SourceId sourceId = sources.iterator().next();
            return estimateCost(
                input,
                sourceId,
                analysis.bindingForSource(sourceId),
                scope,
                statisticsSnapshot
            );
        }
        List<FederationCostEstimate> scanCosts = new ArrayList<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof TableScan scan) {
                    String binding = bindingFromQualifiedName(
                        scan.getTable().getQualifiedName(),
                        scope,
                        snapshot
                    );
                    scanCosts.add(estimateCost(
                        scan,
                        snapshot.runtime(binding).definition().sourceId(),
                        binding,
                        scope,
                        statisticsSnapshot
                    ));
                }
                super.visit(node, ordinal, parent);
            }
        }.go(input);
        var metadata = input.getCluster().getMetadataQuery();
        double rows = safeEstimate(metadata.getRowCount(input), 100D);
        Double averageSize = metadata.getAverageRowSize(input);
        long rowWidth = averageSize == null || !Double.isFinite(averageSize)
            ? Math.max(1L, input.getRowType().getFieldCount() * 16L)
            : Math.max(1L, (long) Math.ceil(averageSize));
        FederationStatisticsStatus status = aggregateStatisticsStatus(scanCosts);
        Instant oldestCollection = scanCosts.stream()
            .map(FederationCostEstimate::statisticsCollectedAt)
            .filter(value -> !Instant.EPOCH.equals(value))
            .min(Instant::compareTo)
            .orElse(Instant.EPOCH);
        String source = scanCosts.stream()
            .map(FederationCostEstimate::statisticsSource)
            .distinct()
            .sorted()
            .collect(Collectors.joining(","));
        return new FederationCostEstimate(
            rows,
            rowWidth,
            saturatingMultiply(rows, rowWidth),
            source.isBlank() ? "calcite-default" : source,
            statisticsSnapshot.version(),
            oldestCollection,
            scanCosts.stream().anyMatch(FederationCostEstimate::statisticsMissing),
            status,
            scanCosts.stream().allMatch(FederationCostEstimate::estimateAvailable)
        );
    }

    /**
     * 判断成本是否包含可用于 Hash Join 构建侧选择的表级估算。
     *
     * <p>数据库仅提供表行数和平均行宽时状态为 PARTIAL，但只要实际扫描表均有
     * 有效估算，就足以比较两侧搬运量。缺失或过期统计仍保留稳定输入顺序。</p>
     *
     * @param cost 输入成本
     * @return 可安全比较搬运量时为 true
     */
    private static boolean hasUsableBuildEstimate(FederationCostEstimate cost) {
        return cost.estimateAvailable()
            && !cost.statisticsMissing()
            && (cost.statisticsStatus() == FederationStatisticsStatus.COMPLETE
                || cost.statisticsStatus() == FederationStatisticsStatus.PARTIAL);
    }

    private static FederationStatisticsStatus aggregateStatisticsStatus(
        List<FederationCostEstimate> costs
    ) {
        if (costs.isEmpty()) {
            return FederationStatisticsStatus.MISSING;
        }
        if (costs.stream().anyMatch(cost ->
            cost.statisticsStatus() == FederationStatisticsStatus.STALE)) {
            return FederationStatisticsStatus.STALE;
        }
        boolean allComplete = costs.stream().allMatch(cost ->
            cost.statisticsStatus() == FederationStatisticsStatus.COMPLETE);
        if (allComplete) {
            return FederationStatisticsStatus.COMPLETE;
        }
        boolean allMissing = costs.stream().allMatch(cost ->
            cost.statisticsStatus() == FederationStatisticsStatus.MISSING);
        return allMissing
            ? FederationStatisticsStatus.MISSING
            : FederationStatisticsStatus.PARTIAL;
    }

    private static List<String> bindingsForSources(
        Set<SourceId> sources,
        RelSourceAnalysis analysis
    ) {
        return sources.stream()
            .map(analysis::bindingForSource)
            .distinct()
            .sorted(String.CASE_INSENSITIVE_ORDER)
            .toList();
    }

    private static String bindingFromQualifiedName(
        List<String> qualifiedName,
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        for (String component : qualifiedName) {
            for (String binding : snapshot.runtimesByBinding().keySet()) {
                if (binding.equals(component) || binding.equalsIgnoreCase(component)) {
                    return binding;
                }
            }
        }
        if (snapshot.runtimesByBinding().size() == 1) {
            return snapshot.runtimesByBinding().keySet().iterator().next();
        }
        if (snapshot.runtimesByBinding().containsKey(scope.defaultBinding())) {
            return scope.defaultBinding();
        }
        throw new FederationSqlException(
            FederationSqlErrorCode.INVALID_QUERY_SCOPE,
            "validated table cannot be assigned to a query binding: " + qualifiedName
        );
    }

    private static List<RelNode> maximalSingleSourceFragments(
        RelNode root,
        RelSourceAnalysis analysis
    ) {
        List<RelNode> fragments = new ArrayList<>();
        collectFragments(root, Set.of(), analysis, fragments);
        return List.copyOf(fragments);
    }

    private static void collectFragments(
        RelNode node,
        Set<SourceId> parentSources,
        RelSourceAnalysis analysis,
        List<RelNode> fragments
    ) {
        Set<SourceId> current = analysis.sources(node);
        if (current.size() == 1 && parentSources.size() > 1) {
            fragments.add(node);
            return;
        }
        for (RelNode input : node.getInputs()) {
            collectFragments(input, current, analysis, fragments);
        }
    }

    private static LocalLogicalPlan replaceFragments(
        RelNode root,
        IdentityHashMap<RelNode, CompiledFederationFragment> fragments
    ) {
        SchemaPlus rootSchema = CalciteSchema.createRootSchema(true, false).plus();
        SchemaPlus fragmentSchema = rootSchema.add("__fragments", new AbstractSchema());
        IdentityHashMap<RelNode, RelNode> replacements = new IdentityHashMap<>();
        fragments.forEach((fragmentRoot, compiled) -> {
            FederationFragmentTable table = new FederationFragmentTable(
                compiled.plan().fragmentId(),
                compiled.rowType(),
                compiled.plan().costEstimate().estimatedRows()
            );
            fragmentSchema.add(compiled.plan().fragmentId(), table);
            RelOptTableImpl relOptTable = RelOptTableImpl.create(
                null,
                compiled.rowType(),
                List.of("__fragments", compiled.plan().fragmentId()),
                table,
                requestedClass -> Schemas.getTableExpression(
                    fragmentSchema,
                    compiled.plan().fragmentId(),
                    table,
                    requestedClass
                )
            );
            replacements.put(
                fragmentRoot,
                LogicalTableScan.create(fragmentRoot.getCluster(), relOptTable, List.of())
            );
        });
        RelNode replaced = root.accept(new RelHomogeneousShuttle() {
            @Override
            public RelNode visit(RelNode node) {
                RelNode replacement = replacements.get(node);
                return replacement == null ? super.visit(node) : replacement;
            }
        });
        return new LocalLogicalPlan(replaced, rootSchema);
    }

    private static void validateLocalOperators(RelNode root) {
        AtomicReference<FederationSqlException> failure = new AtomicReference<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (failure.get() != null) {
                    return;
                }
                if (hasTemporalPrecisionBeyondMilliseconds(node.getRowType())) {
                    failure.set(unsupported(
                        "local federation supports TIME/TIMESTAMP precision up to 3 digits"
                    ));
                    return;
                }
                if (node instanceof Join join) {
                    org.apache.calcite.rel.core.JoinInfo joinInfo = join.analyzeCondition();
                    if ((join.getJoinType() != JoinRelType.INNER
                        && join.getJoinType() != JoinRelType.LEFT)
                        || joinInfo.pairs().isEmpty()) {
                        failure.set(unsupported("only equi INNER and LEFT JOIN are supported"));
                        return;
                    }
                    if (hasCharacterJoinKey(join, joinInfo)) {
                        failure.set(unsupported(
                            "character join keys require an explicit cross-source collation adapter"
                        ));
                        return;
                    }
                    if (hasCharacterComparison(join.getCondition())) {
                        failure.set(unsupported(
                            "local character comparisons require an explicit "
                                + "cross-source collation adapter"
                        ));
                        return;
                    }
                } else if (node instanceof Union union) {
                    if (!union.all) {
                        failure.set(unsupported("only UNION ALL is supported"));
                        return;
                    }
                } else if (node instanceof Aggregate aggregate) {
                    if (hasCharacterGroupingOrAggregate(aggregate)) {
                        failure.set(unsupported(
                            "local character grouping and aggregates require an explicit "
                                + "cross-source collation adapter"
                        ));
                        return;
                    }
                    if (aggregate.getGroupSets().size() != 1) {
                        failure.set(unsupported("GROUPING SETS, ROLLUP and CUBE are not supported"));
                        return;
                    }
                    for (AggregateCall call : aggregate.getAggCallList()) {
                        SqlKind kind = call.getAggregation().getKind();
                        if (call.isDistinct() || !(kind == SqlKind.COUNT
                            || kind == SqlKind.SUM
                            || kind == SqlKind.SUM0
                            || kind == SqlKind.MIN
                            || kind == SqlKind.MAX
                            || kind == SqlKind.AVG)) {
                            failure.set(unsupported(
                                "aggregate is not supported: " + call.getAggregation().getName()
                            ));
                            return;
                        }
                    }
                } else if (node instanceof Sort sort) {
                    if (hasCharacterSortKey(sort)) {
                        failure.set(unsupported(
                            "local character sorting requires an explicit "
                                + "cross-source collation adapter"
                        ));
                        return;
                    }
                } else if (node instanceof Filter filter) {
                    if (hasCharacterComparison(filter.getCondition())) {
                        failure.set(unsupported(
                            "local character comparisons require an explicit "
                                + "cross-source collation adapter"
                        ));
                        return;
                    }
                } else if (node instanceof Project project) {
                    if (project.getProjects().stream()
                        .anyMatch(CalciteFederationSqlCompiler::hasCharacterComparison)) {
                        failure.set(unsupported(
                            "local character comparisons require an explicit "
                                + "cross-source collation adapter"
                        ));
                        return;
                    }
                } else if (!(node instanceof Project)
                    && !(node instanceof Filter)
                    && !(node instanceof Values)
                    && !(node instanceof TableScan)) {
                    failure.set(unsupported(
                        "local relational operator is not supported: " + node.getRelTypeName()
                    ));
                    return;
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        if (failure.get() != null) {
            throw failure.get();
        }
    }

    /**
     * 在优化器改写前校验跨源 Join 条件，避免 WHERE 残余谓词被合并进 Join 后误判。
     *
     * @param root 优化前关系树
     * @param analysis 关系树来源分析
     * @throws FederationSqlException 跨源 Join 使用非纯等值 ON 条件时抛出
     */
    private static void validateCrossSourceJoinConditions(
        RelNode root,
        RelSourceAnalysis analysis
    ) {
        AtomicReference<FederationSqlException> failure = new AtomicReference<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (failure.get() != null) {
                    return;
                }
                if (node instanceof Join join) {
                    Set<SourceId> leftSources = analysis.sources(join.getLeft());
                    Set<SourceId> rightSources = analysis.sources(join.getRight());
                    LinkedHashSet<SourceId> joinedSources =
                        new LinkedHashSet<>(leftSources);
                    joinedSources.addAll(rightSources);
                    boolean crossesSources = !leftSources.isEmpty()
                        && !rightSources.isEmpty()
                        && joinedSources.size() > 1;
                    if (crossesSources && !join.analyzeCondition().isEqui()) {
                        failure.set(unsupported(
                            "cross-source JOIN ON supports equality predicates only"
                        ));
                        return;
                    }
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        if (failure.get() != null) {
            throw failure.get();
        }
    }

    /**
     * 判断本地 Enumerable 是否会截断时间字段的亚毫秒精度。
     *
     * @param rowType 本地算子行类型
     * @return 是否存在超过毫秒的时间精度
     */
    private static boolean hasTemporalPrecisionBeyondMilliseconds(RelDataType rowType) {
        return rowType.getFieldList().stream().anyMatch(field -> {
            RelDataType type = field.getType();
            return switch (type.getSqlTypeName()) {
                case TIME, TIME_WITH_LOCAL_TIME_ZONE, TIME_TZ,
                     TIMESTAMP, TIMESTAMP_WITH_LOCAL_TIME_ZONE, TIMESTAMP_TZ ->
                    type.getPrecision() > 3;
                default -> false;
            };
        });
    }

    private static boolean hasCharacterJoinKey(
        Join join,
        org.apache.calcite.rel.core.JoinInfo joinInfo
    ) {
        for (int index = 0; index < joinInfo.leftKeys.size(); index++) {
            RelDataType leftType = join.getLeft().getRowType().getFieldList()
                .get(joinInfo.leftKeys.getInt(index)).getType();
            RelDataType rightType = join.getRight().getRowType().getFieldList()
                .get(joinInfo.rightKeys.getInt(index)).getType();
            if (SqlTypeUtil.isCharacter(leftType) || SqlTypeUtil.isCharacter(rightType)) {
                return true;
            }
        }
        return false;
    }

    private static boolean hasCharacterComparison(RexNode expression) {
        AtomicReference<Boolean> found = new AtomicReference<>(false);
        expression.accept(new RexVisitorImpl<Void>(true) {
            @Override
            public Void visitCall(RexCall call) {
                if (isCharacterSemanticOperation(call.getKind())
                    && call.getOperands().stream()
                    .anyMatch(operand -> SqlTypeUtil.isCharacter(operand.getType()))) {
                    found.set(true);
                    return null;
                }
                return super.visitCall(call);
            }
        });
        return found.get();
    }

    private static boolean isCharacterSemanticOperation(SqlKind kind) {
        return SqlKind.COMPARISON.contains(kind)
            || kind == SqlKind.LIKE
            || kind == SqlKind.RLIKE
            || kind == SqlKind.SIMILAR
            || kind == SqlKind.POSIX_REGEX_CASE_SENSITIVE
            || kind == SqlKind.POSIX_REGEX_CASE_INSENSITIVE
            || kind == SqlKind.SEARCH
            || kind == SqlKind.IN
            || kind == SqlKind.NOT_IN
            || kind == SqlKind.BETWEEN
            || kind == SqlKind.STARTS_WITH
            || kind == SqlKind.ENDS_WITH
            || kind == SqlKind.CONTAINS_SUBSTR;
    }

    private static boolean hasCharacterSortKey(Sort sort) {
        return sort.getCollation().getFieldCollations().stream()
            .map(field -> sort.getInput().getRowType().getFieldList()
                .get(field.getFieldIndex()).getType())
            .anyMatch(SqlTypeUtil::isCharacter);
    }

    private static boolean hasCharacterGroupingOrAggregate(Aggregate aggregate) {
        List<RelDataTypeField> fields = aggregate.getInput().getRowType().getFieldList();
        if (aggregate.getGroupSet().asList().stream()
            .map(index -> fields.get(index).getType())
            .anyMatch(SqlTypeUtil::isCharacter)) {
            return true;
        }
        return aggregate.getAggCallList().stream().anyMatch(call ->
            (call.getAggregation().getKind() == SqlKind.MIN
                || call.getAggregation().getKind() == SqlKind.MAX)
                && call.getArgList().stream()
                .map(index -> fields.get(index).getType())
                .anyMatch(SqlTypeUtil::isCharacter)
        );
    }

    private static RelNode optimizeForFederation(RelNode root) {
        HepPlanner normalizationPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addRuleInstance(CoreRules.FILTER_INTO_JOIN)
                .addRuleInstance(CoreRules.PROJECT_JOIN_TRANSPOSE)
                .addRuleInstance(CoreRules.FILTER_PROJECT_TRANSPOSE)
                .addRuleInstance(CoreRules.FILTER_MERGE)
                .addRuleInstance(CoreRules.PROJECT_MERGE)
                .addRuleInstance(CoreRules.PROJECT_REMOVE)
                .build()
        );
        normalizationPlanner.setRoot(root);
        RelNode normalized = normalizationPlanner.findBestExp();
        if (RelOptUtil.countJoins(normalized) < 2 || containsNonInnerJoin(normalized)) {
            return normalized;
        }

        if (supportsStatisticsDrivenJoinSearch(normalized)) {
            RelNode optimized = optimizeWithHyperGraph(normalized);
            if (optimized != null) {
                return optimized;
            }
        }
        return optimizeWithMultiJoin(normalized);
    }

    /**
     * 判断计划是否适合有界的统计驱动 Join 搜索。
     *
     * <p>首版仅处理三个不同物理源、两个简单 INNER 等值 Join，避免搜索空间失控，
     * 统计缺失时则沿用稳定的基础规则。</p>
     *
     * @param root 规范化计划
     * @return 是否允许启用 DPHyp 搜索
     */
    private static boolean supportsStatisticsDrivenJoinSearch(RelNode root) {
        List<FederationStatisticsTableScan> scans = new ArrayList<>();
        AtomicReference<Boolean> eligible = new AtomicReference<>(true);
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof Join join
                    && (join.getJoinType() != JoinRelType.INNER
                        || !join.analyzeCondition().isEqui()
                        || join.analyzeCondition().pairs().isEmpty())) {
                    eligible.set(false);
                    return;
                }
                if (node instanceof TableScan scan) {
                    if (!(scan instanceof FederationStatisticsTableScan statisticsScan)
                        || statisticsScan.statistics().status()
                            != FederationStatisticsStatus.COMPLETE) {
                        eligible.set(false);
                        return;
                    }
                    scans.add(statisticsScan);
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        return eligible.get()
            && RelOptUtil.countJoins(root) == 2
            && scans.size() == 3
            && scans.stream().map(FederationStatisticsTableScan::sourceId)
                .collect(Collectors.toSet()).size() == 3;
    }

    /**
     * 使用 Calcite DPHyp 在受控三表范围内按累计联邦成本选择结合顺序。
     *
     * @param root 规范化计划
     * @return 优化计划；规则无法完整消解 HyperGraph 时返回 null
     */
    private static RelNode optimizeWithHyperGraph(RelNode root) {
        HepPlanner graphPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addMatchOrder(HepMatchOrder.BOTTOM_UP)
                .addRuleInstance(CoreRules.JOIN_TO_HYPER_GRAPH)
                .build()
        );
        graphPlanner.setRoot(root);
        RelNode graph = graphPlanner.findBestExp();
        if (!containsHyperGraph(graph)) {
            return null;
        }
        HepPlanner orderPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addRuleInstance(CoreRules.HYPER_GRAPH_OPTIMIZE)
                .build()
        );
        orderPlanner.setRoot(graph);
        RelNode optimized = orderPlanner.findBestExp();
        return containsHyperGraph(optimized) ? null : optimized;
    }

    /**
     * 使用基础 MultiJoin 规则生成稳定的 bushy 计划。
     *
     * @param root 规范化计划
     * @return Join 重排后的计划
     */
    private static RelNode optimizeWithMultiJoin(RelNode root) {
        HepPlanner joinPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addMatchOrder(HepMatchOrder.BOTTOM_UP)
                .addRuleInstance(CoreRules.JOIN_TO_MULTI_JOIN)
                .build()
        );
        joinPlanner.setRoot(root);
        RelNode joinGraph = joinPlanner.findBestExp();
        HepPlanner joinOrderPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addRuleInstance(CoreRules.MULTI_JOIN_OPTIMIZE_BUSHY)
                .build()
        );
        joinOrderPlanner.setRoot(joinGraph);
        return joinOrderPlanner.findBestExp();
    }

    /**
     * 判断计划中是否仍含 DPHyp 的临时 HyperGraph 节点。
     *
     * @param root 计划根节点
     * @return 是否包含临时节点
     */
    private static boolean containsHyperGraph(RelNode root) {
        AtomicReference<Boolean> found = new AtomicReference<>(false);
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof HyperGraph) {
                    found.set(true);
                    return;
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        return found.get();
    }

    /**
     * 判断计划是否包含不能安全改变结合顺序的外连接。
     *
     * @param root 关系计划根节点
     * @return 是否包含非 INNER Join
     */
    private static boolean containsNonInnerJoin(RelNode root) {
        AtomicReference<Boolean> found = new AtomicReference<>(false);
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof Join join && join.getJoinType() != JoinRelType.INNER) {
                    found.set(true);
                    return;
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        return found.get();
    }

    private static Bindable<?> compileLocalBindable(RelNode localLogical) {
        HepPlanner normalizationPlanner = new HepPlanner(
            new HepProgramBuilder()
                .addRuleInstance(CoreRules.FILTER_TO_CALC)
                .addRuleInstance(CoreRules.PROJECT_TO_CALC)
                .addRuleInstance(CoreRules.FILTER_CALC_MERGE)
                .addRuleInstance(CoreRules.PROJECT_CALC_MERGE)
                .addRuleInstance(CoreRules.CALC_MERGE)
                .build()
        );
        normalizationPlanner.setRoot(localLogical);
        RelNode normalized = normalizationPlanner.findBestExp();
        RelNode converted = Programs.ofRules(LOCAL_ENUMERABLE_RULES).run(
            normalized.getCluster().getPlanner(),
            normalized,
            normalized.getTraitSet().replace(EnumerableConvention.INSTANCE),
            List.of(),
            List.of()
        );
        RelNode guarded = addLocalBudgetGuards(converted);
        if (!(guarded instanceof EnumerableRel enumerable)) {
            throw unsupported(
                "Calcite could not convert the local plan to Enumerable convention"
            );
        }
        return EnumerableInterpretable.toBindable(
            Map.of(),
            (CalcitePrepare.SparkHandler) null,
            enumerable,
            EnumerableRel.Prefer.ARRAY
        );
    }

    private static RelNode addLocalBudgetGuards(RelNode node) {
        List<RelNode> inputs = node.getInputs();
        List<RelNode> guardedInputs = inputs.stream()
            .map(CalciteFederationSqlCompiler::addLocalBudgetGuards)
            .toList();
        RelNode guardedNode = inputs.equals(guardedInputs)
            ? node
            : node.copy(node.getTraitSet(), guardedInputs);
        if (guardedNode instanceof Join
            || guardedNode instanceof Aggregate
            || guardedNode instanceof Sort
            || guardedNode instanceof Union
            || guardedNode instanceof org.apache.calcite.rel.core.Calc) {
            return new EnumerableFederationBudgetRel(
                guardedNode,
                guardedNode.getRelTypeName()
            );
        }
        return guardedNode;
    }

    private static FederationSqlException unsupported(String message) {
        return new FederationSqlException(
            FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED,
            message
        );
    }

    private static SqlString toSql(RelNode relNode, SourceRuntime runtime) {
        return new RelToSqlConverter(runtime.dialect())
            .visitRoot(relNode)
            .asStatement()
            .toSqlString(runtime.dialect());
    }

    private static List<FederationSourceRuntimeIdentity> identitiesForBindings(
        FederationQueryScopeSnapshot snapshot,
        Set<String> bindings
    ) {
        return snapshot.identities().stream()
            .filter(identity -> bindings.contains(identity.bindingName()))
            .toList();
    }

    private static void discoverBindings(
        SqlNode node,
        FederationQueryScopeDefinition scope,
        Set<String> visibleCteNames,
        Set<String> bindings
    ) {
        if (node == null) {
            return;
        }
        if (node instanceof SqlWith with) {
            Set<String> withScope = new HashSet<>(visibleCteNames);
            for (SqlNode entry : with.withList) {
                SqlWithItem item = (SqlWithItem) entry;
                Set<String> itemScope = new HashSet<>(withScope);
                if (item.recursive != null && item.recursive.booleanValue()) {
                    // 只有递归 CTE 的查询体可以引用自身。
                    itemScope.add(normalizeCteName(item.name));
                }
                discoverBindings(item.query, scope, itemScope, bindings);
                withScope.add(normalizeCteName(item.name));
            }
            discoverBindings(with.body, scope, withScope, bindings);
            return;
        }
        if (node instanceof SqlSelect select) {
            collectFromBindings(select.getFrom(), scope, visibleCteNames, bindings);
        }
        if (node instanceof SqlCall call) {
            for (SqlNode operand : call.getOperandList()) {
                discoverBindings(operand, scope, visibleCteNames, bindings);
            }
            return;
        }
        if (node instanceof org.apache.calcite.sql.SqlNodeList list) {
            for (SqlNode child : list) {
                discoverBindings(child, scope, visibleCteNames, bindings);
            }
        }
    }

    private static String normalizeCteName(SqlIdentifier identifier) {
        // 预解析使用 Lex.ORACLE：未引号名称已转大写，引号名称保留原始大小写。
        return identifier.getSimple();
    }

    private static void collectFromBindings(
        SqlNode from,
        FederationQueryScopeDefinition scope,
        Set<String> visibleCteNames,
        Set<String> bindings
    ) {
        if (from == null || from instanceof SqlSelect) {
            return;
        }
        if (from instanceof SqlIdentifier identifier) {
            if (identifier.names.size() == 1
                && visibleCteNames.contains(normalizeCteName(identifier))) {
                return;
            }
            if (identifier.names.size() == 1) {
                String logicalBinding = LogicalTableSqlResolver.bindingForShortName(
                    identifier.getSimple(),
                    scope
                );
                bindings.add(logicalBinding == null ? scope.defaultBinding() : logicalBinding);
            } else if (identifier.names.size() >= 3) {
                String first = identifier.names.get(0);
                scope.bindings().keySet().stream()
                    .filter(binding -> identifier.isComponentQuoted(0)
                        ? binding.equals(first)
                        : binding.equalsIgnoreCase(first))
                    .findFirst()
                    .ifPresentOrElse(bindings::add, () -> bindings.add(scope.defaultBinding()));
            } else {
                bindings.add(scope.defaultBinding());
            }
            return;
        }
        if (from instanceof SqlJoin join) {
            collectFromBindings(join.getLeft(), scope, visibleCteNames, bindings);
            collectFromBindings(join.getRight(), scope, visibleCteNames, bindings);
            return;
        }
        if (from instanceof SqlCall call && (call.getKind() == SqlKind.AS
            || call.getKind() == SqlKind.LATERAL
            || call.getKind() == SqlKind.TABLESAMPLE
            || call.getKind() == SqlKind.SNAPSHOT)) {
            List<SqlNode> operands = call.getOperandList();
            if (!operands.isEmpty()) {
                collectFromBindings(operands.get(0), scope, visibleCteNames, bindings);
            }
        }
    }

    private static void requireReadOnly(SqlNode parsed) {
        if (!SqlKind.QUERY.contains(parsed.getKind())) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_NOT_READ_ONLY,
                "only one read-only query is allowed"
            );
        }
    }

    private static int countDynamicParameters(SqlNode node) {
        Set<Integer> parameterIndexes = new HashSet<>();
        node.accept(new SqlBasicVisitor<Void>() {
            @Override
            public Void visit(SqlDynamicParam parameter) {
                parameterIndexes.add(parameter.getIndex());
                return null;
            }
        });
        return parameterIndexes.size();
    }

    private static void requireMatchingParameterTypes(
        SqlCompileRequest request,
        int dynamicParameterCount
    ) {
        if (!request.parameterJdbcTypes().isEmpty()
            && request.parameterJdbcTypes().size() != dynamicParameterCount) {
            throw new FederationSqlException(
                FederationSqlErrorCode.PARAMETER_COUNT_MISMATCH,
                "SQL expects " + dynamicParameterCount + " parameters but received "
                    + request.parameterJdbcTypes().size() + " types"
            );
        }
    }

    private static SqlNode applyDeclaredParameterTypes(
        SqlNode parsed,
        SqlCompileRequest request,
        SourceRuntime runtime
    ) {
        if (request.parameterJdbcTypes().isEmpty()) {
            return parsed;
        }
        Set<Integer> paginationParameters = paginationParameterIndexes(parsed);
        return parsed.accept(new SqlShuttle() {
            @Override
            public SqlNode visit(SqlDynamicParam parameter) {
                int jdbcType = request.parameterJdbcTypes().get(parameter.getIndex());
                SqlDataTypeSpec typeSpec = runtime.adapter().parameterTypeSpec(
                    jdbcType,
                    parameter.getParserPosition()
                );
                if (typeSpec == null) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.INVALID_ARGUMENT,
                        "JDBC parameter type " + jdbcType + " has no scalar Calcite mapping"
                    );
                }
                if (!paginationParameters.contains(parameter.getIndex())) {
                    return SqlStdOperatorTable.CAST.createCall(
                        parameter.getParserPosition(),
                        parameter,
                        typeSpec
                    );
                }
                return new TypedSqlDynamicParam(
                    parameter.getIndex(),
                    parameter.getParserPosition(),
                    typeSpec
                );
            }
        });
    }

    private static Set<Integer> paginationParameterIndexes(SqlNode parsed) {
        Set<Integer> indexes = new HashSet<>();
        parsed.accept(new SqlBasicVisitor<Void>() {
            @Override
            public Void visit(SqlCall call) {
                if (call instanceof SqlSelect select) {
                    addDynamicParameterIndex(select.getOffset(), indexes);
                    addDynamicParameterIndex(select.getFetch(), indexes);
                } else if (call instanceof SqlOrderBy orderBy) {
                    addDynamicParameterIndex(orderBy.offset, indexes);
                    addDynamicParameterIndex(orderBy.fetch, indexes);
                }
                return super.visit(call);
            }
        });
        return Set.copyOf(indexes);
    }

    private static void addDynamicParameterIndex(SqlNode node, Set<Integer> indexes) {
        if (node instanceof SqlDynamicParam parameter) {
            indexes.add(parameter.getIndex());
        }
    }

    private static RelNode stripSyntheticParameterCasts(RelNode root) {
        RexShuttle castStripper = new RexShuttle() {
            @Override
            public RexNode visitCall(RexCall call) {
                if (call.getKind() == SqlKind.CAST
                    && call.getOperands().size() == 1
                    && call.getOperands().get(0) instanceof RexDynamicParam) {
                    return call.getOperands().get(0);
                }
                return super.visitCall(call);
            }
        };
        return root.accept(new RelHomogeneousShuttle() {
            @Override
            public RelNode visit(RelNode node) {
                RelNode withRewrittenInputs = super.visit(node);
                return withRewrittenInputs.accept(castStripper);
            }
        });
    }

    private static String policyFingerprint(List<FederationSqlPolicy> policies) {
        StringBuilder fingerprint = new StringBuilder();
        fingerprint.append(policies.size()).append(':');
        for (FederationSqlPolicy policy : policies) {
            String version = policy.version();
            if (version == null || version.isBlank()) {
                throw new IllegalArgumentException(
                    "federation SQL policy version must not be blank"
                );
            }
            appendFingerprintField(fingerprint, policy.getClass().getName());
            appendFingerprintField(fingerprint, version);
        }
        return fingerprint.toString();
    }

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

    private static List<Integer> parameterMapping(
        SqlString targetSql,
        int parameterCount,
        boolean singleSource
    ) {
        List<Integer> mapping = targetSql.getDynamicParameters();
        if ((mapping == null || mapping.isEmpty()) && parameterCount > 0 && singleSource) {
            List<Integer> identity = new ArrayList<>(parameterCount);
            for (int index = 0; index < parameterCount; index++) {
                identity.add(index);
            }
            return List.copyOf(identity);
        }
        if (mapping == null) {
            return List.of();
        }
        for (Integer sourceIndex : mapping) {
            if (sourceIndex == null || sourceIndex < 0 || sourceIndex >= parameterCount) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.SQL_COMPILE_FAILED,
                    "Calcite returned an invalid dynamic parameter mapping"
                );
            }
        }
        return List.copyOf(mapping);
    }

    private static List<FederationColumn> columns(RelDataType rowType) {
        List<FederationColumn> columns = new ArrayList<>();
        for (RelDataTypeField field : rowType.getFieldList()) {
            SqlTypeName sqlTypeName = field.getType().getSqlTypeName();
            int jdbcType = switch (sqlTypeName) {
                case TIME_TZ -> java.sql.Types.TIME_WITH_TIMEZONE;
                case TIMESTAMP_TZ -> java.sql.Types.TIMESTAMP_WITH_TIMEZONE;
                default -> sqlTypeName.getJdbcOrdinal();
            };
            columns.add(new FederationColumn(
                field.getIndex() + 1,
                field.getName(),
                jdbcType,
                sqlTypeName.getName(),
                field.getType().isNullable()
            ));
        }
        return List.copyOf(columns);
    }

    private static void ensureNoEnumerableResidual(RelNode root) {
        AtomicReference<RelNode> enumerableNode = new AtomicReference<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node.getConvention() == EnumerableConvention.INSTANCE) {
                    enumerableNode.compareAndSet(null, node);
                    return;
                }
                super.visit(node, ordinal, parent);
            }
        }.go(root);
        RelNode residual = enumerableNode.get();
        if (residual != null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_NOT_FULLY_PUSHDOWN,
                "single-source plan contains an Enumerable residual: "
                    + residual.getRelTypeName()
            );
        }
    }

    private record UnifiedSchema(SchemaPlus defaultSchema) {
    }

    private record LocalLogicalPlan(RelNode root, SchemaPlus rootSchema) {
    }

    private record TableReference(String schema, String table) {
    }

    private record RelSourceAnalysis(
        IdentityHashMap<RelNode, Set<SourceId>> sourcesByNode,
        Map<SourceId, String> bindingsBySource
    ) {

        private Set<SourceId> sources(RelNode node) {
            return sourcesByNode.getOrDefault(node, Set.of());
        }

        private String bindingForSource(SourceId sourceId) {
            String binding = bindingsBySource.get(sourceId);
            if (binding == null) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.INVALID_QUERY_SCOPE,
                    "no query binding resolved for source: " + sourceId
                );
            }
            return binding;
        }
    }
}
