package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.adapter.AdapterDialectContext;
import com.easyagents.federation.sql.adapter.AdapterHints;
import com.easyagents.federation.sql.adapter.AdapterSchemaContext;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterProvider;
import com.easyagents.federation.sql.api.SqlCompletionItem;
import com.easyagents.federation.sql.api.SqlCompletionKind;
import com.easyagents.federation.sql.api.SqlCompletionRequest;
import com.easyagents.federation.sql.api.SqlCompletionResult;
import com.easyagents.federation.sql.compile.FederationSqlPlan;
import com.easyagents.federation.sql.compile.SqlCompileRequest;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.federation.FederationColumnStatistics;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationJoinOptimization;
import com.easyagents.federation.sql.federation.FederationJoinSelectionReason;
import com.easyagents.federation.sql.federation.FederationLogicalTableDefinition;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationQueryMode;
import com.easyagents.federation.sql.federation.FederationStatisticsSnapshot;
import com.easyagents.federation.sql.federation.FederationStatisticsStatus;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
import com.easyagents.federation.sql.federation.FederationTableStatistics;
import com.easyagents.federation.sql.federation.FederationTableStatisticsProvider;
import com.easyagents.federation.sql.source.ExternalSchemaDefinition;
import com.easyagents.federation.sql.source.FederationDataSourceHandles;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.RuntimeFingerprint;
import com.easyagents.federation.sql.source.SourceId;
import java.lang.reflect.Proxy;
import java.sql.DatabaseMetaData;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.sql.DataSource;
import org.apache.calcite.avatica.util.Casing;
import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelVisitor;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.rules.HyperGraph;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.sql.SqlDialect;
import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlDialect.DatabaseProduct;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.util.ImmutableBitSet;
import org.junit.Assert;
import org.junit.Test;

/**
 * Calcite 编译器的目标方言解析契约测试。
 */
public class CalciteFederationSqlCompilerTest {

    /**
     * 验证未加引号名称遵循目标数据库大小写，同时保留 ANSI 双引号输入。
     */
    @Test
    public void shouldUseTargetDialectIdentifierCasingWithAnsiQuotes() {
        SqlDialect dialect = new SqlDialect(SqlDialect.EMPTY_CONTEXT
            .withDatabaseProduct(DatabaseProduct.MYSQL)
            .withIdentifierQuoteString("`")
            .withQuotedCasing(Casing.UNCHANGED)
            .withUnquotedCasing(Casing.UNCHANGED)
            .withCaseSensitive(true));
        SchemaPlus rootSchema = CalciteSchema.createRootSchema(true, false).plus();
        SchemaPlus sourceSchema = rootSchema.add("main", new AbstractSchema());
        SchemaPlus defaultSchema = sourceSchema.add("main", lowerCaseSchema());
        FederationSourceDefinition definition = new FederationSourceDefinition(
            new SourceId("main"),
            1,
            "test-adapter",
            List.of(new ExternalSchemaDefinition("main", "test", 1)),
            Map.of()
        );
        DataSource dataSource = (DataSource) Proxy.newProxyInstance(
            getClass().getClassLoader(),
            new Class<?>[] {DataSource.class},
            (proxy, method, arguments) -> {
                throw new UnsupportedOperationException(method.getName());
            }
        );
        SourceRuntime runtime = new SourceRuntime(
            definition,
            FederationDataSourceHandles.shared(
                dataSource,
                new RuntimeFingerprint("test", "1", "test", "1", "1")
            ),
            new TestAdapter(dialect),
            dialect,
            rootSchema,
            defaultSchema,
            new AdapterCompatibility(
                AdapterCompatibilityStatus.VERIFIED,
                "test",
                "1",
                "test",
                "1",
                "test"
            ),
            ignored -> { }
        );
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(List.of(), false);
        Assert.assertTrue(runtime.adapter().parserConfig(dialect).caseSensitive());

        FederationSqlPlan unquoted = compiler.compile(
            SqlCompileRequest.of("SELECT id FROM orders", definition.sourceId(), 1),
            runtime,
            Set.of(definition.sourceId())
        );
        FederationSqlPlan quoted = compiler.compile(
            SqlCompileRequest.of("SELECT \"id\" FROM \"orders\"", definition.sourceId(), 1),
            runtime,
            Set.of(definition.sourceId())
        );

        Assert.assertTrue(unquoted.executable());
        Assert.assertTrue(quoted.executable());
        Assert.assertTrue(unquoted.executableSql().contains("`orders`"));
        Assert.assertTrue(quoted.executableSql().contains("`orders`"));
    }

    /**
     * 验证嵌套 CTE 名称不会污染外层物理表的 Binding 发现作用域。
     */
    @Test
    public void shouldDiscoverBindingsWithLexicallyScopedCteNames() {
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "cte-scope",
            1,
            Map.of(
                "DEFAULT", FederationSourceBindingDefinition.of(new SourceId("default"), 1),
                "REMOTE", FederationSourceBindingDefinition.of(new SourceId("remote"), 1)
            ),
            "DEFAULT",
            FederationExecutionPolicy.basic()
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM x WHERE EXISTS ("
                + "WITH x AS (SELECT * FROM REMOTE.APP.T) SELECT * FROM x)",
            scope
        );

        Set<String> bindings = new CalciteFederationSqlCompiler(List.of(), true)
            .discoverBindings(request);

        Assert.assertEquals(Set.of("DEFAULT", "REMOTE"), bindings);

        SqlCompileRequest nonRecursive = SqlCompileRequest.of(
            "WITH x AS (SELECT d.ID FROM x d JOIN REMOTE.APP.T r ON d.ID = r.ID) "
                + "SELECT * FROM x",
            scope
        );
        Assert.assertEquals(
            Set.of("DEFAULT", "REMOTE"),
            new CalciteFederationSqlCompiler(List.of(), true)
                .discoverBindings(nonRecursive)
        );

        SqlCompileRequest quoted = SqlCompileRequest.of(
            "WITH \"x\" AS (SELECT * FROM REMOTE.APP.T) SELECT * FROM \"X\"",
            scope
        );
        org.apache.calcite.sql.SqlWith parsedQuoted;
        try {
            parsedQuoted = (org.apache.calcite.sql.SqlWith)
                org.apache.calcite.sql.parser.SqlParser.create(
                    quoted.sql(),
                    org.apache.calcite.sql.parser.SqlParser.config()
                        .withLex(org.apache.calcite.config.Lex.ORACLE)
                ).parseQuery();
        } catch (org.apache.calcite.sql.parser.SqlParseException exception) {
            throw new AssertionError(exception);
        }
        SqlIdentifier declared = ((org.apache.calcite.sql.SqlWithItem)
            parsedQuoted.withList.get(0)).name;
        SqlIdentifier referenced = (SqlIdentifier)
            ((org.apache.calcite.sql.SqlSelect) parsedQuoted.body).getFrom();
        Assert.assertEquals("x", declared.getSimple());
        Assert.assertEquals("X", referenced.getSimple());
        Assert.assertEquals(
            Set.of("DEFAULT", "REMOTE"),
            new CalciteFederationSqlCompiler(List.of(), true).discoverBindings(quoted)
        );
    }

    /**
     * 验证短逻辑表名按映射选择候选 Binding，避免错误加载默认物理源。
     */
    @Test
    public void shouldDiscoverBindingsFromLogicalTableMappings() {
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "logical-binding-scope",
            1,
            Map.of(
                "MYSQL_1", FederationSourceBindingDefinition.of(new SourceId("mysql"), 1),
                "PG_1", FederationSourceBindingDefinition.of(new SourceId("postgres"), 1)
            ),
            "MYSQL_1",
            List.of(
                FederationLogicalTableDefinition.of(
                    "outlet", "MYSQL_1", "MAIN", "physical_outlet"
                ),
                FederationLogicalTableDefinition.of(
                    "outlet_region", "PG_1", "PUBLIC", "physical_region"
                )
            ),
            FederationExecutionPolicy.basic()
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM outlet o JOIN outlet_region r ON o.ID = r.ID",
            scope
        );

        Assert.assertEquals(
            Set.of("MYSQL_1", "PG_1"),
            new CalciteFederationSqlCompiler(List.of(), true).discoverBindings(request)
        );
    }

    /**
     * 验证 Calcite Advisor 只暴露查询范围逻辑表，并支持短名、三段名与字段补全。
     */
    @Test
    public void shouldCompleteLogicalTablesAndColumnsWithoutLeakingPhysicalNames() {
        SourceId sourceId = new SourceId("source");
        SchemaPlus rootSchema = CalciteSchema.createRootSchema(true, false).plus();
        SchemaPlus sourceRoot = rootSchema.add(sourceId.value(), new AbstractSchema());
        SchemaPlus defaultSchema = sourceRoot.add("main", new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of("physical_orders", table("customer_id"));
            }
        });
        FederationSourceDefinition definition = new FederationSourceDefinition(
            sourceId,
            1,
            "test-adapter",
            List.of(new ExternalSchemaDefinition("main", "test", 1)),
            Map.of()
        );
        SourceRuntime runtime = runtime(definition, rootSchema, defaultSchema);
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "completion-scope",
            1,
            Map.of(
                "sales",
                FederationSourceBindingDefinition.of(sourceId, 1, Map.of("main", "main"))
            ),
            "sales",
            List.of(FederationLogicalTableDefinition.of(
                "orders", "sales", "main", "physical_orders"
            )),
            FederationExecutionPolicy.basic()
        );

        try (FederationQueryScopeSnapshot snapshot =
                 FederationQueryScopeSnapshot.borrowedSingle(scope, runtime)) {
            CalciteSqlCompleter completer = new CalciteSqlCompleter();
            SqlCompletionResult shortName = completer.complete(
                request(scope, "SELECT * FROM ord"), snapshot
            );
            SqlCompletionResult qualified = completer.complete(
                request(scope, "SELECT * FROM sales.main.o"), snapshot
            );
            String columnSql = "SELECT orders. FROM orders";
            SqlCompletionResult column = completer.complete(
                new SqlCompletionRequest(scope, columnSql, columnSql.indexOf('.') + 1),
                snapshot
            );
            SqlCompletionResult keyword = completer.complete(
                request(scope, "SEL"), snapshot
            );
            SqlCompletionResult function = completer.complete(
                request(scope, "SELECT COU"), snapshot
            );

            Assert.assertTrue(has(shortName, "orders", SqlCompletionKind.TABLE));
            Assert.assertTrue(qualified.toString(), has(qualified, "orders", SqlCompletionKind.TABLE));
            Assert.assertTrue(column.toString(), has(column, "customer_id", SqlCompletionKind.COLUMN));
            Assert.assertTrue(keyword.toString(), has(keyword, "SELECT", SqlCompletionKind.KEYWORD));
            Assert.assertTrue(function.toString(), has(function, "COUNT", SqlCompletionKind.FUNCTION));
            Assert.assertFalse(shortName.items().stream()
                .map(SqlCompletionItem::label)
                .anyMatch("physical_orders"::equalsIgnoreCase));
        }
    }

    /**
     * 验证冻结统计进入 Calcite 元数据与分片成本，并让等值 INNER JOIN 选择较小构建侧。
     */
    @Test
    public void shouldUseFrozenStatisticsForFragmentCostsAndJoinBuildSide() {
        SourceId smallSource = new SourceId("small-source");
        SourceId largeSource = new SourceId("large-source");
        SourceRuntime smallRuntime = runtime(
            definition(smallSource, "small_table"),
            rootSchema(smallSource, "small_table"),
            null
        );
        SourceRuntime largeRuntime = runtime(
            definition(largeSource, "large_table"),
            rootSchema(largeSource, "large_table"),
            null
        );
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "statistics-scope",
            1,
            Map.of(
                "SMALL_DB", FederationSourceBindingDefinition.of(
                    smallSource, 1, Map.of("main", "main")
                ),
                "LARGE_DB", FederationSourceBindingDefinition.of(
                    largeSource, 1, Map.of("main", "main")
                )
            ),
            "LARGE_DB",
            FederationExecutionPolicy.basic()
        );
        Instant collectedAt = Instant.parse("2026-08-23T00:00:00Z");
        FederationStatisticsSnapshot frozen = new FederationStatisticsSnapshot(
            "statistics-v1",
            Map.of(
                new FederationStatisticsSnapshot.TableKey(
                    smallSource, "main", "small_table"
                ),
                statistics(10, 32, 10, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    largeSource, "main", "large_table"
                ),
                statistics(10_000, 64, 8_000, collectedAt)
            )
        );
        FederationTableStatisticsProvider provider = () -> frozen;
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(
            List.of(),
            true,
            FederationExecutionPolicy.basic(),
            provider
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM SMALL_DB.main.small_table s "
                + "JOIN LARGE_DB.main.large_table l ON s.id = l.id",
            scope
        );

        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.borrowed(
            scope,
            Map.of("SMALL_DB", smallRuntime, "LARGE_DB", largeRuntime)
        )) {
            FederationSqlPlan plan = compiler.compile(request, snapshot, frozen);

            Assert.assertEquals(FederationQueryMode.FEDERATED, plan.queryMode());
            Assert.assertEquals(2, plan.fragments().size());
            Assert.assertTrue(plan.fragments().stream().allMatch(fragment ->
                fragment.costEstimate().statisticsStatus()
                    == FederationStatisticsStatus.COMPLETE
                    && fragment.costEstimate().statisticsSnapshotVersion()
                        .equals("statistics-v1")
            ));
            Assert.assertEquals(1, plan.joinOptimizations().size());
            Assert.assertEquals(
                "SMALL_DB",
                plan.joinOptimizations().get(0).buildBinding()
            );
            Assert.assertEquals(
                320D,
                plan.joinOptimizations().get(0).estimatedBuildBytes(),
                0.001D
            );
            assertFrozenMetadata(plan);
        }
    }

    /**
     * 验证仅含表行数和行宽的 MySQL 级 PARTIAL 统计仍可选择较小构建侧。
     */
    @Test
    public void shouldUsePartialTableStatisticsForJoinBuildSide() {
        SourceId smallSource = new SourceId("partial-small-source");
        SourceId largeSource = new SourceId("complete-large-source");
        SourceRuntime smallRuntime = runtime(
            definition(smallSource, "small_table"),
            rootSchema(smallSource, "small_table"),
            null
        );
        SourceRuntime largeRuntime = runtime(
            definition(largeSource, "large_table"),
            rootSchema(largeSource, "large_table"),
            null
        );
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "partial-statistics-scope",
            1,
            Map.of(
                "SMALL_DB", FederationSourceBindingDefinition.of(
                    smallSource, 1, Map.of("main", "main")
                ),
                "LARGE_DB", FederationSourceBindingDefinition.of(
                    largeSource, 1, Map.of("main", "main")
                )
            ),
            "LARGE_DB",
            FederationExecutionPolicy.basic()
        );
        Instant collectedAt = Instant.parse("2026-08-23T00:00:00Z");
        FederationStatisticsSnapshot frozen = new FederationStatisticsSnapshot(
            "partial-statistics-v1",
            Map.of(
                new FederationStatisticsSnapshot.TableKey(
                    smallSource, "main", "small_table"
                ),
                partialStatistics(10, 32, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    largeSource, "main", "large_table"
                ),
                statistics(10_000, 64, 8_000, collectedAt)
            )
        );
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(
            List.of(),
            true,
            FederationExecutionPolicy.basic(),
            () -> frozen
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM SMALL_DB.main.small_table s "
                + "JOIN LARGE_DB.main.large_table l ON s.id = l.id",
            scope
        );

        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.borrowed(
            scope,
            Map.of("SMALL_DB", smallRuntime, "LARGE_DB", largeRuntime)
        )) {
            FederationSqlPlan plan = compiler.compile(request, snapshot, frozen);

            Assert.assertTrue(plan.fragments().stream().anyMatch(fragment ->
                fragment.bindingName().equals("SMALL_DB")
                    && fragment.costEstimate().statisticsStatus()
                        == FederationStatisticsStatus.PARTIAL
                    && !fragment.costEstimate().statisticsMissing()
            ));
            Assert.assertEquals("SMALL_DB", plan.joinOptimizations().get(0).buildBinding());
            Assert.assertEquals(
                FederationJoinSelectionReason.SMALLER_BUILD_SIDE,
                plan.joinOptimizations().get(0).reason()
            );
            Assert.assertEquals(
                320D,
                plan.joinOptimizations().get(0).estimatedBuildBytes(),
                0.001D
            );
        }
    }

    /**
     * 验证三数据源 Join 生成两个有序阶段，并保留每阶段完整输入集合。
     */
    @Test
    public void shouldDescribeEveryJoinStageForThreeSources() {
        FederationExecutionPolicy threeSourcePolicy = new FederationExecutionPolicy(
            3,
            8,
            2,
            100_000,
            64L * 1024L * 1024L,
            60_000
        );
        SourceId firstSource = new SourceId("first-source");
        SourceId secondSource = new SourceId("second-source");
        SourceId thirdSource = new SourceId("third-source");
        SourceRuntime firstRuntime = runtime(
            definition(firstSource, "first_table"),
            rootSchema(firstSource, "first_table"),
            null
        );
        SourceRuntime secondRuntime = runtime(
            definition(secondSource, "second_table"),
            rootSchema(secondSource, "second_table"),
            null
        );
        SourceRuntime thirdRuntime = runtime(
            definition(thirdSource, "third_table"),
            rootSchema(thirdSource, "third_table"),
            null
        );
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "three-source-statistics-scope",
            1,
            Map.of(
                "FIRST_DB", FederationSourceBindingDefinition.of(
                    firstSource, 1, Map.of("main", "main")
                ),
                "SECOND_DB", FederationSourceBindingDefinition.of(
                    secondSource, 1, Map.of("main", "main")
                ),
                "THIRD_DB", FederationSourceBindingDefinition.of(
                    thirdSource, 1, Map.of("main", "main")
                )
            ),
            "FIRST_DB",
            threeSourcePolicy
        );
        Instant collectedAt = Instant.parse("2026-08-23T00:00:00Z");
        FederationStatisticsSnapshot frozen = new FederationStatisticsSnapshot(
            "three-source-v1",
            Map.of(
                new FederationStatisticsSnapshot.TableKey(
                    firstSource, "main", "first_table"
                ),
                statistics(10_000, 64, 8_000, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    secondSource, "main", "second_table"
                ),
                statistics(100, 32, 100, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    thirdSource, "main", "third_table"
                ),
                statistics(10, 24, 10, collectedAt)
            )
        );
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(
            List.of(),
            true,
            threeSourcePolicy,
            () -> frozen
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM FIRST_DB.main.first_table a "
                + "JOIN SECOND_DB.main.second_table b ON a.id = b.id "
                + "JOIN THIRD_DB.main.third_table c ON b.id = c.id",
            scope
        );

        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.borrowed(
            scope,
            Map.of(
                "FIRST_DB", firstRuntime,
                "SECOND_DB", secondRuntime,
                "THIRD_DB", thirdRuntime
            )
        )) {
            FederationSqlPlan plan = compiler.compile(request, snapshot, frozen);

            Assert.assertEquals(FederationQueryMode.FEDERATED, plan.queryMode());
            assertNoPlannerPlaceholder(plan);
            Assert.assertEquals(3, plan.fragments().size());
            Assert.assertEquals(2, plan.joinOptimizations().size());
            Assert.assertEquals(1, plan.joinOptimizations().get(0).stageIndex());
            Assert.assertEquals(2, plan.joinOptimizations().get(1).stageIndex());
            var firstStage = plan.joinOptimizations().get(0);
            java.util.LinkedHashSet<String> firstInputs = new java.util.LinkedHashSet<>(
                firstStage.leftBindings()
            );
            firstInputs.addAll(firstStage.rightBindings());
            Assert.assertEquals(Set.of("SECOND_DB", "THIRD_DB"), firstInputs);
            var finalStage = plan.joinOptimizations().get(1);
            java.util.LinkedHashSet<String> finalInputs = new java.util.LinkedHashSet<>(
                finalStage.leftBindings()
            );
            finalInputs.addAll(finalStage.rightBindings());
            Assert.assertEquals(Set.of("FIRST_DB", "SECOND_DB", "THIRD_DB"), finalInputs);
            Assert.assertTrue(
                finalStage.leftBindings().size() > 1 || finalStage.rightBindings().size() > 1
            );
            Assert.assertTrue(plan.joinOptimizations().stream().allMatch(optimization ->
                optimization.estimatedBuildBytes() >= 0D
                    && Double.isFinite(optimization.estimatedBuildBytes())
            ));
        }
    }

    /**
     * 验证中间结果行数更少但字节更大的候选不会击败低搬运成本 Join 顺序。
     */
    @Test
    public void shouldPreferLowerTransferBytesOverLowerIntermediateRowCount() {
        FederationExecutionPolicy threeSourcePolicy = new FederationExecutionPolicy(
            3,
            8,
            2,
            100_000,
            64L * 1024L * 1024L,
            60_000
        );
        SourceId wideSource = new SourceId("wide-source");
        SourceId bridgeSource = new SourceId("bridge-source");
        SourceId narrowSource = new SourceId("narrow-source");
        SourceRuntime wideRuntime = runtime(
            definition(wideSource, "wide_table"),
            rootSchema(wideSource, "wide_table"),
            null
        );
        SourceRuntime bridgeRuntime = runtime(
            definition(bridgeSource, "bridge_table"),
            rootSchema(bridgeSource, "bridge_table"),
            null
        );
        SourceRuntime narrowRuntime = runtime(
            definition(narrowSource, "narrow_table"),
            rootSchema(narrowSource, "narrow_table"),
            null
        );
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "transfer-cost-scope",
            1,
            Map.of(
                "WIDE_DB", FederationSourceBindingDefinition.of(
                    wideSource, 1, Map.of("main", "main")
                ),
                "BRIDGE_DB", FederationSourceBindingDefinition.of(
                    bridgeSource, 1, Map.of("main", "main")
                ),
                "NARROW_DB", FederationSourceBindingDefinition.of(
                    narrowSource, 1, Map.of("main", "main")
                )
            ),
            "WIDE_DB",
            threeSourcePolicy
        );
        Instant collectedAt = Instant.parse("2026-08-23T00:00:00Z");
        FederationStatisticsSnapshot frozen = new FederationStatisticsSnapshot(
            "transfer-cost-v1",
            Map.of(
                new FederationStatisticsSnapshot.TableKey(
                    wideSource, "main", "wide_table"
                ),
                singleColumnStatistics(1, 1_000_000, 1, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    bridgeSource, "main", "bridge_table"
                ),
                singleColumnStatistics(100, 16, 100, collectedAt),
                new FederationStatisticsSnapshot.TableKey(
                    narrowSource, "main", "narrow_table"
                ),
                singleColumnStatistics(10_000, 16, 10_000, collectedAt)
            )
        );
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(
            List.of(),
            true,
            threeSourcePolicy,
            () -> frozen
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM WIDE_DB.main.wide_table a "
                + "JOIN BRIDGE_DB.main.bridge_table b ON a.id = b.id "
                + "JOIN NARROW_DB.main.narrow_table c ON b.id = c.id",
            scope
        );

        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.borrowed(
            scope,
            Map.of(
                "WIDE_DB", wideRuntime,
                "BRIDGE_DB", bridgeRuntime,
                "NARROW_DB", narrowRuntime
            )
        )) {
            FederationSqlPlan plan = compiler.compile(request, snapshot, frozen);

            Assert.assertEquals(2, plan.joinOptimizations().size());
            FederationJoinOptimization firstStage = plan.joinOptimizations().get(0);
            java.util.LinkedHashSet<String> firstInputs = new java.util.LinkedHashSet<>(
                firstStage.leftBindings()
            );
            firstInputs.addAll(firstStage.rightBindings());
            Assert.assertEquals(Set.of("BRIDGE_DB", "NARROW_DB"), firstInputs);
            Assert.assertTrue(firstStage.estimatedBuildBytes() < 10_000D);
        }
    }

    /**
     * 断言优化器内部占位节点已经完全转换为可执行关系节点。
     *
     * @param plan 编译计划
     */
    private static void assertNoPlannerPlaceholder(FederationSqlPlan plan) {
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                Assert.assertFalse(node instanceof HyperGraph);
                super.visit(node, ordinal, parent);
            }
        }.go(plan.relRoot().rel);
    }

    /**
     * 创建包含表级和列级信息的完整统计。
     *
     * @param rows 表行数
     * @param rowWidth 平均行宽
     * @param distinctIds ID 基数
     * @param collectedAt 采集时间
     * @return 完整统计快照
     */
    private static FederationTableStatistics statistics(
        double rows,
        long rowWidth,
        double distinctIds,
        Instant collectedAt
    ) {
        return new FederationTableStatistics(
            rows,
            rowWidth,
            collectedAt,
            "catalog",
            Map.of("id", new FederationColumnStatistics(distinctIds, 0.1D, 8)),
            List.of(List.of("id")),
            Instant.MAX,
            FederationStatisticsStatus.COMPLETE
        );
    }

    /**
     * 创建仅包含表级行数和平均行宽的部分统计。
     *
     * @param rows 表行数
     * @param rowWidth 平均行宽
     * @param collectedAt 采集时间
     * @return 部分统计快照
     */
    private static FederationTableStatistics partialStatistics(
        double rows,
        long rowWidth,
        Instant collectedAt
    ) {
        return new FederationTableStatistics(
            rows,
            rowWidth,
            collectedAt,
            "catalog",
            Map.of(),
            List.of(),
            Instant.MAX,
            FederationStatisticsStatus.PARTIAL
        );
    }

    /**
     * 创建单列测试表统计，使列宽与整行宽度一致以判别字节成本。
     *
     * @param rows 表行数
     * @param rowWidth 平均行宽及唯一列宽
     * @param distinctIds ID 基数
     * @param collectedAt 采集时间
     * @return 完整统计快照
     */
    private static FederationTableStatistics singleColumnStatistics(
        double rows,
        long rowWidth,
        double distinctIds,
        Instant collectedAt
    ) {
        return new FederationTableStatistics(
            rows,
            rowWidth,
            collectedAt,
            "catalog",
            Map.of("id", new FederationColumnStatistics(distinctIds, 0.1D, rowWidth)),
            List.of(List.of("id")),
            Instant.MAX,
            FederationStatisticsStatus.COMPLETE
        );
    }

    /**
     * 断言冻结统计可由 Calcite 元数据查询读取。
     *
     * @param plan 编译计划
     */
    private static void assertFrozenMetadata(FederationSqlPlan plan) {
        List<FederationStatisticsTableScan> scans = new java.util.ArrayList<>();
        new RelVisitor() {
            @Override
            public void visit(RelNode node, int ordinal, RelNode parent) {
                if (node instanceof FederationStatisticsTableScan scan) {
                    scans.add(scan);
                }
                super.visit(node, ordinal, parent);
            }
        }.go(plan.relRoot().rel);
        Assert.assertEquals(2, scans.size());

        FederationStatisticsTableScan smallScan = scans.stream()
            .filter(scan -> scan.statistics().estimatedRows() == 10D)
            .findFirst()
            .orElseThrow();
        RelMetadataQuery metadataQuery = smallScan.getCluster().getMetadataQuery();
        Assert.assertEquals(10D, metadataQuery.getRowCount(smallScan), 0.001D);
        Assert.assertEquals(32D, metadataQuery.getAverageRowSize(smallScan), 0.001D);
        Assert.assertEquals(
            Double.valueOf(8D),
            metadataQuery.getAverageColumnSizes(smallScan).get(0)
        );
        Assert.assertEquals(
            10D,
            metadataQuery.getDistinctRowCount(smallScan, ImmutableBitSet.of(0), null),
            0.001D
        );
        Assert.assertTrue(metadataQuery.getUniqueKeys(smallScan).contains(ImmutableBitSet.of(0)));
        Assert.assertEquals(
            0.09D,
            metadataQuery.getSelectivity(
                smallScan,
                smallScan.getCluster().getRexBuilder().makeCall(
                    SqlStdOperatorTable.EQUALS,
                    smallScan.getCluster().getRexBuilder().makeInputRef(smallScan, 0),
                    smallScan.getCluster().getRexBuilder().makeExactLiteral(
                        java.math.BigDecimal.ONE
                    )
                )
            ),
            0.001D
        );
        Assert.assertEquals(
            1D,
            metadataQuery.getSelectivity(
                smallScan,
                smallScan.getCluster().getRexBuilder().makeCall(
                    SqlStdOperatorTable.IS_NOT_DISTINCT_FROM,
                    smallScan.getCluster().getRexBuilder().makeInputRef(smallScan, 0),
                    smallScan.getCluster().getRexBuilder().makeInputRef(smallScan, 0)
                )
            ),
            0.001D
        );
        var nullLiteral = smallScan.getCluster().getRexBuilder().makeNullLiteral(
            smallScan.getRowType().getFieldList().get(0).getType()
        );
        var inputReference = smallScan.getCluster().getRexBuilder().makeInputRef(smallScan, 0);
        Assert.assertEquals(
            0.1D,
            metadataQuery.getSelectivity(
                smallScan,
                smallScan.getCluster().getRexBuilder().makeCall(
                    SqlStdOperatorTable.IS_NOT_DISTINCT_FROM,
                    inputReference,
                    nullLiteral
                )
            ),
            0.001D
        );
        Assert.assertEquals(
            0.1D,
            metadataQuery.getSelectivity(
                smallScan,
                smallScan.getCluster().getRexBuilder().makeCall(
                    SqlStdOperatorTable.IS_NOT_DISTINCT_FROM,
                    nullLiteral,
                    inputReference
                )
            ),
            0.001D
        );
    }

    /**
     * 验证同一编译只使用快照捕获时刻判断有效期，过期统计不会驱动 Join 交换。
     */
    @Test
    public void shouldKeepJoinOrderWhenStatisticsAreStaleAtSnapshotTime() {
        SourceId smallSource = new SourceId("stale-small-source");
        SourceId largeSource = new SourceId("stale-large-source");
        SourceRuntime smallRuntime = runtime(
            definition(smallSource, "small_table"),
            rootSchema(smallSource, "small_table"),
            null
        );
        SourceRuntime largeRuntime = runtime(
            definition(largeSource, "large_table"),
            rootSchema(largeSource, "large_table"),
            null
        );
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "stale-statistics-scope",
            1,
            Map.of(
                "SMALL_DB", FederationSourceBindingDefinition.of(
                    smallSource, 1, Map.of("main", "main")
                ),
                "LARGE_DB", FederationSourceBindingDefinition.of(
                    largeSource, 1, Map.of("main", "main")
                )
            ),
            "SMALL_DB",
            FederationExecutionPolicy.basic()
        );
        Instant capturedAt = Instant.parse("2026-08-23T12:00:00Z");
        Instant expiresAt = capturedAt.minusSeconds(1);
        FederationStatisticsSnapshot frozen = new FederationStatisticsSnapshot(
            "stale-v1",
            capturedAt,
            Map.of(
                new FederationStatisticsSnapshot.TableKey(
                    smallSource, "main", "small_table"
                ),
                new FederationTableStatistics(
                    10,
                    32,
                    capturedAt.minusSeconds(60),
                    "catalog",
                    Map.of(),
                    List.of(),
                    expiresAt,
                    FederationStatisticsStatus.COMPLETE
                ),
                new FederationStatisticsSnapshot.TableKey(
                    largeSource, "main", "large_table"
                ),
                new FederationTableStatistics(
                    10_000,
                    64,
                    capturedAt.minusSeconds(60),
                    "catalog",
                    Map.of(),
                    List.of(),
                    expiresAt,
                    FederationStatisticsStatus.COMPLETE
                )
            )
        );
        CalciteFederationSqlCompiler compiler = new CalciteFederationSqlCompiler(
            List.of(),
            true,
            FederationExecutionPolicy.basic(),
            () -> frozen
        );
        SqlCompileRequest request = SqlCompileRequest.of(
            "SELECT * FROM SMALL_DB.main.small_table s "
                + "JOIN LARGE_DB.main.large_table l ON s.id = l.id",
            scope
        );

        try (FederationQueryScopeSnapshot snapshot = FederationQueryScopeSnapshot.borrowed(
            scope,
            Map.of("SMALL_DB", smallRuntime, "LARGE_DB", largeRuntime)
        )) {
            FederationSqlPlan plan = compiler.compile(request, snapshot, frozen);

            Assert.assertTrue(plan.fragments().stream().allMatch(fragment ->
                fragment.costEstimate().statisticsStatus()
                    == FederationStatisticsStatus.STALE
            ));
            Assert.assertEquals(1, plan.joinOptimizations().size());
            Assert.assertEquals(
                "LARGE_DB",
                plan.joinOptimizations().get(0).buildBinding()
            );
        }
    }

    private static SqlCompletionRequest request(
        FederationQueryScopeDefinition scope,
        String sql
    ) {
        return new SqlCompletionRequest(scope, sql, sql.length());
    }

    private static boolean has(
        SqlCompletionResult result,
        String label,
        SqlCompletionKind kind
    ) {
        return result.items().stream().anyMatch(item ->
            item.kind() == kind && item.label().equalsIgnoreCase(label)
        );
    }

    private SourceRuntime runtime(
        FederationSourceDefinition definition,
        SchemaPlus rootSchema,
        SchemaPlus defaultSchema
    ) {
        SqlDialect dialect = new SqlDialect(SqlDialect.EMPTY_CONTEXT
            .withDatabaseProduct(DatabaseProduct.CALCITE)
            .withQuotedCasing(Casing.UNCHANGED)
            .withUnquotedCasing(Casing.UNCHANGED)
            .withCaseSensitive(false));
        DataSource dataSource = (DataSource) Proxy.newProxyInstance(
            getClass().getClassLoader(),
            new Class<?>[] {DataSource.class},
            (proxy, method, arguments) -> {
                throw new UnsupportedOperationException(method.getName());
            }
        );
        SchemaPlus resolvedDefaultSchema = defaultSchema == null
            ? rootSchema.getSubSchema(definition.sourceId().value()).getSubSchema("main")
            : defaultSchema;
        return new SourceRuntime(
            definition,
            FederationDataSourceHandles.shared(
                dataSource,
                new RuntimeFingerprint("test", "1", "test", "1", "1")
            ),
            new TestAdapter(dialect),
            dialect,
            rootSchema,
            resolvedDefaultSchema,
            new AdapterCompatibility(
                AdapterCompatibilityStatus.VERIFIED,
                "test",
                "1",
                "test",
                "1",
                "test"
            ),
            ignored -> { }
        );
    }

    private static FederationSourceDefinition definition(
        SourceId sourceId,
        String tableName
    ) {
        return new FederationSourceDefinition(
            sourceId,
            1,
            "test-adapter",
            List.of(new ExternalSchemaDefinition("main", "test", 1)),
            Map.of("table", tableName)
        );
    }

    private static SchemaPlus rootSchema(SourceId sourceId, String tableName) {
        SchemaPlus root = CalciteSchema.createRootSchema(true, false).plus();
        SchemaPlus sourceRoot = root.add(sourceId.value(), new AbstractSchema());
        sourceRoot.add("main", new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of(tableName, table("id"));
            }
        });
        return root;
    }

    private static Schema lowerCaseSchema() {
        return new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of("orders", table("id"));
            }
        };
    }

    private static Table table(String columnName) {
        return new AbstractTable() {
            @Override
            public RelDataType getRowType(RelDataTypeFactory typeFactory) {
                return typeFactory.builder()
                    .add(columnName, SqlTypeName.INTEGER)
                    .build();
            }
        };
    }

    private static final class TestAdapter implements FederationSqlAdapterProvider {

        private final SqlDialect dialect;

        private TestAdapter(SqlDialect dialect) {
            this.dialect = dialect;
        }

        @Override
        public String adapterId() {
            return "test-adapter";
        }

        @Override
        public boolean supports(DatabaseMetaData metadata, AdapterHints hints) {
            return true;
        }

        @Override
        public AdapterCompatibility compatibility(DatabaseMetaData metadata, AdapterHints hints) {
            throw new UnsupportedOperationException();
        }

        @Override
        public Schema createSchema(AdapterSchemaContext context) {
            return lowerCaseSchema();
        }

        @Override
        public SqlDialect createDialect(AdapterDialectContext context) {
            return dialect;
        }

        @Override
        public FederationFragmentExecutor fragmentExecutor() {
            throw new UnsupportedOperationException();
        }
    }
}
