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.adapter.FederationSqlAdapterRegistry;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.api.SqlExecutionContext;
import com.easyagents.federation.sql.compile.FederationSqlPlan;
import com.easyagents.federation.sql.compile.SqlCompileRequest;
import com.easyagents.federation.sql.compile.SqlExplainRequest;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.execute.FederationQueryAdmissionController;
import com.easyagents.federation.sql.execute.FederationQueryPermit;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.LocalFederationQueryAdmissionController;
import com.easyagents.federation.sql.execute.QueryAdmissionRequest;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
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.FederationSourceState;
import com.easyagents.federation.sql.source.FederationSourceStateProvider;
import com.easyagents.federation.sql.source.PreparedSourceRuntime;
import com.easyagents.federation.sql.source.RuntimeFingerprint;
import com.easyagents.federation.sql.source.SourceApplyOptions;
import com.easyagents.federation.sql.source.SourceApplyStatus;
import com.easyagents.federation.sql.source.SourceId;
import com.easyagents.federation.sql.source.SourceStateSubscription;
import com.easyagents.federation.sql.source.SourceTombstone;
import java.io.PrintWriter;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.logging.Logger;
import javax.sql.DataSource;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.sql.SqlDialect;
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
import org.apache.calcite.sql.type.SqlTypeName;
import org.junit.Assert;
import org.junit.Test;

/**
 * 数据源 revision、single-flight 和 Runtime lease 生命周期测试。
 */
public class DefaultFederationSourceManagerTest {

    /**
     * 验证 Adapter Schema 不缓存物理元数据，确保 scope-only 接入新表后冷编译可见。
     */
    @Test
    public void shouldExposeNewTablesWithoutRebuildingRuntime() {
        Map<String, Table> tables = new HashMap<>();
        Schema mutableSchema = new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return tables;
            }
        };
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.shared(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1")
            ),
            new FederationSqlAdapterRegistry(
                List.of(new FakeAdapter(mutableSchema)),
                getClass().getClassLoader()
            ),
            FederationSourceStateProvider.none()
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());

        try (SourceRuntime.RuntimeLease lease = manager.acquireRuntime(new SourceId("orders"), 1)) {
            Assert.assertFalse(lease.runtime().defaultSchema().isCacheEnabled());
            Assert.assertNull(lease.runtime().defaultSchema().getTable("new_orders"));
            tables.put("new_orders", integerTable());
            Assert.assertNotNull(lease.runtime().defaultSchema().getTable("new_orders"));
        } finally {
            manager.close();
        }
    }

    /**
     * 创建仅含整数主键的测试表。
     *
     * @return Calcite 测试表
     */
    private static Table integerTable() {
        return new org.apache.calcite.schema.impl.AbstractTable() {
            @Override
            public RelDataType getRowType(RelDataTypeFactory typeFactory) {
                return typeFactory.builder().add("ID", SqlTypeName.INTEGER).build();
            }
        };
    }

    /**
     * 验证并发预热只解析一次 Handle，并覆盖版本冲突、切换和墓碑语义。
     *
     * @throws Exception 并发测试失败
     */
    @Test
    public void shouldInitializeOnceAndDrainOldRuntime() throws Exception {
        AtomicInteger resolves = new AtomicInteger();
        AtomicInteger closes = new AtomicInteger();
        DataSource dataSource = fakeDataSource();
        FederationSqlAdapterRegistry adapters = new FederationSqlAdapterRegistry(
            List.of(new FakeAdapter()),
            getClass().getClassLoader()
        );
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> {
                resolves.incrementAndGet();
                return FederationDataSourceHandles.owned(
                    dataSource,
                    new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1"),
                    closes::incrementAndGet
                );
            },
            adapters,
            FederationSourceStateProvider.none()
        );

        FederationSourceDefinition revisionOne = definition(1, Map.of());
        Assert.assertEquals(SourceApplyStatus.APPLIED, manager.apply(revisionOne).status());
        Assert.assertEquals(0, resolves.get());

        int concurrency = 32;
        ExecutorService executor = Executors.newFixedThreadPool(8);
        CountDownLatch start = new CountDownLatch(1);
        CountDownLatch done = new CountDownLatch(concurrency);
        for (int index = 0; index < concurrency; index++) {
            executor.execute(() -> {
                try {
                    start.await();
                    manager.ensureReady(new SourceId("orders"), 1);
                } catch (InterruptedException exception) {
                    Thread.currentThread().interrupt();
                } finally {
                    done.countDown();
                }
            });
        }
        start.countDown();
        Assert.assertTrue(done.await(5, TimeUnit.SECONDS));
        executor.shutdownNow();
        Assert.assertEquals(1, resolves.get());

        Assert.assertEquals(SourceApplyStatus.IDEMPOTENT, manager.apply(revisionOne).status());
        Assert.assertEquals(
            SourceApplyStatus.CONFLICT,
            manager.apply(definition(1, Map.of("changed", "true"))).status()
        );
        Assert.assertEquals(SourceApplyStatus.IGNORED_STALE, manager.apply(definition(0, Map.of())).status());

        SourceRuntime.RuntimeLease oldLease = manager.acquireRuntime(new SourceId("orders"), 1);
        Assert.assertEquals(
            SourceApplyStatus.APPLIED,
            manager.apply(definition(2, Map.of()), SourceApplyOptions.prewarmNow()).status()
        );
        Assert.assertEquals(2, resolves.get());
        Assert.assertEquals(0, closes.get());
        oldLease.close();
        Assert.assertEquals(1, closes.get());

        Assert.assertEquals(
            SourceApplyStatus.APPLIED,
            manager.remove(SourceTombstone.of(new SourceId("orders"), 3)).result().status()
        );
        Assert.assertEquals(2, closes.get());
        Assert.assertEquals(SourceApplyStatus.IGNORED_STALE, manager.apply(definition(2, Map.of())).status());
        manager.close();
    }

    /**
     * 验证新 revision 初始化失败时，满足最低 revision 的旧 Runtime 仍可服务查询。
     */
    @Test
    public void shouldKeepLastReadyRuntimeWhenNewRevisionFails() {
        DataSource dataSource = fakeDataSource();
        AtomicInteger closes = new AtomicInteger();
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> {
                if (definition.revision() == 2) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED,
                        "simulated revision two failure"
                    );
                }
                return FederationDataSourceHandles.owned(
                    dataSource,
                    new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1"),
                    closes::incrementAndGet
                );
            },
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            FederationSourceStateProvider.none()
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());
        manager.apply(definition(2, Map.of()));

        Assert.assertEquals(1, manager.ensureReady(new SourceId("orders"), 1).readyRevision());
        Assert.assertEquals(2, manager.view(new SourceId("orders")).orElseThrow().desiredRevision());
        try {
            manager.ensureReady(new SourceId("orders"), 2);
            Assert.fail("minimum revision two should fail");
        } catch (FederationSqlException exception) {
            Assert.assertEquals(FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED, exception.errorCode());
        }
        manager.close();
        Assert.assertEquals(1, closes.get());
    }

    /**
     * 验证预构建 Runtime 在 commit 前不改变共享 Slot，commit 后再原子替换旧版本。
     */
    @Test
    public void shouldPrepareWithoutPublishingAndCommitAtomically() {
        DataSource dataSource = fakeDataSource();
        AtomicInteger closes = new AtomicInteger();
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.owned(
                dataSource,
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1"),
                closes::incrementAndGet
            ),
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            FederationSourceStateProvider.none()
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());

        try (PreparedSourceRuntime prepared = manager.prepare(definition(2, Map.of()))) {
            Assert.assertEquals(1L, manager.view(new SourceId("orders"))
                .orElseThrow().desiredRevision());
            Assert.assertEquals(1L, manager.view(new SourceId("orders"))
                .orElseThrow().readyRevision());

            Assert.assertEquals(SourceApplyStatus.APPLIED, manager.commit(prepared).status());
        }

        Assert.assertEquals(2L, manager.view(new SourceId("orders"))
            .orElseThrow().desiredRevision());
        Assert.assertEquals(2L, manager.view(new SourceId("orders"))
            .orElseThrow().readyRevision());
        Assert.assertEquals(1, closes.get());
        manager.close();
        Assert.assertEquals(2, closes.get());
    }

    /**
     * 验证放弃预构建 Runtime 会关闭候选 Handle，且不改变当前版本。
     */
    @Test
    public void shouldDiscardPreparedRuntimeWithoutChangingSlot() {
        DataSource dataSource = fakeDataSource();
        AtomicInteger closes = new AtomicInteger();
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.owned(
                dataSource,
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1"),
                closes::incrementAndGet
            ),
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            FederationSourceStateProvider.none()
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());

        try (PreparedSourceRuntime ignored = manager.prepare(definition(2, Map.of()))) {
            Assert.assertEquals(0, closes.get());
        }

        Assert.assertEquals(1, closes.get());
        Assert.assertEquals(1L, manager.view(new SourceId("orders"))
            .orElseThrow().desiredRevision());
        Assert.assertEquals(1L, manager.view(new SourceId("orders"))
            .orElseThrow().readyRevision());
        manager.close();
        Assert.assertEquals(2, closes.get());
    }

    /**
     * 验证连接池关闭连续失败后仍保留可重试引用，且成功后保持幂等。
     */
    @Test
    public void shouldRetryRuntimeClosureAfterTransientFailures() {
        AtomicInteger closeAttempts = new AtomicInteger();
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.owned(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1"),
                () -> {
                    if (closeAttempts.incrementAndGet() <= 2) {
                        throw new IllegalStateException("simulated transient close failure");
                    }
                }
            ),
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            FederationSourceStateProvider.none()
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());

        try {
            manager.close();
            Assert.fail("first manager close should report exhausted bounded retries");
        } catch (IllegalStateException expected) {
            Assert.assertEquals(2, closeAttempts.get());
        }

        manager.close();
        Assert.assertEquals(3, closeAttempts.get());
        manager.close();
        Assert.assertEquals(3, closeAttempts.get());
    }

    /**
     * 验证共享状态订阅关闭失败后，重复关闭管理器会继续释放订阅。
     */
    @Test
    public void shouldRetrySubscriptionClosureAfterFailure() {
        AtomicInteger closeAttempts = new AtomicInteger();
        FederationSourceStateProvider provider = new FederationSourceStateProvider() {
            @Override
            public Optional<FederationSourceState> find(SourceId sourceId) {
                return Optional.empty();
            }

            @Override
            public SourceStateSubscription subscribe(
                Consumer<FederationSourceState> consumer
            ) {
                return () -> {
                    if (closeAttempts.incrementAndGet() == 1) {
                        throw new IllegalStateException(
                            "simulated subscription close failure"
                        );
                    }
                };
            }
        };
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.shared(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1")
            ),
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            provider
        );

        try {
            manager.close();
            Assert.fail("first subscription close should fail");
        } catch (IllegalStateException expected) {
            Assert.assertEquals(1, closeAttempts.get());
        }
        manager.close();
        manager.close();
        Assert.assertEquals(2, closeAttempts.get());
    }

    /**
     * 验证注册表代次只跟随启用 SourceId 集合变化，避免无关 revision 冲掉热计划。
     */
    @Test
    public void shouldAdvanceCatalogGenerationOnlyWhenActiveMembershipChanges() {
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.shared(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1")
            ),
            new FederationSqlAdapterRegistry(List.of(new FakeAdapter()), getClass().getClassLoader()),
            FederationSourceStateProvider.none()
        );

        manager.apply(definition(1, Map.of()));
        SourceCatalogSnapshot first = manager.catalogSnapshot();
        Assert.assertEquals(Set.of(new SourceId("orders")), first.sourceIds());

        manager.apply(definition(2, Map.of()));
        SourceCatalogSnapshot revised = manager.catalogSnapshot();
        Assert.assertEquals(first.generation(), revised.generation());

        manager.remove(SourceTombstone.of(new SourceId("orders"), 3));
        SourceCatalogSnapshot removed = manager.catalogSnapshot();
        Assert.assertTrue(removed.sourceIds().isEmpty());
        Assert.assertEquals(first.generation() + 1, removed.generation());
        manager.close();
    }

    /**
     * 验证显式注册重复 AdapterId 会直接报告冲突。
     */
    @Test
    public void shouldRejectDuplicateExplicitAdapterIds() {
        try {
            new FederationSqlAdapterRegistry(
                List.of(new FakeAdapter(), new FakeAdapter()),
                getClass().getClassLoader()
            );
            Assert.fail("duplicate adapter ids should be rejected");
        } catch (FederationSqlException exception) {
            Assert.assertEquals(FederationSqlErrorCode.SOURCE_DEFINITION_CONFLICT, exception.errorCode());
        }
    }

    /**
     * 验证关闭 Engine 会及时取消已登记查询，且不会等待阻塞的 Adapter 执行返回。
     *
     * @throws Exception 并发测试失败
     */
    @Test
    public void shouldCancelBlockingQueryDuringEngineClose() throws Exception {
        CountDownLatch registered = new CountDownLatch(1);
        CountDownLatch cancelled = new CountDownLatch(1);
        AtomicInteger cancelCalls = new AtomicInteger();
        Statement statement = (Statement) Proxy.newProxyInstance(
            getClass().getClassLoader(),
            new Class<?>[] {Statement.class},
            (proxy, method, args) -> {
                if ("cancel".equals(method.getName())) {
                    cancelCalls.incrementAndGet();
                    cancelled.countDown();
                    return null;
                }
                return defaultValue(method.getReturnType());
            }
        );
        FederationFragmentExecutor blockingExecutor = context -> {
            context.statementLifecycle().register(statement);
            registered.countDown();
            try {
                if (!cancelled.await(5, TimeUnit.SECONDS)) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.EXECUTION_FAILED,
                        "test query was not cancelled"
                    );
                }
                throw new FederationSqlException(
                    context.statementLifecycle().cancellationRequested()
                        ? FederationSqlErrorCode.QUERY_CANCELLED
                        : FederationSqlErrorCode.EXECUTION_FAILED,
                    "test query stopped"
                );
            } catch (InterruptedException exception) {
                Thread.currentThread().interrupt();
                throw new FederationSqlException(
                    FederationSqlErrorCode.EXECUTION_FAILED,
                    "test query was interrupted",
                    exception
                );
            } finally {
                context.statementLifecycle().unregister(statement);
            }
        };
        DefaultFederationSourceManager manager = new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.shared(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1")
            ),
            new FederationSqlAdapterRegistry(
                List.of(new FakeAdapter(blockingExecutor)),
                getClass().getClassLoader()
            ),
            FederationSourceStateProvider.none()
        );
        DefaultFederationSqlEngine engine = new DefaultFederationSqlEngine(
            manager,
            new LocalFederationQueryAdmissionController(4),
            List.of(),
            16,
            false
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());
        FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(
            "VALUES (1)",
            new SourceId("orders"),
            1
        ));

        ExecutorService executor = Executors.newFixedThreadPool(2);
        try {
            Future<FederationSqlErrorCode> query = executor.submit(() -> {
                try {
                    engine.execute(plan, SqlExecutionContext.of(List.of()));
                    return null;
                } catch (FederationSqlException exception) {
                    return exception.errorCode();
                }
            });
            Assert.assertTrue(registered.await(2, TimeUnit.SECONDS));
            Future<?> closing = executor.submit(engine::close);
            closing.get(2, TimeUnit.SECONDS);
            Assert.assertEquals(1, cancelCalls.get());
            Assert.assertEquals(FederationSqlErrorCode.QUERY_CANCELLED, query.get(2, TimeUnit.SECONDS));
        } finally {
            executor.shutdownNow();
            engine.close();
        }
    }

    /**
     * 验证活动 JDBC 游标永久阻塞关闭时，Engine 关闭仍保持有界并暴露未完成清理。
     *
     * @throws Exception 并发测试失败
     */
    @Test
    public void shouldCloseEngineWithinBoundWhenCursorCloseBlocks() throws Exception {
        CountDownLatch closeStarted = new CountDownLatch(1);
        CountDownLatch releaseClose = new CountDownLatch(1);
        CountDownLatch closeFinished = new CountDownLatch(1);
        FederationFragmentExecutor blockingCursorExecutor = context -> new FederationResultCursor() {
            @Override
            public QueryId queryId() {
                return context.queryId();
            }

            @Override
            public List<com.easyagents.federation.sql.execute.FederationColumn> columns() {
                return List.of();
            }

            @Override
            public boolean next() {
                return false;
            }

            @Override
            public Object getObject(int columnIndex) {
                return null;
            }

            @Override
            public List<Object> row() {
                return List.of();
            }

            @Override
            public void close() {
                closeStarted.countDown();
                try {
                    while (releaseClose.getCount() > 0L) {
                        try {
                            releaseClose.await();
                        } catch (InterruptedException ignored) {
                            // 模拟忽略线程中断且永久阻塞 close() 的 JDBC Driver。
                        }
                    }
                } finally {
                    closeFinished.countDown();
                }
            }
        };
        Schema tableSchema = new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of("ORDERS", integerTable());
            }
        };
        DefaultFederationSourceManager manager = sourceManager(
            new FakeAdapter(blockingCursorExecutor, tableSchema)
        );
        DefaultFederationSqlEngine engine = new DefaultFederationSqlEngine(
            manager,
            new LocalFederationQueryAdmissionController(4),
            List.of(),
            16,
            false
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());
        FederationSqlPlan plan = engine.compile(SqlCompileRequest.of(
            "SELECT id FROM orders",
            new SourceId("orders"),
            1
        ));
        FederationResultCursor cursor = engine.execute(
            plan,
            SqlExecutionContext.of(List.of())
        );
        ExecutorService executor = Executors.newSingleThreadExecutor();
        try {
            Future<?> closing = executor.submit(engine::close);
            Assert.assertTrue(closeStarted.await(2, TimeUnit.SECONDS));
            closing.get(2, TimeUnit.SECONDS);
            Assert.assertTrue(engine.cleanupMetrics().unresolvedCleanups() > 0L);
        } finally {
            releaseClose.countDown();
            Assert.assertTrue(closeFinished.await(2, TimeUnit.SECONDS));
            cursor.close();
            executor.shutdownNow();
            engine.close();
        }
    }

    /**
     * 验证 Explain 准入等待受统一编译截止时间约束，并返回精确超时错误。
     */
    @Test
    public void shouldApplyUnifiedDeadlineToExplainAdmission() {
        CountDownLatch admissionStarted = new CountDownLatch(1);
        FederationQueryAdmissionController blockingAdmission =
            new FederationQueryAdmissionController() {
                @Override
                public FederationQueryPermit acquire(
                    SourceId sourceId,
                    QueryId queryId,
                    Duration timeout
                ) {
                    throw new AssertionError("query-level admission request is required");
                }

                @Override
                public FederationQueryPermit acquire(QueryAdmissionRequest request) {
                    admissionStarted.countDown();
                    while (!request.cancellationRequested().getAsBoolean()) {
                        try {
                            Thread.sleep(5L);
                        } catch (InterruptedException exception) {
                            Thread.currentThread().interrupt();
                            throw new FederationSqlException(
                                FederationSqlErrorCode.QUERY_ADMISSION_TIMEOUT,
                                "Explain admission was interrupted",
                                exception
                            );
                        }
                    }
                    throw new FederationSqlException(
                        FederationSqlErrorCode.QUERY_ADMISSION_TIMEOUT,
                        "Explain admission exceeded its deadline"
                    );
                }
            };
        Schema tableSchema = new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of("ORDERS", integerTable());
            }
        };
        DefaultFederationSourceManager manager = sourceManager(new FakeAdapter(tableSchema));
        DefaultFederationSqlEngine engine = new DefaultFederationSqlEngine(
            manager,
            blockingAdmission,
            List.of(),
            16,
            4,
            false,
            new FederationExecutionPolicy(
                2,
                8,
                2,
                100_000,
                64L * 1024L * 1024L,
                1_000
            )
        );
        manager.apply(definition(1, Map.of()), SourceApplyOptions.prewarmNow());
        long startedAt = System.nanoTime();
        try {
            engine.explain(new SqlExplainRequest(SqlCompileRequest.of(
                "SELECT id FROM orders",
                new SourceId("orders"),
                1
            )));
            Assert.fail("Explain admission should respect the unified deadline");
        } catch (FederationSqlException exception) {
            Assert.assertEquals(
                FederationSqlErrorCode.SQL_COMPILE_TIMEOUT,
                exception.errorCode()
            );
            Assert.assertEquals(0L, admissionStarted.getCount());
            long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(
                System.nanoTime() - startedAt
            );
            Assert.assertTrue("Explain took " + elapsedMillis + "ms", elapsedMillis < 2_000L);
        } finally {
            engine.close();
        }
    }

    private DefaultFederationSourceManager sourceManager(FakeAdapter adapter) {
        return new DefaultFederationSourceManager(
            definition -> FederationDataSourceHandles.shared(
                fakeDataSource(),
                new RuntimeFingerprint("fake", "1", "fake-driver", "1", "1")
            ),
            new FederationSqlAdapterRegistry(
                List.of(adapter),
                getClass().getClassLoader()
            ),
            FederationSourceStateProvider.none()
        );
    }

    private static FederationSourceDefinition definition(long revision, Map<String, String> options) {
        return new FederationSourceDefinition(
            new SourceId("orders"),
            revision,
            "fake",
            List.of(new ExternalSchemaDefinition("app", "snapshot", revision)),
            options
        );
    }

    private static DataSource fakeDataSource() {
        DatabaseMetaData metadata = (DatabaseMetaData) Proxy.newProxyInstance(
            DefaultFederationSourceManagerTest.class.getClassLoader(),
            new Class<?>[] {DatabaseMetaData.class},
            (proxy, method, args) -> switch (method.getName()) {
                case "getDatabaseProductName" -> "fake";
                case "getDatabaseProductVersion" -> "1";
                case "getDriverName" -> "fake-driver";
                case "getDriverVersion" -> "1";
                default -> defaultValue(method.getReturnType());
            }
        );
        Connection connection = (Connection) Proxy.newProxyInstance(
            DefaultFederationSourceManagerTest.class.getClassLoader(),
            new Class<?>[] {Connection.class},
            (proxy, method, args) -> switch (method.getName()) {
                case "getMetaData" -> metadata;
                case "close" -> null;
                case "isClosed" -> false;
                default -> defaultValue(method.getReturnType());
            }
        );
        return new DataSource() {
            @Override
            public Connection getConnection() {
                return connection;
            }

            @Override
            public Connection getConnection(String username, String password) {
                return connection;
            }

            @Override
            public <T> T unwrap(Class<T> iface) throws SQLException {
                throw new SQLException("not a wrapper");
            }

            @Override
            public boolean isWrapperFor(Class<?> iface) {
                return false;
            }

            @Override
            public PrintWriter getLogWriter() {
                return null;
            }

            @Override
            public void setLogWriter(PrintWriter out) {
            }

            @Override
            public void setLoginTimeout(int seconds) {
            }

            @Override
            public int getLoginTimeout() {
                return 0;
            }

            @Override
            public Logger getParentLogger() {
                return Logger.getGlobal();
            }
        };
    }

    private static Object defaultValue(Class<?> type) {
        if (!type.isPrimitive()) {
            return null;
        }
        if (type == boolean.class) {
            return false;
        }
        if (type == byte.class) {
            return (byte) 0;
        }
        if (type == short.class) {
            return (short) 0;
        }
        if (type == int.class) {
            return 0;
        }
        if (type == long.class) {
            return 0L;
        }
        if (type == float.class) {
            return 0F;
        }
        if (type == double.class) {
            return 0D;
        }
        if (type == char.class) {
            return '\0';
        }
        return null;
    }

    private static final class FakeAdapter implements FederationSqlAdapterProvider {

        private final FederationFragmentExecutor fragmentExecutor;
        private final Schema schema;

        private FakeAdapter() {
            this(context -> {
                throw new UnsupportedOperationException("not required by lifecycle test");
            }, new AbstractSchema());
        }

        private FakeAdapter(Schema schema) {
            this(context -> {
                throw new UnsupportedOperationException("not required by lifecycle test");
            }, schema);
        }

        private FakeAdapter(FederationFragmentExecutor fragmentExecutor) {
            this(fragmentExecutor, new AbstractSchema());
        }

        private FakeAdapter(FederationFragmentExecutor fragmentExecutor, Schema schema) {
            this.fragmentExecutor = fragmentExecutor;
            this.schema = schema;
        }

        @Override
        public String adapterId() {
            return "fake";
        }

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

        @Override
        public AdapterCompatibility compatibility(DatabaseMetaData metadata, AdapterHints hints) {
            return new AdapterCompatibility(
                AdapterCompatibilityStatus.VERIFIED,
                "fake",
                "1",
                "fake-driver",
                "1",
                "test adapter"
            );
        }

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

        @Override
        public SqlDialect createDialect(AdapterDialectContext context) {
            return AnsiSqlDialect.DEFAULT;
        }

        @Override
        public FederationFragmentExecutor fragmentExecutor() {
            return fragmentExecutor;
        }
    }
}
