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.api.SqlExecutionContext;
import com.easyagents.federation.sql.execute.FederationFragmentExecutionContext;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.SqlExecutionOptions;
import com.easyagents.federation.sql.execute.SqlParameter;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationFragmentPlan;
import java.math.BigDecimal;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.calcite.avatica.util.ByteString;
import org.apache.calcite.linq4j.AbstractEnumerable;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.sql.type.SqlTypeName;

/**
 * 一次联邦查询的分片执行、类型桥接、预算和取消上下文。
 */
final class FederationExecutionSession implements AutoCloseable {

    static final String DATA_CONTEXT_KEY = "easyagents.federation.session";

    private final DefaultFederationSqlPlan plan;
    private final SqlExecutionContext context;
    private final FederationQueryScopeSnapshot runtimeSnapshot;
    private final QueryCancellationRegistry.QueryRegistration registration;
    private final FederationExecutionPolicy policy;
    private final FederationQueryMetricsTracker metrics;
    private final QueryDeadline deadline;
    private final FederationFragmentScheduler fragmentScheduler;
    private final Semaphore fragmentSlots;
    private final Set<FederationResultCursor> openFragmentCursors = ConcurrentHashMap.newKeySet();
    private final Set<Future<?>> fragmentTasks = ConcurrentHashMap.newKeySet();
    private final Map<String, CompletableFuture<PreparedFragment>> fragmentPreparations =
        new ConcurrentHashMap<>();
    private final Object preparationLock = new Object();
    private final ArrayDeque<CompiledFederationFragment> pendingPreparations =
        new ArrayDeque<>();
    private final AtomicBoolean closed = new AtomicBoolean();
    private final AtomicReference<Throwable> preparationFailure = new AtomicReference<>();
    private int runningPreparations;
    private boolean preparationFailed;

    /**
     * 创建查询执行会话。
     *
     * @param plan 联邦计划
     * @param context 执行上下文
     * @param runtimeSnapshot 实际引用源 Runtime 快照
     * @param registration 查询取消登记
     * @param policy 有效资源策略
     * @param metrics 指标跟踪器
     * @param deadline 请求级统一截止时间
     * @param fragmentScheduler Engine 级有界分片调度器
     */
    FederationExecutionSession(
        DefaultFederationSqlPlan plan,
        SqlExecutionContext context,
        FederationQueryScopeSnapshot runtimeSnapshot,
        QueryCancellationRegistry.QueryRegistration registration,
        FederationExecutionPolicy policy,
        FederationQueryMetricsTracker metrics,
        QueryDeadline deadline,
        FederationFragmentScheduler fragmentScheduler
    ) {
        this.plan = plan;
        this.context = context;
        this.runtimeSnapshot = runtimeSnapshot;
        this.registration = registration;
        this.policy = policy;
        this.metrics = metrics;
        this.deadline = deadline;
        this.fragmentScheduler = fragmentScheduler;
        this.fragmentSlots = new Semaphore(policy.maximumConcurrentFragments(), true);
    }

    /**
     * 异步启动多源物理分片，在查询级中间结果预算内完成有界物化。
     */
    void prepareFragments() {
        if (plan.fragments().size() < 2 || policy.maximumConcurrentFragments() < 2) {
            return;
        }
        try {
            for (FederationFragmentPlan fragment : plan.fragments()) {
                CompiledFederationFragment compiled = plan.compiledFragments().get(
                    fragment.fragmentId()
                );
                if (compiled == null) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.EXECUTION_FAILED,
                        "compiled fragment is missing: " + fragment.fragmentId()
                    );
                }
                CompletableFuture<PreparedFragment> preparation = new CompletableFuture<>();
                fragmentPreparations.put(fragment.fragmentId(), preparation);
                synchronized (preparationLock) {
                    pendingPreparations.addLast(compiled);
                }
            }
            schedulePreparations();
        } catch (RuntimeException | Error exception) {
            cancelPreparation();
            throw exception;
        }
    }

    /**
     * 为 Calcite ScannableTable 创建一个按需执行物理分片的 Enumerable。
     *
     * @param fragmentId 分片标识
     * @return 分片行 Enumerable
     */
    Enumerable<Object[]> scan(String fragmentId) {
        CompiledFederationFragment compiled = plan.compiledFragments().get(fragmentId);
        if (compiled == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "compiled fragment is missing: " + fragmentId
            );
        }
        return new AbstractEnumerable<>() {
            @Override
            public Enumerator<Object[]> enumerator() {
                CompletableFuture<PreparedFragment> preparation =
                    fragmentPreparations.get(fragmentId);
                if (preparation != null) {
                    PreparedFragment prepared = awaitPrepared(preparation);
                    fragmentPreparations.remove(fragmentId, preparation);
                    return prepared.enumerator();
                }
                return openFragment(compiled);
            }
        };
    }

    /**
     * 返回 Calcite DataContext 使用的查询参数值。
     *
     * @return 参数键值
     */
    Map<String, Object> dataContextValues() {
        Map<String, Object> values = new java.util.LinkedHashMap<>();
        values.put(DATA_CONTEXT_KEY, this);
        for (int index = 0; index < context.parameters().size(); index++) {
            values.put("?" + index, normalizeParameter(context.parameters().get(index)));
        }
        return java.util.Collections.unmodifiableMap(values);
    }

    /**
     * 在本地算子或分片读取前检查取消、关闭和总时限。
     */
    void ensureExecutionAllowed() {
        if (closed.get()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "federation execution session is closed"
            );
        }
        if (registration.cancellationRequested()) {
            if (registration.timeoutRequested()) {
                metrics.markTimedOut();
            } else {
                metrics.markCancelled();
            }
            registration.ensureNotCancelled();
        }
        throwPreparationFailure();
        deadline.ensureAllowed();
    }

    /**
     * 计量一个 Calcite 本地算子输出行并执行统一硬预算检查。
     *
     * @param operatorName 本地算子名称
     * @param value Calcite 内部行值
     * @param elapsedNanos 算子产生当前行的耗时
     */
    void recordLocalIntermediate(String operatorName, Object value, long elapsedNanos) {
        ensureExecutionAllowed();
        Object[] row = value instanceof Object[] values ? values : new Object[] {value};
        FederationQueryMetricsTracker.IntermediateUsage usage =
            metrics.recordLocalIntermediate(operatorName, row, elapsedNanos);
        if (usage.rows() > policy.maximumIntermediateRows()
            || usage.bytes() > policy.maximumIntermediateBytes()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
                "federation local operator " + operatorName
                    + " exceeded the intermediate result limit"
            );
        }
    }

    private Enumerator<Object[]> openFragment(CompiledFederationFragment compiled) {
        ensureExecutionAllowed();
        acquireFragmentSlot();
        try {
            return new FragmentEnumerator(compiled, createFragmentCursor(compiled), true);
        } catch (RuntimeException exception) {
            fragmentSlots.release();
            throw exception;
        }
    }

    private void schedulePreparations() {
        List<PreparationTask> tasks = new ArrayList<>();
        synchronized (preparationLock) {
            while (!closed.get()
                && !preparationFailed
                && runningPreparations < policy.maximumConcurrentFragments()
                && !pendingPreparations.isEmpty()) {
                CompiledFederationFragment compiled = pendingPreparations.removeFirst();
                CompletableFuture<PreparedFragment> preparation =
                    fragmentPreparations.get(
                        compiled.plan().fragmentId()
                    );
                if (preparation == null || preparation.isCancelled()) {
                    continue;
                }
                runningPreparations++;
                tasks.add(new PreparationTask(compiled, preparation));
            }
        }
        for (PreparationTask task : tasks) {
            submitPreparation(task);
        }
    }

    private void submitPreparation(PreparationTask preparationTask) {
        FutureTask<Void> task = new FutureTask<>(() -> {
            runPreparation(preparationTask);
            return null;
        }) {
            @Override
            protected void done() {
                fragmentTasks.remove(this);
            }
        };
        fragmentTasks.add(task);
        try {
            fragmentScheduler.execute(task);
        } catch (RuntimeException | Error exception) {
            fragmentTasks.remove(task);
            task.cancel(true);
            preparationTask.preparation().completeExceptionally(exception);
            failPreparations(exception);
            preparationFinished();
        }
    }

    private void runPreparation(PreparationTask task) {
        try {
            task.preparation().complete(prepareFragment(task.compiled()));
        } catch (RuntimeException | Error exception) {
            task.preparation().completeExceptionally(exception);
            failPreparations(exception);
        } finally {
            preparationFinished();
        }
    }

    private void preparationFinished() {
        boolean scheduleNext;
        synchronized (preparationLock) {
            runningPreparations--;
            scheduleNext = !closed.get() && !preparationFailed;
        }
        if (scheduleNext) {
            schedulePreparations();
        }
    }

    private void failPreparations(Throwable failure) {
        if (!preparationFailure.compareAndSet(null, failure)) {
            return;
        }
        synchronized (preparationLock) {
            preparationFailed = true;
            pendingPreparations.clear();
        }
        for (CompletableFuture<PreparedFragment> preparation : fragmentPreparations.values()) {
            preparation.completeExceptionally(failure);
        }
        registration.terminateStatementsForCleanup();
    }

    private void throwPreparationFailure() {
        Throwable failure = preparationFailure.get();
        if (failure == null) {
            return;
        }
        if (failure instanceof RuntimeException runtimeException) {
            throw runtimeException;
        }
        if (failure instanceof Error error) {
            throw error;
        }
        throw new FederationSqlException(
            FederationSqlErrorCode.EXECUTION_FAILED,
            "federation fragment preparation failed",
            failure
        );
    }

    private PreparedFragment awaitPrepared(Future<PreparedFragment> preparation) {
        try {
            while (true) {
                ensureExecutionAllowed();
                long remaining = deadline.remainingNanos();
                if (remaining <= 0L) {
                    ensureExecutionAllowed();
                }
                try {
                    return preparation.get(
                        Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(50)),
                        TimeUnit.NANOSECONDS
                    );
                } catch (TimeoutException ignored) {
                    // 周期性复核共享 Deadline 和取消状态。
                }
            }
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            cancelPreparation();
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "waiting for federation fragment preparation was interrupted",
                exception
            );
        } catch (ExecutionException exception) {
            cancelPreparation();
            Throwable cause = exception.getCause();
            if (cause instanceof RuntimeException runtimeException) {
                throw runtimeException;
            }
            if (cause instanceof Error error) {
                throw error;
            }
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "federation fragment preparation failed",
                cause
            );
        } catch (RuntimeException | Error exception) {
            cancelPreparation();
            throw exception;
        }
    }

    private PreparedFragment prepareFragment(CompiledFederationFragment compiled) {
        ensureExecutionAllowed();
        FederationResultCursor cursor = null;
        try {
            cursor = createFragmentCursor(compiled);
            List<Object[]> rows = new ArrayList<>();
            while (true) {
                ensureExecutionAllowed();
                if (!cursor.next()) {
                    releaseFragment(compiled, cursor, true, false);
                    cursor = null;
                    return new PreparedFragment(rows);
                }
                Object[] row = normalizeRow(cursor, compiled.rowType());
                recordFragmentIntermediate(compiled, row);
                rows.add(row);
            }
        } catch (RuntimeException | Error exception) {
            if (cursor != null) {
                releaseFragment(compiled, cursor, false, false);
            }
            throw exception;
        }
    }

    private FederationResultCursor createFragmentCursor(
        CompiledFederationFragment compiled
    ) {
        FederationFragmentPlan fragment = compiled.plan();
        SourceRuntime runtime = runtimeSnapshot.runtime(fragment.bindingName());
        FederationResultCursor cursor = runtime.adapter().fragmentExecutor().execute(
            new FederationFragmentExecutionContext(
                context.queryId(),
                fragment.executableSql(),
                remapParameters(fragment, context.parameters()),
                fragmentOptions(context.options()),
                runtime.handle().dataSource(),
                runtime.compatibility(),
                runtime.definition().adapterOptions(),
                registration.statementLifecycle(),
                metrics.fragmentObserver(fragment.fragmentId()),
                deadline
            )
        );
        if (cursor == null || !context.queryId().equals(cursor.queryId())) {
            closeQuietly(cursor);
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "adapter returned an invalid fragment cursor"
            );
        }
        openFragmentCursors.add(cursor);
        try {
            ensureExecutionAllowed();
            return cursor;
        } catch (RuntimeException | Error exception) {
            openFragmentCursors.remove(cursor);
            closeQuietly(cursor);
            throw exception;
        }
    }

    private void recordFragmentIntermediate(
        CompiledFederationFragment compiled,
        Object[] row
    ) {
        FederationQueryMetricsTracker.IntermediateUsage usage =
            metrics.recordIntermediate(compiled.plan().fragmentId(), row);
        if (usage.rows() > policy.maximumIntermediateRows()
            || usage.bytes() > policy.maximumIntermediateBytes()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_RESOURCE_LIMIT_EXCEEDED,
                "federation intermediate result limit was exceeded"
            );
        }
    }

    private void cancelPreparation() {
        synchronized (preparationLock) {
            preparationFailed = true;
            pendingPreparations.clear();
        }
        for (Future<?> task : List.copyOf(fragmentTasks)) {
            task.cancel(true);
        }
        fragmentTasks.clear();
        for (FederationResultCursor cursor : List.copyOf(openFragmentCursors)) {
            openFragmentCursors.remove(cursor);
            try {
                cursor.close();
            } catch (RuntimeException ignored) {
                // 保留触发取消的原始分片异常；Statement 取消通道继续收口剩余资源。
            }
        }
        for (CompletableFuture<PreparedFragment> preparation : fragmentPreparations.values()) {
            preparation.cancel(true);
        }
        fragmentPreparations.clear();
        registration.terminateStatementsForCleanup();
    }

    private void acquireFragmentSlot() {
        while (true) {
            ensureExecutionAllowed();
            long remaining = deadline.remainingNanos();
            if (remaining <= 0L) {
                ensureExecutionAllowed();
            }
            try {
                if (fragmentSlots.tryAcquire(
                    Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(50)),
                    TimeUnit.NANOSECONDS
                )) {
                    return;
                }
            } catch (InterruptedException exception) {
                Thread.currentThread().interrupt();
                throw new FederationSqlException(
                    FederationSqlErrorCode.EXECUTION_FAILED,
                    "waiting for a federation fragment slot was interrupted",
                    exception
                );
            }
        }
    }

    private SqlExecutionOptions fragmentOptions(SqlExecutionOptions requested) {
        // 最终 maxRows 不能截断 Join/Aggregate 输入，否则会改变查询结果。
        return new SqlExecutionOptions(
            requested.fetchSize(),
            0,
            effectiveJdbcTimeoutSeconds(requested.queryTimeoutSeconds()),
            requested.readOnly()
        );
    }

    private int effectiveJdbcTimeoutSeconds(int requestedSeconds) {
        return deadline.boundedQueryTimeoutSeconds(requestedSeconds);
    }

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

    /**
     * 关闭所有仍活动的物理分片游标。
     */
    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        boolean terminationRequired = !fragmentTasks.isEmpty()
            || !openFragmentCursors.isEmpty();
        synchronized (preparationLock) {
            terminationRequired = terminationRequired || runningPreparations > 0;
            preparationFailed = true;
            pendingPreparations.clear();
        }
        for (Future<?> task : List.copyOf(fragmentTasks)) {
            task.cancel(true);
        }
        fragmentTasks.clear();
        for (CompletableFuture<PreparedFragment> preparation : fragmentPreparations.values()) {
            preparation.cancel(true);
        }
        fragmentPreparations.clear();
        RuntimeException failure = null;
        for (FederationResultCursor cursor : List.copyOf(openFragmentCursors)) {
            try {
                cursor.close();
            } catch (RuntimeException exception) {
                if (failure == null) {
                    failure = exception;
                } else {
                    failure.addSuppressed(exception);
                }
            } finally {
                openFragmentCursors.remove(cursor);
            }
        }
        if (terminationRequired) {
            registration.terminateStatementsForCleanup();
        }
        if (failure != null) {
            throw failure;
        }
    }

    private final class FragmentEnumerator implements Enumerator<Object[]> {

        private final CompiledFederationFragment compiled;
        private final FederationResultCursor cursor;
        private final boolean fragmentSlotHeld;
        private final AtomicBoolean released = new AtomicBoolean();
        private Object[] current;

        private FragmentEnumerator(
            CompiledFederationFragment compiled,
            FederationResultCursor cursor,
            boolean fragmentSlotHeld
        ) {
            this.compiled = compiled;
            this.cursor = cursor;
            this.fragmentSlotHeld = fragmentSlotHeld;
        }

        /** {@inheritDoc} */
        @Override
        public Object[] current() {
            return current;
        }

        /** {@inheritDoc} */
        @Override
        public boolean moveNext() {
            ensureExecutionAllowed();
            try {
                if (!cursor.next()) {
                    release(true);
                    return false;
                }
                Object[] row = normalizeRow(cursor, compiled.rowType());
                recordFragmentIntermediate(compiled, row);
                current = row;
                return true;
            } catch (RuntimeException exception) {
                release(false);
                throw exception;
            }
        }

        /** {@inheritDoc} */
        @Override
        public void reset() {
            throw new UnsupportedOperationException("fragment cursor cannot be reset");
        }

        /** {@inheritDoc} */
        @Override
        public void close() {
            release(false);
        }

        private void release(boolean exhausted) {
            if (!released.compareAndSet(false, true)) {
                return;
            }
            openFragmentCursors.remove(cursor);
            try {
                cursor.close();
            } finally {
                if (fragmentSlotHeld) {
                    fragmentSlots.release();
                }
                if (exhausted) {
                    metrics.finishFragment(compiled.plan().fragmentId());
                }
            }
        }
    }

    private static final class PreparedFragment {

        private final List<Object[]> rows;

        private PreparedFragment(List<Object[]> rows) {
            this.rows = rows;
        }

        private Enumerator<Object[]> enumerator() {
            return new Enumerator<>() {

                private int index = -1;
                private Object[] current;

                @Override
                public Object[] current() {
                    return current;
                }

                @Override
                public boolean moveNext() {
                    int next = index + 1;
                    if (next >= rows.size()) {
                        current = null;
                        return false;
                    }
                    index = next;
                    current = rows.set(index, null);
                    return true;
                }

                @Override
                public void reset() {
                    throw new UnsupportedOperationException(
                        "fragment cursor cannot be reset"
                    );
                }

                @Override
                public void close() {
                    rows.clear();
                    current = null;
                }
            };
        }
    }

    private record PreparationTask(
        CompiledFederationFragment compiled,
        CompletableFuture<PreparedFragment> preparation
    ) {
    }

    private void releaseFragment(
        CompiledFederationFragment compiled,
        FederationResultCursor cursor,
        boolean exhausted,
        boolean fragmentSlotHeld
    ) {
        openFragmentCursors.remove(cursor);
        try {
            cursor.close();
        } finally {
            if (fragmentSlotHeld) {
                fragmentSlots.release();
            }
            if (exhausted) {
                metrics.finishFragment(compiled.plan().fragmentId());
            }
        }
    }

    private static Object[] normalizeRow(
        FederationResultCursor cursor,
        RelDataType rowType
    ) {
        List<RelDataTypeField> fields = rowType.getFieldList();
        Object[] values = new Object[fields.size()];
        for (int index = 0; index < fields.size(); index++) {
            values[index] = normalizeValue(
                cursor.getObject(index + 1),
                fields.get(index).getType().getSqlTypeName()
            );
        }
        return values;
    }

    private static Object normalizeValue(Object value, SqlTypeName typeName) {
        if (value == null) {
            return null;
        }
        return switch (typeName) {
            case BOOLEAN -> (Boolean) value;
            case TINYINT -> ((Number) value).byteValue();
            case SMALLINT -> ((Number) value).shortValue();
            case INTEGER -> ((Number) value).intValue();
            case BIGINT -> ((Number) value).longValue();
            case REAL -> ((Number) value).floatValue();
            case FLOAT, DOUBLE -> ((Number) value).doubleValue();
            case DECIMAL -> value instanceof BigDecimal
                ? value
                : new BigDecimal(value.toString());
            case DATE -> normalizeDate(value);
            case TIME, TIME_WITH_LOCAL_TIME_ZONE, TIME_TZ -> normalizeTime(value);
            case TIMESTAMP, TIMESTAMP_WITH_LOCAL_TIME_ZONE, TIMESTAMP_TZ ->
                normalizeTimestamp(value);
            case BINARY, VARBINARY -> value instanceof ByteString
                ? value
                : new ByteString((byte[]) value);
            case CHAR, VARCHAR -> value.toString();
            case UUID -> value instanceof java.util.UUID
                ? value
                : java.util.UUID.fromString(value.toString());
            case ANY, OTHER -> normalizeKnownUnknownScalar(value);
            case ARRAY, MULTISET, MAP, ROW -> throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED,
                "complex JDBC values are not supported by local federation operators"
            );
            default -> throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED,
                "JDBC type cannot be safely normalized for local federation: " + typeName
            );
        };
    }

    /**
     * 接受驱动元数据无法精确声明、但仍可稳定序列化和计量的标准 JDBC 标量。
     *
     * @param value JDBC 返回值
     * @return 可供本地 Enumerable 透传的值
     * @throws FederationSqlException 值不在安全白名单内
     */
    private static Object normalizeKnownUnknownScalar(Object value) {
        if (value instanceof java.time.OffsetTime offsetTime) {
            return offsetTime.withOffsetSameInstant(ZoneOffset.UTC);
        }
        if (value instanceof java.time.OffsetDateTime offsetDateTime) {
            return offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC);
        }
        if (value instanceof java.time.LocalDate
            || value instanceof java.time.LocalTime
            || value instanceof java.time.LocalDateTime
            || value instanceof java.time.Instant
            || value instanceof java.sql.Date
            || value instanceof java.sql.Time
            || value instanceof java.sql.Timestamp
            || value instanceof java.util.UUID) {
            return value;
        }
        throw new FederationSqlException(
            FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED,
            "JDBC value cannot be safely normalized for local federation: "
                + value.getClass().getName()
        );
    }

    private static Integer normalizeDate(Object value) {
        LocalDate date = value instanceof Date sqlDate
            ? sqlDate.toLocalDate()
            : (LocalDate) value;
        return Math.toIntExact(date.toEpochDay());
    }

    private static Integer normalizeTime(Object value) {
        LocalTime time;
        if (value instanceof Time sqlTime) {
            time = sqlTime.toLocalTime();
        } else if (value instanceof java.time.OffsetTime offsetTime) {
            time = offsetTime.withOffsetSameInstant(ZoneOffset.UTC).toLocalTime();
        } else {
            time = (LocalTime) value;
        }
        requireMillisecondPrecision(time.getNano(), "TIME");
        return Math.toIntExact(time.toNanoOfDay() / 1_000_000L);
    }

    private static Long normalizeTimestamp(Object value) {
        if (value instanceof Timestamp timestamp) {
            requireMillisecondPrecision(timestamp.getNanos(), "TIMESTAMP");
            return timestamp.getTime();
        }
        if (value instanceof Instant instant) {
            requireMillisecondPrecision(instant.getNano(), "TIMESTAMP");
            return instant.toEpochMilli();
        }
        if (value instanceof OffsetDateTime offset) {
            requireMillisecondPrecision(offset.getNano(), "TIMESTAMP");
            return offset.toInstant().toEpochMilli();
        }
        LocalDateTime local = (LocalDateTime) value;
        requireMillisecondPrecision(local.getNano(), "TIMESTAMP");
        return local.toInstant(ZoneOffset.UTC).toEpochMilli();
    }

    /**
     * 拒绝 Calcite 毫秒内部表示无法保真的时间值。
     *
     * @param nanos 秒内纳秒值
     * @param typeName SQL 时间类型名
     * @throws FederationSqlException 时间值包含亚毫秒精度
     */
    private static void requireMillisecondPrecision(int nanos, String typeName) {
        if (nanos % 1_000_000 != 0) {
            throw new FederationSqlException(
                FederationSqlErrorCode.FEDERATION_OPERATOR_UNSUPPORTED,
                typeName + " values with precision above milliseconds are not supported "
                    + "by local federation"
            );
        }
    }

    private static Object normalizeParameter(SqlParameter parameter) {
        Object value = parameter.value();
        if (value == null) {
            return null;
        }
        return switch (parameter.jdbcType()) {
            case Types.DATE -> normalizeDate(value);
            case Types.TIME, Types.TIME_WITH_TIMEZONE -> normalizeTime(value);
            case Types.TIMESTAMP, Types.TIMESTAMP_WITH_TIMEZONE -> normalizeTimestamp(value);
            case Types.BINARY, Types.VARBINARY, Types.LONGVARBINARY ->
                value instanceof ByteString ? value : new ByteString((byte[]) value);
            default -> value;
        };
    }

    private static void closeQuietly(FederationResultCursor cursor) {
        if (cursor != null) {
            cursor.close();
        }
    }
}
